agent/codex
1
0
Fork 0
mirror of https://github.com/pchuan98/codex.git synced 2026-07-01 00:31:56 +08:00
Code Issues Packages Projects Releases Wiki Activity

[codex-analytics] add compaction analytics event (#17155)

Browse Source 
- event for compaction analytics
- introduces thread-connection and thread metadata caches for data
denormalization, expected to be useful for denormalization onto core
emitted events in general
- threads analytics event client into core (mirrors approved
implementation in #16640)
- denormalizes key thread metadata: thread_source, subagent_source,
parent_thread_id, as well as app-server client and runtime metadata)
- compaction strategy defaults to memento, forward compatible with
expected prefill_compaction strategy

1. Manual standalone compact, local
`INFO | 2026-04-09 17:35:50 | codex_backend.routers.analytics_events |
analytics_events.track_analytics_events:526 | Tracked
codex_compaction_event event params={'thread_id':
'019d74d0-5cfb-70c0-bef9-165c3bf9b2df', 'turn_id':
'019d74d0-d7f6-7c81-acc6-aae2030243d6', 'product_surface': 'codex',
'app_server_client': {'product_client_id': 'CODEX_CLI', 'client_name':
'codex-tui', 'client_version': '0.0.0', 'rpc_transport': 'in_process',
'experimental_api_enabled': True}, 'runtime': {'codex_rs_version':
'0.0.0', 'runtime_os': 'macos', 'runtime_os_version': '26.4.0',
'runtime_arch': 'aarch64'}, 'trigger': 'manual', 'reason':
'user_requested', 'implementation': 'responses', 'phase':
'standalone_turn', 'strategy': 'memento', 'status': 'completed',
'active_context_tokens_before': 20170, 'active_context_tokens_after':
4830, 'started_at': 1775781337, 'completed_at': 1775781350,
'thread_source': 'user', 'subagent_source': None, 'parent_thread_id':
None, 'error': None, 'duration_ms': 13524} | `

2. Auto pre-turn compact, local
`INFO | 2026-04-09 17:37:30 | codex_backend.routers.analytics_events |
analytics_events.track_analytics_events:526 | Tracked
codex_compaction_event event params={'thread_id':
'019d74d2-45ef-71d1-9c93-23cc0c13d988', 'turn_id':
'019d74d2-7b42-7372-9f0e-c0da3f352328', 'product_surface': 'codex',
'app_server_client': {'product_client_id': 'CODEX_CLI', 'client_name':
'codex-tui', 'client_version': '0.0.0', 'rpc_transport': 'in_process',
'experimental_api_enabled': True}, 'runtime': {'codex_rs_version':
'0.0.0', 'runtime_os': 'macos', 'runtime_os_version': '26.4.0',
'runtime_arch': 'aarch64'}, 'trigger': 'auto', 'reason':
'context_limit', 'implementation': 'responses', 'phase': 'pre_turn',
'strategy': 'memento', 'status': 'completed',
'active_context_tokens_before': 20063, 'active_context_tokens_after':
4822, 'started_at': 1775781444, 'completed_at': 1775781449,
'thread_source': 'user', 'subagent_source': None, 'parent_thread_id':
None, 'error': None, 'duration_ms': 5497} | `

3. Auto mid-turn compact, local
`INFO | 2026-04-09 17:38:28 | codex_backend.routers.analytics_events |
analytics_events.track_analytics_events:526 | Tracked
codex_compaction_event event params={'thread_id':
'019d74d3-212f-7a20-8c0a-4816a978675e', 'turn_id':
'019d74d3-3ee1-7462-89f6-2ffbeefcd5e3', 'product_surface': 'codex',
'app_server_client': {'product_client_id': 'CODEX_CLI', 'client_name':
'codex-tui', 'client_version': '0.0.0', 'rpc_transport': 'in_process',
'experimental_api_enabled': True}, 'runtime': {'codex_rs_version':
'0.0.0', 'runtime_os': 'macos', 'runtime_os_version': '26.4.0',
'runtime_arch': 'aarch64'}, 'trigger': 'auto', 'reason':
'context_limit', 'implementation': 'responses', 'phase': 'mid_turn',
'strategy': 'memento', 'status': 'completed',
'active_context_tokens_before': 20325, 'active_context_tokens_after':
14641, 'started_at': 1775781500, 'completed_at': 1775781508,
'thread_source': 'user', 'subagent_source': None, 'parent_thread_id':
None, 'error': None, 'duration_ms': 7507} | `

4. Remote /responses/compact, manual standalone
`INFO | 2026-04-09 17:40:20 | codex_backend.routers.analytics_events |
analytics_events.track_analytics_events:526 | Tracked
codex_compaction_event event params={'thread_id':
'019d74d4-7a11-78a1-89f7-0535a1149416', 'turn_id':
'019d74d4-e087-7183-9c20-b1e40b7578c0', 'product_surface': 'codex',
'app_server_client': {'product_client_id': 'CODEX_CLI', 'client_name':
'codex-tui', 'client_version': '0.0.0', 'rpc_transport': 'in_process',
'experimental_api_enabled': True}, 'runtime': {'codex_rs_version':
'0.0.0', 'runtime_os': 'macos', 'runtime_os_version': '26.4.0',
'runtime_arch': 'aarch64'}, 'trigger': 'manual', 'reason':
'user_requested', 'implementation': 'responses_compact', 'phase':
'standalone_turn', 'strategy': 'memento', 'status': 'completed',
'active_context_tokens_before': 23461, 'active_context_tokens_after':
6171, 'started_at': 1775781601, 'completed_at': 1775781620,
'thread_source': 'user', 'subagent_source': None, 'parent_thread_id':
None, 'error': None, 'duration_ms': 18971} | `
This commit is contained in:
rhan-oai
2026-04-10 13:03:54 -07:00
committed by GitHub
Unverified
parent 029fc63d13
commit 5779be314a
19 changed files with 717 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/analytics/src/analytics_client_tests.rs
+236 -2
View File
@@ -3,6 +3,7 @@ use crate::events::AppServerRpcTransport;
use crate::events::CodexAppMentionedEventRequest;
use crate::events::CodexAppServerClientMetadata;
use crate::events::CodexAppUsedEventRequest;
use crate::events::CodexCompactionEventRequest;
use crate::events::CodexPluginEventRequest;
use crate::events::CodexPluginUsedEventRequest;
use crate::events::CodexRuntimeMetadata;
@@ -18,6 +19,13 @@ use crate::facts::AnalyticsFact;
use crate::facts::AppInvocation;
use crate::facts::AppMentionedInput;
use crate::facts::AppUsedInput;
use crate::facts::CodexCompactionEvent;
use crate::facts::CompactionImplementation;
use crate::facts::CompactionPhase;
use crate::facts::CompactionReason;
use crate::facts::CompactionStatus;
use crate::facts::CompactionStrategy;
use crate::facts::CompactionTrigger;
use crate::facts::CustomAnalyticsFact;
use crate::facts::InvocationType;
use crate::facts::PluginState;
@@ -59,6 +67,14 @@ use std::sync::Mutex;
use tokio::sync::mpsc;
fn sample_thread(thread_id: &str, ephemeral: bool) -> Thread {
sample_thread_with_source(thread_id, ephemeral, AppServerSessionSource::Exec)
}
fn sample_thread_with_source(
thread_id: &str,
ephemeral: bool,
source: AppServerSessionSource,
) -> Thread {
Thread {
id: thread_id.to_string(),
forked_from_id: None,
@@ -71,7 +87,7 @@ fn sample_thread(thread_id: &str, ephemeral: bool) -> Thread {
path: None,
cwd: PathBuf::from("/tmp"),
cli_version: "0.0.0".to_string(),
source: AppServerSessionSource::Exec,
source,
agent_nickname: None,
agent_role: None,
git_info: None,
@@ -97,11 +113,44 @@ fn sample_thread_start_response(thread_id: &str, ephemeral: bool, model: &str) -
}
}
fn sample_app_server_client_metadata() -> CodexAppServerClientMetadata {
CodexAppServerClientMetadata {
product_client_id: DEFAULT_ORIGINATOR.to_string(),
client_name: Some("codex-tui".to_string()),
client_version: Some("1.0.0".to_string()),
rpc_transport: AppServerRpcTransport::Stdio,
experimental_api_enabled: Some(true),
}
}
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(),
}
}
fn sample_thread_resume_response(thread_id: &str, ephemeral: bool, model: &str) -> ClientResponse {
sample_thread_resume_response_with_source(
thread_id,
ephemeral,
model,
AppServerSessionSource::Exec,
)
}
fn sample_thread_resume_response_with_source(
thread_id: &str,
ephemeral: bool,
model: &str,
source: AppServerSessionSource,
) -> ClientResponse {
ClientResponse::ThreadResume {
request_id: RequestId::Integer(2),
response: ThreadResumeResponse {
thread: sample_thread(thread_id, ephemeral),
thread: sample_thread_with_source(thread_id, ephemeral, source),
model: model.to_string(),
model_provider: "openai".to_string(),
service_tier: None,
@@ -254,6 +303,77 @@ fn app_used_event_serializes_expected_shape() {
);
}
#[test]
fn compaction_event_serializes_expected_shape() {
let event = TrackEventRequest::Compaction(Box::new(CodexCompactionEventRequest {
event_type: "codex_compaction_event",
event_params: crate::events::codex_compaction_event_params(
CodexCompactionEvent {
thread_id: "thread-1".to_string(),
turn_id: "turn-1".to_string(),
trigger: CompactionTrigger::Auto,
reason: CompactionReason::ContextLimit,
implementation: CompactionImplementation::ResponsesCompact,
phase: CompactionPhase::MidTurn,
strategy: CompactionStrategy::Memento,
status: CompactionStatus::Completed,
error: None,
active_context_tokens_before: 120_000,
active_context_tokens_after: 18_000,
started_at: 100,
completed_at: 106,
duration_ms: Some(6543),
},
sample_app_server_client_metadata(),
sample_runtime_metadata(),
Some("user"),
/*subagent_source*/ None,
/*parent_thread_id*/ None,
),
}));
let payload = serde_json::to_value(&event).expect("serialize compaction event");
assert_eq!(
payload,
json!({
"event_type": "codex_compaction_event",
"event_params": {
"thread_id": "thread-1",
"turn_id": "turn-1",
"app_server_client": {
"product_client_id": DEFAULT_ORIGINATOR,
"client_name": "codex-tui",
"client_version": "1.0.0",
"rpc_transport": "stdio",
"experimental_api_enabled": true
},
"runtime": {
"codex_rs_version": "0.1.0",
"runtime_os": "macos",
"runtime_os_version": "15.3.1",
"runtime_arch": "aarch64"
},
"thread_source": "user",
"subagent_source": null,
"parent_thread_id": null,
"trigger": "auto",
"reason": "context_limit",
"implementation": "responses_compact",
"phase": "mid_turn",
"strategy": "memento",
"status": "completed",
"error": null,
"active_context_tokens_before": 120000,
"active_context_tokens_after": 18000,
"started_at": 100,
"completed_at": 106,
"duration_ms": 6543
}
})
);
}
#[test]
fn app_used_dedupe_is_keyed_by_turn_and_connector() {
let (sender, _receiver) = mpsc::channel(1);
@@ -449,6 +569,120 @@ async fn initialize_caches_client_and_thread_lifecycle_publishes_once_initialize
assert_eq!(payload[0]["event_params"]["parent_thread_id"], json!(null));
}
#[tokio::test]
async fn compaction_event_ingests_custom_fact() {
let mut reducer = AnalyticsReducer::default();
let mut events = Vec::new();
let parent_thread_id =
codex_protocol::ThreadId::from_string("22222222-2222-2222-2222-222222222222")
.expect("valid parent thread id");
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: Some(InitializeCapabilities {
experimental_api: false,
opt_out_notification_methods: None,
}),
},
product_client_id: DEFAULT_ORIGINATOR.to_string(),
runtime: sample_runtime_metadata(),
rpc_transport: AppServerRpcTransport::Websocket,
},
&mut events,
)
.await;
reducer
.ingest(
AnalyticsFact::Response {
connection_id: 7,
response: Box::new(sample_thread_resume_response_with_source(
"thread-1",
/*ephemeral*/ false,
"gpt-5",
AppServerSessionSource::SubAgent(SubAgentSource::ThreadSpawn {
parent_thread_id,
depth: 1,
agent_path: None,
agent_nickname: None,
agent_role: None,
}),
)),
},
&mut events,
)
.await;
events.clear();
reducer
.ingest(
AnalyticsFact::Custom(CustomAnalyticsFact::Compaction(Box::new(
CodexCompactionEvent {
thread_id: "thread-1".to_string(),
turn_id: "turn-compact".to_string(),
trigger: CompactionTrigger::Manual,
reason: CompactionReason::UserRequested,
implementation: CompactionImplementation::Responses,
phase: CompactionPhase::StandaloneTurn,
strategy: CompactionStrategy::Memento,
status: CompactionStatus::Failed,
error: Some("context limit exceeded".to_string()),
active_context_tokens_before: 131_000,
active_context_tokens_after: 131_000,
started_at: 100,
completed_at: 101,
duration_ms: Some(1200),
},
))),
&mut events,
)
.await;
let payload = serde_json::to_value(&events).expect("serialize events");
assert_eq!(payload.as_array().expect("events array").len(), 1);
assert_eq!(payload[0]["event_type"], "codex_compaction_event");
assert_eq!(payload[0]["event_params"]["thread_id"], "thread-1");
assert_eq!(payload[0]["event_params"]["turn_id"], "turn-compact");
assert_eq!(
payload[0]["event_params"]["app_server_client"]["product_client_id"],
DEFAULT_ORIGINATOR
);
assert_eq!(
payload[0]["event_params"]["app_server_client"]["client_name"],
"codex-tui"
);
assert_eq!(
payload[0]["event_params"]["app_server_client"]["rpc_transport"],
"websocket"
);
assert_eq!(
payload[0]["event_params"]["runtime"]["codex_rs_version"],
"0.1.0"
);
assert_eq!(payload[0]["event_params"]["thread_source"], "subagent");
assert_eq!(
payload[0]["event_params"]["subagent_source"],
"thread_spawn"
);
assert_eq!(
payload[0]["event_params"]["parent_thread_id"],
"22222222-2222-2222-2222-222222222222"
);
assert_eq!(payload[0]["event_params"]["trigger"], "manual");
assert_eq!(payload[0]["event_params"]["reason"], "user_requested");
assert_eq!(payload[0]["event_params"]["implementation"], "responses");
assert_eq!(payload[0]["event_params"]["phase"], "standalone_turn");
assert_eq!(payload[0]["event_params"]["strategy"], "memento");
assert_eq!(payload[0]["event_params"]["status"], "failed");
}
#[test]
fn subagent_thread_started_review_serializes_expected_shape() {
let event = TrackEventRequest::ThreadInitialized(subagent_thread_started_event_request(
codex-rs/analytics/src/client.rs
+6
View File
@@ -178,6 +178,12 @@ impl AnalyticsEventsClient {
)));
}
pub fn track_compaction(&self, event: crate::facts::CodexCompactionEvent) {
self.record_fact(AnalyticsFact::Custom(CustomAnalyticsFact::Compaction(
Box::new(event),
)));
}
pub fn track_plugin_installed(&self, plugin: PluginTelemetryMetadata) {
self.record_fact(AnalyticsFact::Custom(
CustomAnalyticsFact::PluginStateChanged(PluginStateChangedInput {
codex-rs/analytics/src/events.rs
+64 -2
View File
@@ -1,4 +1,5 @@
use crate::facts::AppInvocation;
use crate::facts::CodexCompactionEvent;
use crate::facts::InvocationType;
use crate::facts::PluginState;
use crate::facts::SubAgentThreadStartedInput;
@@ -37,6 +38,7 @@ pub(crate) enum TrackEventRequest {
ThreadInitialized(ThreadInitializedEvent),
AppMentioned(CodexAppMentionedEventRequest),
AppUsed(CodexAppUsedEventRequest),
Compaction(Box<CodexCompactionEventRequest>),
PluginUsed(CodexPluginUsedEventRequest),
PluginInstalled(CodexPluginEventRequest),
PluginUninstalled(CodexPluginEventRequest),
@@ -122,6 +124,35 @@ pub(crate) struct CodexAppUsedEventRequest {
pub(crate) event_params: CodexAppMetadata,
}
#[derive(Serialize)]
pub(crate) struct CodexCompactionEventParams {
pub(crate) thread_id: String,
pub(crate) turn_id: String,
pub(crate) app_server_client: CodexAppServerClientMetadata,
pub(crate) runtime: CodexRuntimeMetadata,
pub(crate) thread_source: Option<&'static str>,
pub(crate) subagent_source: Option<String>,
pub(crate) parent_thread_id: Option<String>,
pub(crate) trigger: crate::facts::CompactionTrigger,
pub(crate) reason: crate::facts::CompactionReason,
pub(crate) implementation: crate::facts::CompactionImplementation,
pub(crate) phase: crate::facts::CompactionPhase,
pub(crate) strategy: crate::facts::CompactionStrategy,
pub(crate) status: crate::facts::CompactionStatus,
pub(crate) error: Option<String>,
pub(crate) active_context_tokens_before: i64,
pub(crate) active_context_tokens_after: i64,
pub(crate) started_at: u64,
pub(crate) completed_at: u64,
pub(crate) duration_ms: Option<u64>,
}
#[derive(Serialize)]
pub(crate) struct CodexCompactionEventRequest {
pub(crate) event_type: &'static str,
pub(crate) event_params: CodexCompactionEventParams,
}
#[derive(Serialize)]
pub(crate) struct CodexPluginMetadata {
pub(crate) plugin_id: Option<String>,
@@ -201,6 +232,37 @@ pub(crate) fn codex_plugin_metadata(plugin: PluginTelemetryMetadata) -> CodexPlu
}
}
pub(crate) fn codex_compaction_event_params(
input: CodexCompactionEvent,
app_server_client: CodexAppServerClientMetadata,
runtime: CodexRuntimeMetadata,
thread_source: Option<&'static str>,
subagent_source: Option<String>,
parent_thread_id: Option<String>,
) -> CodexCompactionEventParams {
CodexCompactionEventParams {
thread_id: input.thread_id,
turn_id: input.turn_id,
app_server_client,
runtime,
thread_source,
subagent_source,
parent_thread_id,
trigger: input.trigger,
reason: input.reason,
implementation: input.implementation,
phase: input.phase,
strategy: input.strategy,
status: input.status,
error: input.error,
active_context_tokens_before: input.active_context_tokens_before,
active_context_tokens_after: input.active_context_tokens_after,
started_at: input.started_at,
completed_at: input.completed_at,
duration_ms: input.duration_ms,
}
}
pub(crate) fn codex_plugin_used_metadata(
tracking: &TrackEventsContext,
plugin: PluginTelemetryMetadata,
@@ -260,7 +322,7 @@ pub(crate) fn subagent_thread_started_event_request(
}
}
fn subagent_source_name(subagent_source: &SubAgentSource) -> String {
pub(crate) fn subagent_source_name(subagent_source: &SubAgentSource) -> String {
match subagent_source {
SubAgentSource::Review => "review".to_string(),
SubAgentSource::Compact => "compact".to_string(),
@@ -270,7 +332,7 @@ fn subagent_source_name(subagent_source: &SubAgentSource) -> String {
}
}
fn subagent_parent_thread_id(subagent_source: &SubAgentSource) -> Option<String> {
pub(crate) fn subagent_parent_thread_id(subagent_source: &SubAgentSource) -> Option<String> {
match subagent_source {
SubAgentSource::ThreadSpawn {
parent_thread_id, ..
codex-rs/analytics/src/facts.rs
+64
View File
@@ -64,6 +64,69 @@ pub struct SubAgentThreadStartedInput {
pub created_at: u64,
}
#[derive(Clone, Copy, Debug, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum CompactionTrigger {
Manual,
Auto,
}
#[derive(Clone, Copy, Debug, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum CompactionReason {
UserRequested,
ContextLimit,
ModelDownshift,
}
#[derive(Clone, Copy, Debug, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum CompactionImplementation {
Responses,
ResponsesCompact,
}
#[derive(Clone, Copy, Debug, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum CompactionPhase {
StandaloneTurn,
PreTurn,
MidTurn,
}
#[derive(Clone, Copy, Debug, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum CompactionStrategy {
Memento,
PrefixCompaction,
}
#[derive(Clone, Copy, Debug, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum CompactionStatus {
Completed,
Failed,
Interrupted,
}
#[derive(Clone)]
pub struct CodexCompactionEvent {
pub thread_id: String,
pub turn_id: String,
pub trigger: CompactionTrigger,
pub reason: CompactionReason,
pub implementation: CompactionImplementation,
pub phase: CompactionPhase,
pub strategy: CompactionStrategy,
pub status: CompactionStatus,
pub error: Option<String>,
pub active_context_tokens_before: i64,
pub active_context_tokens_after: i64,
pub started_at: u64,
pub completed_at: u64,
pub duration_ms: Option<u64>,
}
#[allow(dead_code)]
pub(crate) enum AnalyticsFact {
Initialize {
@@ -90,6 +153,7 @@ pub(crate) enum AnalyticsFact {
pub(crate) enum CustomAnalyticsFact {
SubAgentThreadStarted(SubAgentThreadStartedInput),
Compaction(Box<CodexCompactionEvent>),
SkillInvoked(SkillInvokedInput),
AppMentioned(AppMentionedInput),
AppUsed(AppUsedInput),
codex-rs/analytics/src/lib.rs
+7
View File
@@ -6,6 +6,13 @@ mod reducer;
pub use client::AnalyticsEventsClient;
pub use events::AppServerRpcTransport;
pub use facts::AppInvocation;
pub use facts::CodexCompactionEvent;
pub use facts::CompactionImplementation;
pub use facts::CompactionPhase;
pub use facts::CompactionReason;
pub use facts::CompactionStatus;
pub use facts::CompactionStrategy;
pub use facts::CompactionTrigger;
pub use facts::InvocationType;
pub use facts::SkillInvocation;
pub use facts::SubAgentThreadStartedInput;
codex-rs/analytics/src/reducer.rs
+90 -4
View File
@@ -2,6 +2,7 @@ use crate::events::AppServerRpcTransport;
use crate::events::CodexAppMentionedEventRequest;
use crate::events::CodexAppServerClientMetadata;
use crate::events::CodexAppUsedEventRequest;
use crate::events::CodexCompactionEventRequest;
use crate::events::CodexPluginEventRequest;
use crate::events::CodexPluginUsedEventRequest;
use crate::events::CodexRuntimeMetadata;
@@ -12,14 +13,18 @@ use crate::events::ThreadInitializedEvent;
use crate::events::ThreadInitializedEventParams;
use crate::events::TrackEventRequest;
use crate::events::codex_app_metadata;
use crate::events::codex_compaction_event_params;
use crate::events::codex_plugin_metadata;
use crate::events::codex_plugin_used_metadata;
use crate::events::plugin_state_event_type;
use crate::events::subagent_parent_thread_id;
use crate::events::subagent_source_name;
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::CodexCompactionEvent;
use crate::facts::CustomAnalyticsFact;
use crate::facts::PluginState;
use crate::facts::PluginStateChangedInput;
@@ -40,6 +45,8 @@ use std::path::Path;
#[derive(Default)]
pub(crate) struct AnalyticsReducer {
connections: HashMap<u64, ConnectionState>,
thread_connections: HashMap<String, u64>,
thread_metadata: HashMap<String, ThreadMetadataState>,
}
struct ConnectionState {
@@ -47,6 +54,35 @@ struct ConnectionState {
runtime: CodexRuntimeMetadata,
}
#[derive(Clone)]
struct ThreadMetadataState {
thread_source: Option<&'static str>,
subagent_source: Option<String>,
parent_thread_id: Option<String>,
}
impl ThreadMetadataState {
fn from_session_source(session_source: &SessionSource) -> Self {
let (subagent_source, parent_thread_id) = match session_source {
SessionSource::SubAgent(subagent_source) => (
Some(subagent_source_name(subagent_source)),
subagent_parent_thread_id(subagent_source),
),
SessionSource::Cli
| SessionSource::VSCode
| SessionSource::Exec
| SessionSource::Mcp
| SessionSource::Custom(_)
| SessionSource::Unknown => (None, None),
};
Self {
thread_source: thread_source_name(session_source),
subagent_source,
parent_thread_id,
}
}
}
impl AnalyticsReducer {
pub(crate) async fn ingest(&mut self, input: AnalyticsFact, out: &mut Vec<TrackEventRequest>) {
match input {
@@ -81,6 +117,9 @@ impl AnalyticsReducer {
CustomAnalyticsFact::SubAgentThreadStarted(input) => {
self.ingest_subagent_thread_started(input, out);
}
CustomAnalyticsFact::Compaction(input) => {
self.ingest_compaction(*input, out);
}
CustomAnalyticsFact::SkillInvoked(input) => {
self.ingest_skill_invoked(input, out).await;
}
@@ -254,27 +293,74 @@ impl AnalyticsReducer {
_ => return,
};
let thread_source: SessionSource = thread.source.into();
let thread_id = thread.id;
let Some(connection_state) = self.connections.get(&connection_id) else {
return;
};
let thread_metadata = ThreadMetadataState::from_session_source(&thread_source);
self.thread_connections
.insert(thread_id.clone(), connection_id);
self.thread_metadata
.insert(thread_id.clone(), thread_metadata.clone());
out.push(TrackEventRequest::ThreadInitialized(
ThreadInitializedEvent {
event_type: "codex_thread_initialized",
event_params: ThreadInitializedEventParams {
thread_id: thread.id,
thread_id,
app_server_client: connection_state.app_server_client.clone(),
runtime: connection_state.runtime.clone(),
model,
ephemeral: thread.ephemeral,
thread_source: thread_source_name(&thread_source),
thread_source: thread_metadata.thread_source,
initialization_mode,
subagent_source: None,
parent_thread_id: None,
subagent_source: thread_metadata.subagent_source,
parent_thread_id: thread_metadata.parent_thread_id,
created_at: u64::try_from(thread.created_at).unwrap_or_default(),
},
},
));
}
fn ingest_compaction(&mut self, input: CodexCompactionEvent, out: &mut Vec<TrackEventRequest>) {
let Some(connection_id) = self.thread_connections.get(&input.thread_id) else {
tracing::warn!(
thread_id = %input.thread_id,
turn_id = %input.turn_id,
"dropping compaction analytics event: missing thread connection metadata"
);
return;
};
let Some(connection_state) = self.connections.get(connection_id) else {
tracing::warn!(
thread_id = %input.thread_id,
turn_id = %input.turn_id,
connection_id,
"dropping compaction analytics event: missing connection metadata"
);
return;
};
let Some(thread_metadata) = self.thread_metadata.get(&input.thread_id) else {
tracing::warn!(
thread_id = %input.thread_id,
turn_id = %input.turn_id,
"dropping compaction analytics event: missing thread lifecycle metadata"
);
return;
};
out.push(TrackEventRequest::Compaction(Box::new(
CodexCompactionEventRequest {
event_type: "codex_compaction_event",
event_params: codex_compaction_event_params(
input,
connection_state.app_server_client.clone(),
connection_state.runtime.clone(),
thread_metadata.thread_source,
thread_metadata.subagent_source.clone(),
thread_metadata.parent_thread_id.clone(),
),
},
)));
}
}
pub(crate) fn skill_id_for_local_skill(
codex-rs/app-server/src/message_processor.rs
+6 -5
View File
@@ -223,6 +223,11 @@ impl MessageProcessor {
auth_manager.set_external_auth(Arc::new(ExternalAuthRefreshBridge {
outgoing: outgoing.clone(),
}));
let analytics_events_client = AnalyticsEventsClient::new(
Arc::clone(&auth_manager),
config.chatgpt_base_url.trim_end_matches('/').to_string(),
config.analytics_enabled,
);
let thread_manager = Arc::new(ThreadManager::new(
config.as_ref(),
auth_manager.clone(),
@@ -233,12 +238,8 @@ impl MessageProcessor {
.enabled(Feature::DefaultModeRequestUserInput),
},
environment_manager,
Some(analytics_events_client.clone()),
));
let analytics_events_client = AnalyticsEventsClient::new(
Arc::clone(&auth_manager),
config.chatgpt_base_url.trim_end_matches('/').to_string(),
config.analytics_enabled,
);
thread_manager
.plugins_manager()
.set_analytics_events_client(analytics_events_client.clone());
codex-rs/core/src/codex.rs
+32 -6
View File
@@ -51,6 +51,8 @@ use chrono::Local;
use chrono::Utc;
use codex_analytics::AnalyticsEventsClient;
use codex_analytics::AppInvocation;
use codex_analytics::CompactionPhase;
use codex_analytics::CompactionReason;
use codex_analytics::InvocationType;
use codex_analytics::SubAgentThreadStartedInput;
use codex_analytics::build_track_events_context;
@@ -434,6 +436,7 @@ pub(crate) struct CodexSpawnArgs {
pub(crate) inherited_exec_policy: Option<Arc<ExecPolicyManager>>,
pub(crate) user_shell_override: Option<shell::Shell>,
pub(crate) parent_trace: Option<W3cTraceContext>,
pub(crate) analytics_events_client: Option<AnalyticsEventsClient>,
}
pub(crate) const INITIAL_SUBMIT_ID: &str = "";
@@ -486,6 +489,7 @@ impl Codex {
user_shell_override,
inherited_exec_policy,
parent_trace: _,
analytics_events_client,
} = args;
let (tx_sub, rx_sub) = async_channel::bounded(SUBMISSION_CHANNEL_CAPACITY);
let (tx_event, rx_event) = async_channel::unbounded();
@@ -676,6 +680,7 @@ impl Codex {
skills_watcher,
agent_control,
environment,
analytics_events_client,
)
.await
.map_err(|e| {
@@ -1604,6 +1609,7 @@ impl Session {
skills_watcher: Arc<SkillsWatcher>,
agent_control: AgentControl,
environment: Option<Arc<Environment>>,
analytics_events_client: Option<AnalyticsEventsClient>,
) -> anyhow::Result<Arc<Self>> {
debug!(
"Configuring session: model={}; provider={:?}",
@@ -1984,6 +1990,13 @@ impl Session {
}
let installation_id = resolve_installation_id(&config.codex_home).await?;
let analytics_events_client = analytics_events_client.unwrap_or_else(|| {
AnalyticsEventsClient::new(
Arc::clone(&auth_manager),
config.chatgpt_base_url.trim_end_matches('/').to_string(),
config.analytics_enabled,
)
});
let services = SessionServices {
// Initialize the MCP connection manager with an uninitialized
// instance. It will be replaced with one created via
@@ -2002,11 +2015,7 @@ impl Session {
),
shell_zsh_path: config.zsh_path.clone(),
main_execve_wrapper_exe: config.main_execve_wrapper_exe.clone(),
analytics_events_client: AnalyticsEventsClient::new(
Arc::clone(&auth_manager),
config.chatgpt_base_url.trim_end_matches('/').to_string(),
config.analytics_enabled,
),
analytics_events_client,
hooks,
rollout: Mutex::new(rollout_recorder),
user_shell: Arc::new(default_shell),
@@ -6361,6 +6370,8 @@ pub(crate) async fn run_turn(
&sess,
&turn_context,
InitialContextInjection::BeforeLastUserMessage,
CompactionReason::ContextLimit,
CompactionPhase::MidTurn,
)
.await
.is_err()
@@ -6544,7 +6555,14 @@ async fn run_pre_sampling_compact(
.unwrap_or(i64::MAX);
// Compact if the total usage tokens are greater than the auto compact limit
if total_usage_tokens >= auto_compact_limit {
run_auto_compact(sess, turn_context, InitialContextInjection::DoNotInject).await?;
run_auto_compact(
sess,
turn_context,
InitialContextInjection::DoNotInject,
CompactionReason::ContextLimit,
CompactionPhase::PreTurn,
)
.await?;
pre_sampling_compacted = true;
}
Ok(pre_sampling_compacted)
@@ -6588,6 +6606,8 @@ async fn maybe_run_previous_model_inline_compact(
sess,
&previous_model_turn_context,
InitialContextInjection::DoNotInject,
CompactionReason::ModelDownshift,
CompactionPhase::PreTurn,
)
.await?;
return Ok(true);
@@ -6599,12 +6619,16 @@ async fn run_auto_compact(
sess: &Arc<Session>,
turn_context: &Arc<TurnContext>,
initial_context_injection: InitialContextInjection,
reason: CompactionReason,
phase: CompactionPhase,
) -> CodexResult<()> {
if should_use_remote_compact_task(&turn_context.provider) {
run_inline_remote_auto_compact_task(
Arc::clone(sess),
Arc::clone(turn_context),
initial_context_injection,
reason,
phase,
)
.await?;
} else {
@@ -6612,6 +6636,8 @@ async fn run_auto_compact(
Arc::clone(sess),
Arc::clone(turn_context),
initial_context_injection,
reason,
phase,
)
.await?;
}
codex-rs/core/src/codex_delegate.rs
+1
View File
@@ -77,6 +77,7 @@ pub(crate) async fn run_codex_thread_interactive(
let CodexSpawnOk { codex, .. } = Codex::spawn(CodexSpawnArgs {
config,
auth_manager,
analytics_events_client: Some(parent_session.services.analytics_events_client.clone()),
models_manager,
environment_manager: Arc::new(EnvironmentManager::from_environment(
parent_ctx.environment.as_deref(),
codex-rs/core/src/codex_tests.rs
+1
View File
@@ -2748,6 +2748,7 @@ async fn session_new_fails_when_zsh_fork_enabled_without_zsh_path() {
.await
.expect("create environment"),
)),
/*analytics_events_client*/ None,
)
.await;
codex-rs/core/src/codex_tests_guardian.rs
+1
View File
@@ -436,6 +436,7 @@ async fn guardian_subagent_does_not_inherit_parent_exec_policy_rules() {
let CodexSpawnOk { codex, .. } = Codex::spawn(CodexSpawnArgs {
config,
auth_manager,
analytics_events_client: None,
models_manager,
environment_manager: Arc::new(EnvironmentManager::new(/*exec_server_url*/ None)),
skills_manager,
codex-rs/core/src/compact.rs
+147 -1
View File
@@ -1,4 +1,7 @@
use std::sync::Arc;
use std::time::Instant;
use std::time::SystemTime;
use std::time::UNIX_EPOCH;
use crate::Prompt;
use crate::client::ModelClientSession;
@@ -9,6 +12,14 @@ use crate::codex::Session;
use crate::codex::TurnContext;
use crate::codex::get_last_assistant_message_from_turn;
use crate::util::backoff;
use codex_analytics::CodexCompactionEvent;
use codex_analytics::CompactionImplementation;
use codex_analytics::CompactionPhase;
use codex_analytics::CompactionReason;
use codex_analytics::CompactionStatus;
use codex_analytics::CompactionStrategy;
use codex_analytics::CompactionTrigger;
use codex_features::Feature;
use codex_model_provider_info::ModelProviderInfo;
use codex_protocol::error::CodexErr;
use codex_protocol::error::Result as CodexResult;
@@ -55,6 +66,8 @@ pub(crate) async fn run_inline_auto_compact_task(
sess: Arc<Session>,
turn_context: Arc<TurnContext>,
initial_context_injection: InitialContextInjection,
reason: CompactionReason,
phase: CompactionPhase,
) -> CodexResult<()> {
let prompt = turn_context.compact_prompt().to_string();
let input = vec![UserInput::Text {
@@ -63,7 +76,16 @@ pub(crate) async fn run_inline_auto_compact_task(
text_elements: Vec::new(),
}];
run_compact_task_inner(sess, turn_context, input, initial_context_injection).await?;
run_compact_task_inner(
sess,
turn_context,
input,
initial_context_injection,
CompactionTrigger::Auto,
reason,
phase,
)
.await?;
Ok(())
}
@@ -84,6 +106,9 @@ pub(crate) async fn run_compact_task(
turn_context,
input,
InitialContextInjection::DoNotInject,
CompactionTrigger::Manual,
CompactionReason::UserRequested,
CompactionPhase::StandaloneTurn,
)
.await
}
@@ -93,6 +118,41 @@ async fn run_compact_task_inner(
turn_context: Arc<TurnContext>,
input: Vec<UserInput>,
initial_context_injection: InitialContextInjection,
trigger: CompactionTrigger,
reason: CompactionReason,
phase: CompactionPhase,
) -> CodexResult<()> {
let attempt = CompactionAnalyticsAttempt::begin(
sess.as_ref(),
turn_context.as_ref(),
trigger,
reason,
CompactionImplementation::Responses,
phase,
)
.await;
let result = run_compact_task_inner_impl(
Arc::clone(&sess),
Arc::clone(&turn_context),
input,
initial_context_injection,
)
.await;
attempt
.track(
sess.as_ref(),
compaction_status_from_result(&result),
result.as_ref().err().map(ToString::to_string),
)
.await;
result
}
async fn run_compact_task_inner_impl(
sess: Arc<Session>,
turn_context: Arc<TurnContext>,
input: Vec<UserInput>,
initial_context_injection: InitialContextInjection,
) -> CodexResult<()> {
let compaction_item = TurnItem::ContextCompaction(ContextCompactionItem::new());
sess.emit_turn_item_started(&turn_context, &compaction_item)
@@ -233,6 +293,92 @@ async fn run_compact_task_inner(
Ok(())
}
pub(crate) struct CompactionAnalyticsAttempt {
enabled: bool,
thread_id: String,
turn_id: String,
trigger: CompactionTrigger,
reason: CompactionReason,
implementation: CompactionImplementation,
phase: CompactionPhase,
active_context_tokens_before: i64,
started_at: u64,
start_instant: Instant,
}
impl CompactionAnalyticsAttempt {
pub(crate) async fn begin(
sess: &Session,
turn_context: &TurnContext,
trigger: CompactionTrigger,
reason: CompactionReason,
implementation: CompactionImplementation,
phase: CompactionPhase,
) -> Self {
let enabled = sess.enabled(Feature::GeneralAnalytics);
let active_context_tokens_before = sess.get_total_token_usage().await;
Self {
enabled,
thread_id: sess.conversation_id.to_string(),
turn_id: turn_context.sub_id.clone(),
trigger,
reason,
implementation,
phase,
active_context_tokens_before,
started_at: now_unix_seconds(),
start_instant: Instant::now(),
}
}
pub(crate) async fn track(
self,
sess: &Session,
status: CompactionStatus,
error: Option<String>,
) {
if !self.enabled {
return;
}
let active_context_tokens_after = sess.get_total_token_usage().await;
sess.services
.analytics_events_client
.track_compaction(CodexCompactionEvent {
thread_id: self.thread_id,
turn_id: self.turn_id,
trigger: self.trigger,
reason: self.reason,
implementation: self.implementation,
phase: self.phase,
strategy: CompactionStrategy::Memento,
status,
error,
active_context_tokens_before: self.active_context_tokens_before,
active_context_tokens_after,
started_at: self.started_at,
completed_at: now_unix_seconds(),
duration_ms: Some(
u64::try_from(self.start_instant.elapsed().as_millis()).unwrap_or(u64::MAX),
),
});
}
}
pub(crate) fn compaction_status_from_result<T>(result: &CodexResult<T>) -> CompactionStatus {
match result {
Ok(_) => CompactionStatus::Completed,
Err(CodexErr::Interrupted | CodexErr::TurnAborted) => CompactionStatus::Interrupted,
Err(_) => CompactionStatus::Failed,
}
}
fn now_unix_seconds() -> u64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|duration| duration.as_secs())
.unwrap_or_default()
}
pub fn content_items_to_text(content: &[ContentItem]) -> Option<String> {
let mut pieces = Vec::new();
for item in content {
codex-rs/core/src/compact_remote.rs
+48 -5
View File
@@ -5,12 +5,18 @@ use crate::Prompt;
use crate::codex::Session;
use crate::codex::TurnContext;
use crate::codex::built_tools;
use crate::compact::CompactionAnalyticsAttempt;
use crate::compact::InitialContextInjection;
use crate::compact::compaction_status_from_result;
use crate::compact::insert_initial_context_before_last_real_user_or_summary;
use crate::context_manager::ContextManager;
use crate::context_manager::TotalTokenUsageBreakdown;
use crate::context_manager::estimate_response_item_model_visible_bytes;
use crate::context_manager::is_codex_generated_item;
use codex_analytics::CompactionImplementation;
use codex_analytics::CompactionPhase;
use codex_analytics::CompactionReason;
use codex_analytics::CompactionTrigger;
use codex_protocol::error::CodexErr;
use codex_protocol::error::Result as CodexResult;
use codex_protocol::items::ContextCompactionItem;
@@ -29,8 +35,18 @@ pub(crate) async fn run_inline_remote_auto_compact_task(
sess: Arc<Session>,
turn_context: Arc<TurnContext>,
initial_context_injection: InitialContextInjection,
reason: CompactionReason,
phase: CompactionPhase,
) -> CodexResult<()> {
run_remote_compact_task_inner(&sess, &turn_context, initial_context_injection).await?;
run_remote_compact_task_inner(
&sess,
&turn_context,
initial_context_injection,
CompactionTrigger::Auto,
reason,
phase,
)
.await?;
Ok(())
}
@@ -46,17 +62,44 @@ pub(crate) async fn run_remote_compact_task(
});
sess.send_event(&turn_context, start_event).await;
run_remote_compact_task_inner(&sess, &turn_context, InitialContextInjection::DoNotInject).await
run_remote_compact_task_inner(
&sess,
&turn_context,
InitialContextInjection::DoNotInject,
CompactionTrigger::Manual,
CompactionReason::UserRequested,
CompactionPhase::StandaloneTurn,
)
.await
}
async fn run_remote_compact_task_inner(
sess: &Arc<Session>,
turn_context: &Arc<TurnContext>,
initial_context_injection: InitialContextInjection,
trigger: CompactionTrigger,
reason: CompactionReason,
phase: CompactionPhase,
) -> CodexResult<()> {
if let Err(err) =
run_remote_compact_task_inner_impl(sess, turn_context, initial_context_injection).await
{
let attempt = CompactionAnalyticsAttempt::begin(
sess.as_ref(),
turn_context.as_ref(),
trigger,
reason,
CompactionImplementation::ResponsesCompact,
phase,
)
.await;
let result =
run_remote_compact_task_inner_impl(sess, turn_context, initial_context_injection).await;
attempt
.track(
sess.as_ref(),
compaction_status_from_result(&result),
result.as_ref().err().map(ToString::to_string),
)
.await;
if let Err(err) = result {
let event = EventMsg::Error(
err.to_error_event(Some("Error running remote compact task".to_string())),
);
codex-rs/core/src/prompt_debug.rs
+1
View File
@@ -39,6 +39,7 @@ pub async fn build_prompt_input(
.enabled(Feature::DefaultModeRequestUserInput),
},
Arc::new(EnvironmentManager::from_env()),
/*analytics_events_client*/ None,
);
let thread = thread_manager.start_thread(config).await?;
codex-rs/core/src/thread_manager.rs
+6
View File
@@ -15,6 +15,7 @@ use crate::shell_snapshot::ShellSnapshot;
use crate::skills_watcher::SkillsWatcher;
use crate::skills_watcher::SkillsWatcherEvent;
use crate::tasks::interrupted_turn_history_marker;
use codex_analytics::AnalyticsEventsClient;
use codex_app_server_protocol::ThreadHistoryBuilder;
use codex_app_server_protocol::TurnStatus;
use codex_exec_server::EnvironmentManager;
@@ -210,6 +211,7 @@ pub(crate) struct ThreadManagerState {
mcp_manager: Arc<McpManager>,
skills_watcher: Arc<SkillsWatcher>,
session_source: SessionSource,
analytics_events_client: Option<AnalyticsEventsClient>,
// Captures submitted ops for testing purpose when test mode is enabled.
ops_log: Option<SharedCapturedOps>,
}
@@ -221,6 +223,7 @@ impl ThreadManager {
session_source: SessionSource,
collaboration_modes_config: CollaborationModesConfig,
environment_manager: Arc<EnvironmentManager>,
analytics_events_client: Option<AnalyticsEventsClient>,
) -> Self {
let codex_home = config.codex_home.clone();
let restriction_product = session_source.restriction_product();
@@ -259,6 +262,7 @@ impl ThreadManager {
skills_watcher,
auth_manager,
session_source,
analytics_events_client,
ops_log: should_use_test_thread_manager_behavior()
.then(|| Arc::new(std::sync::Mutex::new(Vec::new()))),
}),
@@ -328,6 +332,7 @@ impl ThreadManager {
skills_watcher,
auth_manager,
session_source: SessionSource::Exec,
analytics_events_client: None,
ops_log: should_use_test_thread_manager_behavior()
.then(|| Arc::new(std::sync::Mutex::new(Vec::new()))),
}),
@@ -916,6 +921,7 @@ impl ThreadManagerState {
inherited_exec_policy,
user_shell_override,
parent_trace,
analytics_events_client: self.analytics_events_client.clone(),
})
.await?;
self.finalize_thread_spawn(codex, thread_id, watch_registration)
codex-rs/core/src/thread_manager_tests.rs
+4
View File
@@ -299,6 +299,7 @@ async fn new_uses_configured_openai_provider_for_model_refresh() {
Arc::new(codex_exec_server::EnvironmentManager::new(
/*exec_server_url*/ None,
)),
/*analytics_events_client*/ None,
);
let _ = manager.list_models(RefreshStrategy::Online).await;
@@ -435,6 +436,7 @@ async fn interrupted_fork_snapshot_does_not_synthesize_turn_id_for_legacy_histor
Arc::new(codex_exec_server::EnvironmentManager::new(
/*exec_server_url*/ None,
)),
/*analytics_events_client*/ None,
);
let source = manager
@@ -537,6 +539,7 @@ async fn interrupted_fork_snapshot_preserves_explicit_turn_id() {
Arc::new(codex_exec_server::EnvironmentManager::new(
/*exec_server_url*/ None,
)),
/*analytics_events_client*/ None,
);
let source = manager
@@ -629,6 +632,7 @@ async fn interrupted_fork_snapshot_uses_persisted_mid_turn_history_without_live_
Arc::new(codex_exec_server::EnvironmentManager::new(
/*exec_server_url*/ None,
)),
/*analytics_events_client*/ None,
);
let source = manager
codex-rs/core/tests/common/test_codex.rs
+1
View File
@@ -521,6 +521,7 @@ impl TestCodexBuilder {
SessionSource::Exec,
CollaborationModesConfig::default(),
Arc::clone(&environment_manager),
/*analytics_events_client*/ None,
)
} else {
codex_core::test_support::thread_manager_with_models_provider_and_home(
codex-rs/core/tests/suite/client.rs
+1
View File
@@ -1103,6 +1103,7 @@ async fn prefers_apikey_when_config_prefers_apikey_even_with_chatgpt_tokens() {
Arc::new(codex_exec_server::EnvironmentManager::new(
/*exec_server_url*/ None,
)),
/*analytics_events_client*/ None,
);
let NewThread { thread: codex, .. } = thread_manager
.start_thread(config)
codex-rs/mcp-server/src/message_processor.rs
+1
View File
@@ -70,6 +70,7 @@ impl MessageProcessor {
.enabled(Feature::DefaultModeRequestUserInput),
},
environment_manager,
/*analytics_events_client*/ None,
));
Self {
outgoing,