diff --git a/codex-rs/analytics-materializer/src/context_windows.rs b/codex-rs/analytics-materializer/src/context_windows.rs new file mode 100644 index 0000000000..aacb7a907f --- /dev/null +++ b/codex-rs/analytics-materializer/src/context_windows.rs @@ -0,0 +1,351 @@ +use crate::synthetic_conversation; +use crate::synthetic_conversation::SYNTHETIC_CONVERSATION_SOURCE; +use anyhow::Context; +use anyhow::Result; +use anyhow::bail; +use duckdb::Connection; +use duckdb::params_from_iter; +use duckdb::types::Value as DuckValue; +use serde_json::Value as JsonValue; +use std::collections::HashMap; + +pub(super) fn materialize_context_windows(connection: &Connection) -> Result<()> { + let calls = load_context_window_calls(connection)?; + let mut windows = reduce_context_windows(calls)?; + assign_window_metadata(&mut windows); + insert_context_windows(connection, &windows)?; + Ok(()) +} + +#[derive(Clone, Debug, Eq, PartialEq)] +struct ContextWindowKey { + session_id: String, + thread_id: String, + context_window_id: String, +} + +#[derive(Clone, Debug)] +struct ResponsesCallRow { + session_id: String, + thread_id: String, + turn_id: Option, + responses_call_id: String, + context_window_id: String, + status: String, + request_started_at_epoch_millis: i64, + response_id: Option, + request_json: String, + response_json: Option, +} + +impl ResponsesCallRow { + fn context_window_key(&self) -> ContextWindowKey { + ContextWindowKey { + session_id: self.session_id.clone(), + thread_id: self.thread_id.clone(), + context_window_id: self.context_window_id.clone(), + } + } +} + +#[derive(Clone)] +struct CompletedCall { + full_input: Vec, + output_items: Vec, +} + +struct ReconstructedPrompt { + request_json: JsonValue, + full_input: Vec, +} + +struct ContextWindowDraft { + session_id: String, + thread_id: String, + context_window_id: String, + context_window_ordinal: i64, + first_turn_id: Option, + last_turn_id: Option, + first_responses_call_id: String, + last_responses_call_id: String, + opened_at_epoch_millis: i64, + closed_at_epoch_millis: Option, + close_reason: Option, + message_count: i64, + conversation_json: String, +} + +fn load_context_window_calls(connection: &Connection) -> Result> { + let mut statement = connection.prepare( + r#" + SELECT + session_id, + thread_id, + turn_id, + responses_call_id, + context_window_id, + status, + request_started_at_epoch_millis, + response_id, + request_json, + response_json + FROM viewer_responses_calls_v1 + WHERE session_id IS NOT NULL + AND thread_id IS NOT NULL + AND context_window_id IS NOT NULL + ORDER BY + session_id, + thread_id, + context_window_id, + request_started_at_epoch_millis, + responses_call_id + "#, + )?; + let calls = statement.query_map([], |row| { + Ok(ResponsesCallRow { + session_id: row.get(0)?, + thread_id: row.get(1)?, + turn_id: row.get(2)?, + responses_call_id: row.get(3)?, + context_window_id: row.get(4)?, + status: row.get(5)?, + request_started_at_epoch_millis: row.get(6)?, + response_id: row.get(7)?, + request_json: row.get(8)?, + response_json: row.get(9)?, + }) + })?; + Ok(calls.collect::>>()?) +} + +fn reduce_context_windows(calls: Vec) -> Result> { + let mut windows = Vec::new(); + let mut current_key: Option = None; + let mut current_calls = Vec::new(); + + for call in calls { + let call_key = call.context_window_key(); + if current_key.as_ref().is_some_and(|key| key != &call_key) { + windows.push(reduce_context_window_calls(std::mem::take( + &mut current_calls, + ))?); + } + current_key = Some(call_key); + current_calls.push(call); + } + if !current_calls.is_empty() { + windows.push(reduce_context_window_calls(current_calls)?); + } + Ok(windows) +} + +fn reduce_context_window_calls(calls: Vec) -> Result { + let first = calls + .first() + .context("context window reducer received no Responses calls")?; + let last = calls + .last() + .context("context window reducer received no Responses calls")?; + let key = first.context_window_key(); + let mut completed_calls = HashMap::::new(); + let mut last_prompt = None; + + for call in &calls { + let request_json: JsonValue = + serde_json::from_str(&call.request_json).with_context(|| { + format!( + "parse request_json for Responses call {}", + call.responses_call_id + ) + })?; + let delta_input = request_input_items(&request_json, call)?; + let full_input = if let Some(previous_response_id) = request_json + .get("previous_response_id") + .and_then(JsonValue::as_str) + { + let previous = completed_calls + .get(previous_response_id) + .with_context(|| { + format!( + "unknown incremental Responses predecessor {previous_response_id} for call {} in context window {}", + call.responses_call_id, call.context_window_id + ) + })?; + let mut full_input = previous.full_input.clone(); + full_input.extend(previous.output_items.clone()); + full_input.extend(delta_input); + full_input + } else { + delta_input + }; + let output_items = response_output_items(call)?; + if call.status == "completed" + && let Some(response_id) = call.response_id.as_ref() + && completed_calls + .insert( + response_id.clone(), + CompletedCall { + full_input: full_input.clone(), + output_items, + }, + ) + .is_some() + { + bail!( + "duplicate completed Responses response_id {response_id} in context window {}", + call.context_window_id + ); + } + last_prompt = Some(ReconstructedPrompt { + request_json, + full_input, + }); + } + + let last_prompt = + last_prompt.context("context window reducer did not reconstruct a Responses prompt")?; + let conversation = synthetic_conversation::synthetic_conversation( + &key.session_id, + &key.thread_id, + &key.context_window_id, + &last_prompt.request_json, + &last_prompt.full_input, + )?; + let message_count = conversation + .get("messages") + .and_then(JsonValue::as_array) + .map_or(0, std::vec::Vec::len); + + Ok(ContextWindowDraft { + session_id: key.session_id, + thread_id: key.thread_id, + context_window_id: key.context_window_id, + context_window_ordinal: 0, + first_turn_id: first.turn_id.clone(), + last_turn_id: last.turn_id.clone(), + first_responses_call_id: first.responses_call_id.clone(), + last_responses_call_id: last.responses_call_id.clone(), + opened_at_epoch_millis: first.request_started_at_epoch_millis, + closed_at_epoch_millis: None, + close_reason: None, + message_count: i64::try_from(message_count)?, + conversation_json: serde_json::to_string(&conversation)?, + }) +} + +fn request_input_items( + request_json: &JsonValue, + call: &ResponsesCallRow, +) -> Result> { + request_json + .get("input") + .and_then(JsonValue::as_array) + .cloned() + .with_context(|| { + format!( + "Responses call {} request_json has no input array", + call.responses_call_id + ) + }) +} + +fn response_output_items(call: &ResponsesCallRow) -> Result> { + let Some(response_json) = call.response_json.as_ref() else { + return Ok(Vec::new()); + }; + let response_json: JsonValue = serde_json::from_str(response_json).with_context(|| { + format!( + "parse response_json for Responses call {}", + call.responses_call_id + ) + })?; + response_json + .get("output_items") + .and_then(JsonValue::as_array) + .cloned() + .with_context(|| { + format!( + "Responses call {} response_json has no output_items array", + call.responses_call_id + ) + }) +} + +fn assign_window_metadata(windows: &mut [ContextWindowDraft]) { + windows.sort_by(|left, right| { + ( + left.session_id.as_str(), + left.thread_id.as_str(), + left.opened_at_epoch_millis, + left.context_window_id.as_str(), + ) + .cmp(&( + right.session_id.as_str(), + right.thread_id.as_str(), + right.opened_at_epoch_millis, + right.context_window_id.as_str(), + )) + }); + + let mut previous_index: Option = None; + let mut context_window_ordinal = 0; + for index in 0..windows.len() { + if let Some(previous_index) = previous_index + && windows[previous_index].session_id == windows[index].session_id + && windows[previous_index].thread_id == windows[index].thread_id + { + context_window_ordinal += 1; + windows[previous_index].closed_at_epoch_millis = + Some(windows[index].opened_at_epoch_millis); + windows[previous_index].close_reason = Some("compaction".to_string()); + } else { + context_window_ordinal = 1; + } + windows[index].context_window_ordinal = context_window_ordinal; + previous_index = Some(index); + } +} + +fn insert_context_windows(connection: &Connection, windows: &[ContextWindowDraft]) -> Result<()> { + let mut statement = connection.prepare( + r#" + INSERT INTO viewer_context_windows_v1 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + "#, + )?; + for window in windows { + statement.execute(params_from_iter([ + text(window.session_id.clone()), + text(window.thread_id.clone()), + text(window.context_window_id.clone()), + integer(window.context_window_ordinal), + optional_text(window.first_turn_id.clone()), + optional_text(window.last_turn_id.clone()), + text(window.first_responses_call_id.clone()), + text(window.last_responses_call_id.clone()), + integer(window.opened_at_epoch_millis), + optional_integer(window.closed_at_epoch_millis), + optional_text(window.close_reason.clone()), + integer(window.message_count), + text(window.conversation_json.clone()), + text(SYNTHETIC_CONVERSATION_SOURCE.to_string()), + DuckValue::Boolean(true), + ]))?; + } + Ok(()) +} + +fn text(value: String) -> DuckValue { + DuckValue::Text(value) +} + +fn optional_text(value: Option) -> DuckValue { + value.map_or(DuckValue::Null, DuckValue::Text) +} + +fn integer(value: i64) -> DuckValue { + DuckValue::BigInt(value) +} + +fn optional_integer(value: Option) -> DuckValue { + value.map_or(DuckValue::Null, DuckValue::BigInt) +} diff --git a/codex-rs/analytics-materializer/src/lib.rs b/codex-rs/analytics-materializer/src/lib.rs index b8ea8184d5..9761a520e4 100644 --- a/codex-rs/analytics-materializer/src/lib.rs +++ b/codex-rs/analytics-materializer/src/lib.rs @@ -16,6 +16,10 @@ use std::io::BufReader; use std::path::Path; use std::path::PathBuf; +mod context_windows; +mod synthetic_chat; +mod synthetic_conversation; + const MATERIALIZE_SQL: &str = r#" CREATE TEMP VIEW codex_events AS SELECT @@ -102,6 +106,7 @@ SELECT thread_id, turn_id, json_extract_string(payload, '$.responses_call_id') AS responses_call_id, + json_extract_string(payload, '$.context_window_id') AS context_window_id, row_number() OVER ( PARTITION BY session_id, thread_id, turn_id ORDER BY TRY_CAST(json_extract_string(payload, '$.request_started_at_epoch_millis') AS BIGINT), json_extract_string(payload, '$.responses_call_id') @@ -195,7 +200,7 @@ SELECT sum(CASE WHEN json_extract_string(payload, '$.event_params.requested_additional_permissions') = 'true' THEN 1 ELSE 0 END) AS tool_calls_requested_additional_permissions_count, coalesce(sum(TRY_CAST(json_extract_string(payload, '$.event_params.duration_ms') AS BIGINT)), 0) AS tool_calls_total_duration_ms FROM codex_events -WHERE event_type IN ('codex_command_execution', 'codex_file_change', 'codex_mcp_tool_call', 'codex_dynamic_tool_call', 'codex_collab_agent_tool_call', 'codex_web_search', 'codex_image_generation') +WHERE event_type IN ('codex_command_execution_event', 'codex_file_change_event', 'codex_mcp_tool_call_event', 'codex_dynamic_tool_call_event', 'codex_collab_agent_tool_call_event', 'codex_web_search_event', 'codex_image_generation_event') GROUP BY thread_id, turn_id; CREATE TEMP VIEW viewer_turn_compaction_aggregates AS @@ -207,7 +212,7 @@ SELECT sum(CASE WHEN json_extract_string(payload, '$.event_params.status') = 'failed' THEN 1 ELSE 0 END) AS compactions_failed_count, bool_or(NULLIF(CAST(json_extract(payload, '$.event_params.error') AS VARCHAR), 'null') IS NOT NULL) AS compactions_any_error FROM codex_events -WHERE event_type = 'codex_compaction' +WHERE event_type = 'codex_compaction_event' GROUP BY thread_id, turn_id; CREATE TEMP VIEW viewer_turn_review_aggregates AS @@ -404,6 +409,7 @@ pub fn process_local_analytics(input: impl AsRef, output: impl AsRef create_raw_records_table(&connection)?; insert_local_records(&connection, input)?; connection.execute_batch(MATERIALIZE_SQL)?; + context_windows::materialize_context_windows(&connection)?; Ok(()) } diff --git a/codex-rs/analytics-materializer/src/lib_tests.rs b/codex-rs/analytics-materializer/src/lib_tests.rs index 54cc2d9745..9c05884591 100644 --- a/codex-rs/analytics-materializer/src/lib_tests.rs +++ b/codex-rs/analytics-materializer/src/lib_tests.rs @@ -22,7 +22,7 @@ fn default_output_replaces_input_extension() { } #[test] -fn materializes_threads_turns_events_responses_and_context_stub() { +fn materializes_threads_turns_events_responses_and_context_windows() { let input = test_path("local-analytics.jsonl"); let output = test_path("local-analytics.duckdb"); let records = [ @@ -107,7 +107,7 @@ fn materializes_threads_turns_events_responses_and_context_stub() { Some("thread-child"), Some("turn-1"), json!({ - "event_type": "codex_command_execution", + "event_type": "codex_command_execution_event", "event_params": { "thread_id": "thread-child", "turn_id": "turn-1", @@ -121,17 +121,41 @@ fn materializes_threads_turns_events_responses_and_context_stub() { } }), ), - responses_record( + local_record( 5, + None, + Some("thread-child"), + Some("turn-1"), + json!({ + "event_type": "codex_compaction_event", + "event_params": { + "thread_id": "thread-child", + "turn_id": "turn-1", + "status": "completed", + "error": null + } + }), + ), + responses_record( + 6, json!({ "responses_call_id": "call-1", + "context_window_id": "thread-child:0", "transport": "http", "status": "completed", "request_started_at_epoch_millis": 100, "completed_at_epoch_millis": 110, "response_id": "response-1", "upstream_request_id": "request-1", - "request_json": {"model": "gpt-5"}, + "request_json": { + "model": "gpt-5", + "instructions": "Be concise.", + "input": [{ + "type": "message", + "role": "user", + "content": [{"type": "input_text", "text": "hello"}] + }] + }, "response_json": {"output_items": []}, "token_usage_json": {"total_tokens": 3}, "error_json": null @@ -151,9 +175,9 @@ fn materializes_threads_turns_events_responses_and_context_stub() { ) .expect("child thread should exist"); assert_eq!(child_thread, ("thread-root".to_string(), false)); - let turn: (i64, i64, i64, i64, i64, i64) = connection + let turn: (i64, i64, i64, i64, i64, i64, i64) = connection .query_row( - "SELECT turn_ordinal, tool_calls_count, tool_calls_failure_count, responses_api_calls_total_count, responses_api_calls_succeeded_count, responses_api_calls_total_latency_ms FROM viewer_turns_v1 WHERE turn_id = 'turn-1'", + "SELECT turn_ordinal, compactions_count, tool_calls_count, tool_calls_failure_count, responses_api_calls_total_count, responses_api_calls_succeeded_count, responses_api_calls_total_latency_ms FROM viewer_turns_v1 WHERE turn_id = 'turn-1'", [], |row| { Ok(( @@ -163,35 +187,243 @@ fn materializes_threads_turns_events_responses_and_context_stub() { row.get(3)?, row.get(4)?, row.get(5)?, + row.get(6)?, )) }, ) .expect("turn should exist"); - assert_eq!(turn, (1, 1, 1, 1, 1, 10)); + assert_eq!(turn, (1, 1, 1, 1, 1, 1, 10)); let turn_event: (String, i64) = connection .query_row( - "SELECT session_id, event_seq FROM viewer_turn_events_v1 WHERE event_type = 'codex_command_execution'", + "SELECT session_id, event_seq FROM viewer_turn_events_v1 WHERE event_type = 'codex_command_execution_event'", [], |row| Ok((row.get(0)?, row.get(1)?)), ) .expect("tool event should exist"); assert_eq!(turn_event, ("session-1".to_string(), 2)); - let response_call: (i64, String) = connection + let response_call: (i64, String, String) = connection .query_row( - "SELECT call_ordinal, request_json FROM viewer_responses_calls_v1", + "SELECT call_ordinal, context_window_id, request_json FROM viewer_responses_calls_v1", + [], + |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)), + ) + .expect("responses call should exist"); + assert_eq!(response_call.0, 1); + assert_eq!(response_call.1, "thread-child:0"); + assert_eq!( + serde_json::from_str::(&response_call.2) + .expect("request json should parse"), + json!({ + "model": "gpt-5", + "instructions": "Be concise.", + "input": [{ + "type": "message", + "role": "user", + "content": [{"type": "input_text", "text": "hello"}] + }] + }) + ); + let context_window: (i64, i64, String, bool, String) = connection + .query_row( + "SELECT context_window_ordinal, message_count, conversation_source, is_synthetic, conversation_json FROM viewer_context_windows_v1", + [], + |row| { + Ok(( + row.get(0)?, + row.get(1)?, + row.get(2)?, + row.get(3)?, + row.get(4)?, + )) + }, + ) + .expect("context window should exist"); + assert_eq!( + ( + context_window.0, + context_window.1, + context_window.2, + context_window.3 + ), + (1, 2, "synthetic_local_responses_request".to_string(), true) + ); + let conversation: serde_json::Value = + serde_json::from_str(&context_window.4).expect("conversation should parse"); + assert_eq!( + conversation_author_roles(&conversation), + vec!["developer".to_string(), "user".to_string()] + ); +} + +#[test] +fn materializes_incremental_responses_prompt_in_context_window() { + let input = test_path("incremental-local-analytics.jsonl"); + let output = test_path("incremental-local-analytics.duckdb"); + let records = [ + responses_record( + 1, + json!({ + "responses_call_id": "call-1", + "context_window_id": "thread-child:0", + "transport": "websocket", + "status": "completed", + "request_started_at_epoch_millis": 100, + "completed_at_epoch_millis": 110, + "response_id": "response-1", + "upstream_request_id": "request-1", + "request_json": { + "type": "response.create", + "model": "gpt-5", + "input": [{ + "type": "message", + "role": "user", + "content": [{"type": "input_text", "text": "hello"}] + }] + }, + "response_json": { + "output_items": [{ + "type": "message", + "role": "assistant", + "content": [{"type": "output_text", "text": "hi"}], + "phase": "final_answer" + }] + }, + "token_usage_json": null, + "error_json": null + }), + ), + responses_record( + 2, + json!({ + "responses_call_id": "call-2", + "context_window_id": "thread-child:0", + "transport": "websocket", + "status": "completed", + "request_started_at_epoch_millis": 120, + "completed_at_epoch_millis": 130, + "response_id": "response-2", + "upstream_request_id": "request-2", + "request_json": { + "type": "response.create", + "model": "gpt-5", + "previous_response_id": "response-1", + "input": [{ + "type": "message", + "role": "user", + "content": [{"type": "input_text", "text": "next"}] + }] + }, + "response_json": {"output_items": []}, + "token_usage_json": null, + "error_json": null + }), + ), + ]; + write_jsonl(&input, &records); + + process_local_analytics(&input, &output).expect("materialization should succeed"); + + let connection = Connection::open(&output).expect("DuckDB should open"); + let (message_count, conversation_json): (i64, String) = connection + .query_row( + "SELECT message_count, conversation_json FROM viewer_context_windows_v1", [], |row| Ok((row.get(0)?, row.get(1)?)), ) - .expect("responses call should exist"); - assert_eq!(response_call, (1, "{\"model\":\"gpt-5\"}".to_string())); - let context_rows: i64 = connection - .query_row( - "SELECT count(*) FROM viewer_context_windows_v1", - [], - |row| row.get(0), + .expect("context window should exist"); + assert_eq!(message_count, 3); + let conversation: serde_json::Value = + serde_json::from_str(&conversation_json).expect("conversation should parse"); + assert_eq!( + conversation_message_texts(&conversation), + vec!["hello".to_string(), "hi".to_string(), "next".to_string()] + ); +} + +#[test] +fn closes_previous_context_window_when_new_codex_window_opens() { + let input = test_path("compacted-local-analytics.jsonl"); + let output = test_path("compacted-local-analytics.duckdb"); + let records = [ + responses_record( + 1, + completed_responses_payload("call-1", "thread-child:0", 100, "response-1"), + ), + responses_record( + 2, + completed_responses_payload("call-2", "thread-child:1", 200, "response-2"), + ), + ]; + write_jsonl(&input, &records); + + process_local_analytics(&input, &output).expect("materialization should succeed"); + + let connection = Connection::open(&output).expect("DuckDB should open"); + let mut statement = connection + .prepare( + "SELECT context_window_id, context_window_ordinal, closed_at_epoch_millis, close_reason FROM viewer_context_windows_v1 ORDER BY context_window_ordinal", ) - .expect("context window stub should exist"); - assert_eq!(context_rows, 0); + .expect("query should prepare"); + let context_windows = statement + .query_map([], |row| { + Ok(( + row.get::<_, String>(0)?, + row.get::<_, i64>(1)?, + row.get::<_, Option>(2)?, + row.get::<_, Option>(3)?, + )) + }) + .expect("query should run") + .collect::>>() + .expect("rows should collect"); + assert_eq!( + context_windows, + vec![ + ( + "thread-child:0".to_string(), + 1, + Some(200), + Some("compaction".to_string()) + ), + ("thread-child:1".to_string(), 2, None, None) + ] + ); +} + +#[test] +fn rejects_unknown_incremental_responses_predecessor() { + let input = test_path("missing-predecessor-local-analytics.jsonl"); + let output = test_path("missing-predecessor-local-analytics.duckdb"); + let records = [responses_record( + 1, + json!({ + "responses_call_id": "call-1", + "context_window_id": "thread-child:0", + "transport": "websocket", + "status": "completed", + "request_started_at_epoch_millis": 100, + "completed_at_epoch_millis": 110, + "response_id": "response-1", + "upstream_request_id": "request-1", + "request_json": { + "type": "response.create", + "model": "gpt-5", + "previous_response_id": "missing-response", + "input": [] + }, + "response_json": {"output_items": []}, + "token_usage_json": null, + "error_json": null + }), + )]; + write_jsonl(&input, &records); + + let err = process_local_analytics(&input, &output) + .expect_err("unknown predecessor should fail materialization"); + assert!( + err.to_string() + .contains("unknown incremental Responses predecessor missing-response") + ); } fn turn_event_payload() -> serde_json::Value { @@ -300,6 +532,63 @@ fn responses_record( } } +fn completed_responses_payload( + responses_call_id: &str, + context_window_id: &str, + request_started_at_epoch_millis: u64, + response_id: &str, +) -> serde_json::Value { + json!({ + "responses_call_id": responses_call_id, + "context_window_id": context_window_id, + "transport": "http", + "status": "completed", + "request_started_at_epoch_millis": request_started_at_epoch_millis, + "completed_at_epoch_millis": request_started_at_epoch_millis + 10, + "response_id": response_id, + "upstream_request_id": null, + "request_json": { + "model": "gpt-5", + "input": [{ + "type": "message", + "role": "user", + "content": [{"type": "input_text", "text": context_window_id}] + }] + }, + "response_json": {"output_items": []}, + "token_usage_json": null, + "error_json": null + }) +} + +fn conversation_author_roles(conversation: &serde_json::Value) -> Vec { + conversation["messages"] + .as_array() + .expect("conversation should have messages") + .iter() + .map(|message| { + message["author"]["role"] + .as_str() + .expect("message should have author role") + .to_string() + }) + .collect() +} + +fn conversation_message_texts(conversation: &serde_json::Value) -> Vec { + conversation["messages"] + .as_array() + .expect("conversation should have messages") + .iter() + .map(|message| { + message["content"]["parts"][0] + .as_str() + .expect("message should have first text part") + .to_string() + }) + .collect() +} + fn write_jsonl(path: &PathBuf, records: &[LocalAnalyticsRecord]) { let contents = records .iter() diff --git a/codex-rs/analytics-materializer/src/synthetic_chat.rs b/codex-rs/analytics-materializer/src/synthetic_chat.rs new file mode 100644 index 0000000000..9250f5dced --- /dev/null +++ b/codex-rs/analytics-materializer/src/synthetic_chat.rs @@ -0,0 +1,166 @@ +use anyhow::Context; +use anyhow::Result; +use anyhow::bail; +use serde_json::Value as JsonValue; +use serde_json::json; +use std::collections::HashMap; + +pub(super) const DEFAULT_RECIPIENT: &str = "all"; + +pub(super) struct SyntheticMessage { + pub(super) role: String, + pub(super) author_name: Option, + pub(super) content: JsonValue, + pub(super) recipient: String, + pub(super) channel: Option, + pub(super) end_turn: Option, +} + +pub(super) fn push_synthetic_message( + messages: &mut Vec, + conversation_seed: &str, + message: SyntheticMessage, +) { + let message_seed = format!("{conversation_seed}:message:{}", messages.len()); + let weight = if message.role == "user" { 0.0 } else { 1.0 }; + messages.push(json!({ + "id": stable_uuid(&message_seed), + "author": { + "role": message.role, + "name": message.author_name, + "metadata": {} + }, + "create_time": null, + "update_time": null, + "content": message.content, + "status": "finished_successfully", + "end_turn": message.end_turn, + "weight": weight, + "metadata": {}, + "recipient": message.recipient, + "channel": message.channel + })); +} + +pub(super) fn developer_tools_content(tools: &[JsonValue]) -> JsonValue { + json!({ + "content_type": "developer_content", + "instructions": [""], + "settings": null, + "function_namespaces": [{ + "name": "functions", + "description": "", + "functions": tools + }], + "response_formats": [] + }) +} + +pub(super) fn text_content(text: String) -> JsonValue { + json!({ + "content_type": "text", + "parts": [text] + }) +} + +pub(super) fn render_content(value: &JsonValue) -> Result { + match value { + JsonValue::Array(parts) => parts + .iter() + .map(render_content_part) + .collect::>>() + .map(|parts| { + parts + .into_iter() + .filter(|part| !part.is_empty()) + .collect::>() + .join("\n") + }), + JsonValue::String(text) => Ok(text.clone()), + JsonValue::Null => Ok(String::new()), + _ => render_content_part(value), + } +} + +pub(super) fn message_channel(role: &str, phase: Option<&str>) -> Option<&'static str> { + if role != "assistant" { + return None; + } + match phase { + Some("final_answer") => Some("final"), + Some("commentary") | None => Some("commentary"), + Some(_) => Some("commentary"), + } +} + +pub(super) fn message_end_turn(role: &str, phase: Option<&str>) -> Option { + (role == "assistant" && phase == Some("final_answer")).then_some(true) +} + +pub(super) fn tool_recipient(namespace: Option<&str>, name: &str) -> String { + format!("{}.{}", namespace.unwrap_or("functions"), name) +} + +pub(super) fn record_call_recipient( + call_recipients: &mut HashMap, + call_id: &str, + recipient: &str, +) -> Result<()> { + if call_recipients + .insert(call_id.to_string(), recipient.to_string()) + .is_some() + { + bail!("duplicate Responses tool call_id {call_id}"); + } + Ok(()) +} + +pub(super) fn required_value<'a>(value: &'a JsonValue, field: &str) -> Result<&'a JsonValue> { + value + .get(field) + .with_context(|| format!("Responses item has no {field} field")) +} + +pub(super) fn required_str<'a>(value: &'a JsonValue, field: &str) -> Result<&'a str> { + required_value(value, field)? + .as_str() + .with_context(|| format!("Responses item field {field} is not a string")) +} + +pub(super) fn compact_json(value: &JsonValue) -> Result { + Ok(serde_json::to_string(value)?) +} + +pub(super) fn stable_uuid(seed: &str) -> String { + let mut value = + (u128::from(fnv1a_64(seed)) << 64) | u128::from(fnv1a_64(&format!("{seed}:uuid"))); + value &= !(0xf_u128 << 76); + value |= 0x5_u128 << 76; + value &= !(0x3_u128 << 62); + value |= 0x2_u128 << 62; + let value = format!("{value:032x}"); + format!( + "{}-{}-{}-{}-{}", + &value[..8], + &value[8..12], + &value[12..16], + &value[16..20], + &value[20..] + ) +} + +fn render_content_part(value: &JsonValue) -> Result { + match value.get("type").and_then(JsonValue::as_str) { + Some("input_text" | "output_text" | "summary_text" | "reasoning_text" | "text") => { + Ok(required_str(value, "text")?.to_string()) + } + Some("encrypted_content") => Ok(required_str(value, "encrypted_content")?.to_string()), + _ => compact_json(value), + } +} + +fn fnv1a_64(value: &str) -> u64 { + value.bytes().fold(0xcbf29ce484222325, |hash, byte| { + (hash ^ u64::from(byte)).wrapping_mul(0x100000001b3) + }) +} diff --git a/codex-rs/analytics-materializer/src/synthetic_conversation.rs b/codex-rs/analytics-materializer/src/synthetic_conversation.rs new file mode 100644 index 0000000000..186273a283 --- /dev/null +++ b/codex-rs/analytics-materializer/src/synthetic_conversation.rs @@ -0,0 +1,425 @@ +use crate::synthetic_chat::DEFAULT_RECIPIENT; +use crate::synthetic_chat::SyntheticMessage; +use crate::synthetic_chat::compact_json; +use crate::synthetic_chat::developer_tools_content; +use crate::synthetic_chat::message_channel; +use crate::synthetic_chat::message_end_turn; +use crate::synthetic_chat::push_synthetic_message; +use crate::synthetic_chat::record_call_recipient; +use crate::synthetic_chat::render_content; +use crate::synthetic_chat::required_str; +use crate::synthetic_chat::required_value; +use crate::synthetic_chat::stable_uuid; +use crate::synthetic_chat::text_content; +use crate::synthetic_chat::tool_recipient; +use anyhow::Context; +use anyhow::Result; +use anyhow::bail; +use serde_json::Value as JsonValue; +use serde_json::json; +use std::collections::HashMap; + +pub(super) const SYNTHETIC_CONVERSATION_SOURCE: &str = "synthetic_local_responses_request"; + +pub(super) fn synthetic_conversation( + session_id: &str, + thread_id: &str, + context_window_id: &str, + request_json: &JsonValue, + full_input: &[JsonValue], +) -> Result { + let conversation_seed = format!("{session_id}:{thread_id}:{context_window_id}"); + let mut messages = Vec::new(); + let mut call_recipients = HashMap::new(); + + if let Some(tools) = request_json + .get("tools") + .and_then(JsonValue::as_array) + .filter(|tools| !tools.is_empty()) + { + push_synthetic_message( + &mut messages, + &conversation_seed, + SyntheticMessage { + role: "developer".to_string(), + author_name: None, + content: developer_tools_content(tools), + recipient: DEFAULT_RECIPIENT.to_string(), + channel: None, + end_turn: None, + }, + ); + } + if let Some(instructions) = request_json + .get("instructions") + .and_then(JsonValue::as_str) + .filter(|instructions| !instructions.is_empty()) + { + push_synthetic_message( + &mut messages, + &conversation_seed, + SyntheticMessage { + role: "developer".to_string(), + author_name: None, + content: text_content(instructions.to_string()), + recipient: DEFAULT_RECIPIENT.to_string(), + channel: None, + end_turn: None, + }, + ); + } + for item in full_input { + append_response_item_messages( + &mut messages, + &mut call_recipients, + &conversation_seed, + item, + )?; + } + + Ok(json!({ + "id": stable_uuid(&conversation_seed), + "messages": messages, + "create_time": null, + "update_time": null, + "metadata": { + "local_analytics": { + "conversation_source": SYNTHETIC_CONVERSATION_SOURCE, + "context_window_id": context_window_id, + "is_synthetic": true + } + } + })) +} + +fn append_response_item_messages( + messages: &mut Vec, + call_recipients: &mut HashMap, + conversation_seed: &str, + item: &JsonValue, +) -> Result<()> { + let item_type = required_str(item, "type")?; + match item_type { + "message" => append_message_item(messages, conversation_seed, item), + "agent_message" => append_agent_message(messages, conversation_seed, item), + "reasoning" => append_reasoning_item(messages, conversation_seed, item), + "function_call" => append_function_call(messages, call_recipients, conversation_seed, item), + "function_call_output" => { + append_tool_output(messages, call_recipients, conversation_seed, item, "output") + } + "custom_tool_call" => { + append_custom_tool_call(messages, call_recipients, conversation_seed, item) + } + "custom_tool_call_output" => { + append_tool_output(messages, call_recipients, conversation_seed, item, "output") + } + "tool_search_call" => { + append_tool_search_call(messages, call_recipients, conversation_seed, item) + } + "tool_search_output" => append_tool_search_output(messages, conversation_seed, item), + "web_search_call" => append_web_search_call(messages, conversation_seed, item), + "local_shell_call" => { + append_raw_assistant_tool_call(messages, conversation_seed, item, "container.exec") + } + "image_generation_call" => { + append_raw_assistant_tool_call(messages, conversation_seed, item, "image_generation") + } + "compaction" => append_compaction(messages, conversation_seed, item), + "compaction_trigger" => append_compaction_trigger(messages, conversation_seed), + "context_compaction" => append_context_compaction(messages, conversation_seed, item), + _ => bail!("unsupported Responses input item type {item_type}"), + } +} + +fn append_message_item( + messages: &mut Vec, + conversation_seed: &str, + item: &JsonValue, +) -> Result<()> { + let role = required_str(item, "role")?; + let phase = item.get("phase").and_then(JsonValue::as_str); + push_synthetic_message( + messages, + conversation_seed, + SyntheticMessage { + role: role.to_string(), + author_name: None, + content: text_content(render_content(required_value(item, "content")?)?), + recipient: DEFAULT_RECIPIENT.to_string(), + channel: message_channel(role, phase).map(str::to_string), + end_turn: message_end_turn(role, phase), + }, + ); + Ok(()) +} + +fn append_agent_message( + messages: &mut Vec, + conversation_seed: &str, + item: &JsonValue, +) -> Result<()> { + push_synthetic_message( + messages, + conversation_seed, + SyntheticMessage { + role: "assistant".to_string(), + author_name: Some(required_str(item, "author")?.to_string()), + content: text_content(render_content(required_value(item, "content")?)?), + recipient: required_str(item, "recipient")?.to_string(), + channel: Some("commentary".to_string()), + end_turn: None, + }, + ); + Ok(()) +} + +fn append_reasoning_item( + messages: &mut Vec, + conversation_seed: &str, + item: &JsonValue, +) -> Result<()> { + let summary = render_content(required_value(item, "summary")?)?; + let content = if summary.is_empty() { + item.get("encrypted_content") + .and_then(JsonValue::as_str) + .unwrap_or_default() + .to_string() + } else { + summary + }; + push_synthetic_message( + messages, + conversation_seed, + SyntheticMessage { + role: "assistant".to_string(), + author_name: None, + content: text_content(content), + recipient: DEFAULT_RECIPIENT.to_string(), + channel: Some("analysis".to_string()), + end_turn: None, + }, + ); + Ok(()) +} + +fn append_function_call( + messages: &mut Vec, + call_recipients: &mut HashMap, + conversation_seed: &str, + item: &JsonValue, +) -> Result<()> { + let recipient = tool_recipient( + item.get("namespace").and_then(JsonValue::as_str), + required_str(item, "name")?, + ); + record_call_recipient(call_recipients, required_str(item, "call_id")?, &recipient)?; + append_assistant_tool_call( + messages, + conversation_seed, + recipient, + required_str(item, "arguments")?.to_string(), + ); + Ok(()) +} + +fn append_custom_tool_call( + messages: &mut Vec, + call_recipients: &mut HashMap, + conversation_seed: &str, + item: &JsonValue, +) -> Result<()> { + let recipient = tool_recipient(None, required_str(item, "name")?); + record_call_recipient(call_recipients, required_str(item, "call_id")?, &recipient)?; + append_assistant_tool_call( + messages, + conversation_seed, + recipient, + required_str(item, "input")?.to_string(), + ); + Ok(()) +} + +fn append_tool_output( + messages: &mut Vec, + call_recipients: &HashMap, + conversation_seed: &str, + item: &JsonValue, + output_field: &str, +) -> Result<()> { + let call_id = required_str(item, "call_id")?; + let recipient = call_recipients + .get(call_id) + .cloned() + .or_else(|| { + item.get("name") + .and_then(JsonValue::as_str) + .map(|name| tool_recipient(None, name)) + }) + .unwrap_or_else(|| tool_recipient(None, call_id)); + push_synthetic_message( + messages, + conversation_seed, + SyntheticMessage { + role: "tool".to_string(), + author_name: Some(recipient), + content: text_content(render_content(required_value(item, output_field)?)?), + recipient: DEFAULT_RECIPIENT.to_string(), + channel: Some("commentary".to_string()), + end_turn: None, + }, + ); + Ok(()) +} + +fn append_tool_search_call( + messages: &mut Vec, + call_recipients: &mut HashMap, + conversation_seed: &str, + item: &JsonValue, +) -> Result<()> { + let recipient = "tool_search.tool_search_tool".to_string(); + if let Some(call_id) = item.get("call_id").and_then(JsonValue::as_str) { + record_call_recipient(call_recipients, call_id, &recipient)?; + } + append_assistant_tool_call( + messages, + conversation_seed, + recipient, + compact_json(required_value(item, "arguments")?)?, + ); + Ok(()) +} + +fn append_tool_search_output( + messages: &mut Vec, + conversation_seed: &str, + item: &JsonValue, +) -> Result<()> { + let tools = item + .get("tools") + .and_then(JsonValue::as_array) + .context("tool_search_output has no tools array")?; + push_synthetic_message( + messages, + conversation_seed, + SyntheticMessage { + role: "developer".to_string(), + author_name: None, + content: developer_tools_content(tools), + recipient: DEFAULT_RECIPIENT.to_string(), + channel: None, + end_turn: None, + }, + ); + Ok(()) +} + +fn append_web_search_call( + messages: &mut Vec, + conversation_seed: &str, + item: &JsonValue, +) -> Result<()> { + let content = item + .get("action") + .map(compact_json) + .transpose()? + .unwrap_or_default(); + append_assistant_tool_call(messages, conversation_seed, "web".to_string(), content); + Ok(()) +} + +fn append_raw_assistant_tool_call( + messages: &mut Vec, + conversation_seed: &str, + item: &JsonValue, + recipient: &str, +) -> Result<()> { + append_assistant_tool_call( + messages, + conversation_seed, + recipient.to_string(), + compact_json(item)?, + ); + Ok(()) +} + +fn append_compaction( + messages: &mut Vec, + conversation_seed: &str, + item: &JsonValue, +) -> Result<()> { + append_summary_message( + messages, + conversation_seed, + "assistant", + required_str(item, "encrypted_content")?.to_string(), + ); + Ok(()) +} + +fn append_compaction_trigger(messages: &mut Vec, conversation_seed: &str) -> Result<()> { + append_summary_message( + messages, + conversation_seed, + "system", + "Context compaction triggered. Summarize the current context.".to_string(), + ); + Ok(()) +} + +fn append_context_compaction( + messages: &mut Vec, + conversation_seed: &str, + item: &JsonValue, +) -> Result<()> { + append_summary_message( + messages, + conversation_seed, + "assistant", + item.get("encrypted_content") + .and_then(JsonValue::as_str) + .unwrap_or_default() + .to_string(), + ); + Ok(()) +} + +fn append_summary_message( + messages: &mut Vec, + conversation_seed: &str, + role: &str, + content: String, +) { + push_synthetic_message( + messages, + conversation_seed, + SyntheticMessage { + role: role.to_string(), + author_name: None, + content: text_content(content), + recipient: DEFAULT_RECIPIENT.to_string(), + channel: Some("summary".to_string()), + end_turn: None, + }, + ); +} + +fn append_assistant_tool_call( + messages: &mut Vec, + conversation_seed: &str, + recipient: String, + content: String, +) { + push_synthetic_message( + messages, + conversation_seed, + SyntheticMessage { + role: "assistant".to_string(), + author_name: None, + content: text_content(content), + recipient, + channel: Some("commentary".to_string()), + end_turn: Some(false), + }, + ); +} diff --git a/codex-rs/analytics/src/client.rs b/codex-rs/analytics/src/client.rs index e23f8c8d14..b56c4d1162 100644 --- a/codex-rs/analytics/src/client.rs +++ b/codex-rs/analytics/src/client.rs @@ -180,6 +180,7 @@ impl AnalyticsEventsClient { session_id: String, thread_id: String, turn_id: String, + context_window_id: String, ) -> LocalResponsesApiCallCapture { let Some(queue) = self.queue.as_ref() else { return LocalResponsesApiCallCapture::disabled(); @@ -187,7 +188,13 @@ impl AnalyticsEventsClient { if queue.local_sink.is_none() { return LocalResponsesApiCallCapture::disabled(); } - LocalResponsesApiCallCapture::enabled(queue.sender.clone(), session_id, thread_id, turn_id) + LocalResponsesApiCallCapture::enabled( + queue.sender.clone(), + session_id, + thread_id, + turn_id, + context_window_id, + ) } #[cfg(test)] diff --git a/codex-rs/analytics/src/client_tests.rs b/codex-rs/analytics/src/client_tests.rs index c026b61e68..448c0bea2f 100644 --- a/codex-rs/analytics/src/client_tests.rs +++ b/codex-rs/analytics/src/client_tests.rs @@ -288,6 +288,7 @@ async fn local_responses_capture_writes_reduced_terminal_record() { "session-1".to_string(), "thread-1".to_string(), "turn-1".to_string(), + "thread-1:0".to_string(), ); let attempt = capture.start_attempt(LocalResponsesApiTransport::Http, &json!({"model": "gpt"})); attempt.record_completed("response-1", Some("request-1"), &None, &[]); @@ -300,6 +301,7 @@ async fn local_responses_capture_writes_reduced_terminal_record() { ); assert_eq!(records[0].session_id.as_deref(), Some("session-1")); assert_eq!(records[0].payload["status"], "completed"); + assert_eq!(records[0].payload["context_window_id"], "thread-1:0"); assert_eq!(records[0].payload["request_json"], json!({"model": "gpt"})); } diff --git a/codex-rs/analytics/src/local_responses.rs b/codex-rs/analytics/src/local_responses.rs index 5f5fe6d2cc..77012dd67e 100644 --- a/codex-rs/analytics/src/local_responses.rs +++ b/codex-rs/analytics/src/local_responses.rs @@ -38,6 +38,7 @@ pub struct LocalResponsesApiCallCapture { session_id: String, thread_id: String, turn_id: String, + context_window_id: String, } /// One locally captured Responses API attempt. @@ -53,6 +54,7 @@ pub(crate) struct LocalResponsesApiCallStartedFact { session_id: String, thread_id: String, turn_id: String, + context_window_id: String, transport: LocalResponsesApiTransport, request_started_at_epoch_millis: u64, request_json: JsonValue, @@ -92,6 +94,7 @@ pub(crate) struct LocalResponsesApiCallReducer { #[derive(Serialize)] struct LocalResponsesApiCallPayload { responses_call_id: String, + context_window_id: String, transport: LocalResponsesApiTransport, status: LocalResponsesApiCallStatus, request_started_at_epoch_millis: u64, @@ -120,6 +123,7 @@ impl LocalResponsesApiCallCapture { session_id: String::new(), thread_id: String::new(), turn_id: String::new(), + context_window_id: String::new(), } } @@ -128,12 +132,14 @@ impl LocalResponsesApiCallCapture { session_id: String, thread_id: String, turn_id: String, + context_window_id: String, ) -> Self { Self { sender: Some(sender), session_id, thread_id, turn_id, + context_window_id, } } @@ -157,6 +163,7 @@ impl LocalResponsesApiCallCapture { session_id: self.session_id.clone(), thread_id: self.thread_id.clone(), turn_id: self.turn_id.clone(), + context_window_id: self.context_window_id.clone(), transport, request_started_at_epoch_millis: now_unix_millis(), request_json, @@ -346,6 +353,7 @@ impl LocalResponsesApiCallTerminalFact { }; let payload = LocalResponsesApiCallPayload { responses_call_id: started.responses_call_id, + context_window_id: started.context_window_id, transport: started.transport, status, request_started_at_epoch_millis: started.request_started_at_epoch_millis, diff --git a/codex-rs/analytics/src/local_responses_tests.rs b/codex-rs/analytics/src/local_responses_tests.rs index 5901cca648..89d9390e6b 100644 --- a/codex-rs/analytics/src/local_responses_tests.rs +++ b/codex-rs/analytics/src/local_responses_tests.rs @@ -14,6 +14,7 @@ fn reducer_emits_one_responses_record_after_terminal_fact() { session_id: "session-1".to_string(), thread_id: "thread-1".to_string(), turn_id: "turn-1".to_string(), + context_window_id: "thread-1:0".to_string(), transport: LocalResponsesApiTransport::Http, request_started_at_epoch_millis: 10, request_json: json!({"model": "gpt-test"}), @@ -39,6 +40,7 @@ fn reducer_emits_one_responses_record_after_terminal_fact() { record.payload, json!({ "responses_call_id": "call-1", + "context_window_id": "thread-1:0", "transport": "http", "status": "completed", "request_started_at_epoch_millis": 10, diff --git a/codex-rs/core/src/session/turn.rs b/codex-rs/core/src/session/turn.rs index adba06bc82..06178decf4 100644 --- a/codex-rs/core/src/session/turn.rs +++ b/codex-rs/core/src/session/turn.rs @@ -231,6 +231,7 @@ pub(crate) async fn run_turn( Arc::clone(&turn_diff_tracker), &mut client_session, turn_metadata_header.as_deref(), + &window_id, sampling_request_input.clone(), cancellation_token.child_token(), ) @@ -978,6 +979,7 @@ async fn run_sampling_request( turn_diff_tracker: SharedTurnDiffTracker, client_session: &mut ModelClientSession, turn_metadata_header: Option<&str>, + context_window_id: &str, input: Vec, cancellation_token: CancellationToken, ) -> CodexResult { @@ -1021,6 +1023,7 @@ async fn run_sampling_request( Arc::clone(&turn_store), client_session, turn_metadata_header, + context_window_id, Arc::clone(&turn_diff_tracker), &prompt, cancellation_token.child_token(), @@ -1754,6 +1757,7 @@ async fn try_run_sampling_request( turn_store: Arc, client_session: &mut ModelClientSession, turn_metadata_header: Option<&str>, + context_window_id: &str, turn_diff_tracker: SharedTurnDiffTracker, prompt: &Prompt, cancellation_token: CancellationToken, @@ -1778,6 +1782,7 @@ async fn try_run_sampling_request( sess.session_id().to_string(), sess.thread_id().to_string(), turn_context.sub_id.clone(), + context_window_id.to_string(), ); let sampling_timing_guard = turn_context.turn_timing_state.begin_sampling(); let mut stream = client_session