openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-09-04 15:08:45 +00:00
Code Issues Packages Projects Releases Wiki Activity

context windows

Browse Source
This commit is contained in:
Owen Lin
2026-06-09 11:21:07 -07:00
parent 6066564b09
commit 8a3c97fd92
10 changed files with 1283 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files

351
codex-rs/analytics-materializer/src/context_windows.rs Normal file
View File

@@ -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<String>,
responses_call_id: String,
context_window_id: String,
status: String,
request_started_at_epoch_millis: i64,
response_id: Option<String>,
request_json: String,
response_json: Option<String>,
}
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<JsonValue>,
output_items: Vec<JsonValue>,
}
struct ReconstructedPrompt {
request_json: JsonValue,
full_input: Vec<JsonValue>,
}
struct ContextWindowDraft {
session_id: String,
thread_id: String,
context_window_id: String,
context_window_ordinal: i64,
first_turn_id: Option<String>,
last_turn_id: Option<String>,
first_responses_call_id: String,
last_responses_call_id: String,
opened_at_epoch_millis: i64,
closed_at_epoch_millis: Option<i64>,
close_reason: Option<String>,
message_count: i64,
conversation_json: String,
}
fn load_context_window_calls(connection: &Connection) -> Result<Vec<ResponsesCallRow>> {
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::<duckdb::Result<Vec<_>>>()?)
}
fn reduce_context_windows(calls: Vec<ResponsesCallRow>) -> Result<Vec<ContextWindowDraft>> {
let mut windows = Vec::new();
let mut current_key: Option<ContextWindowKey> = 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<ResponsesCallRow>) -> Result<ContextWindowDraft> {
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::<String, CompletedCall>::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<Vec<JsonValue>> {
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<Vec<JsonValue>> {
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<usize> = 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<String>) -> DuckValue {
value.map_or(DuckValue::Null, DuckValue::Text)
}
fn integer(value: i64) -> DuckValue {
DuckValue::BigInt(value)
}
fn optional_integer(value: Option<i64>) -> DuckValue {
value.map_or(DuckValue::Null, DuckValue::BigInt)
}

10
codex-rs/analytics-materializer/src/lib.rs
View File

@@ -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<Path>, output: impl AsRef<Path>
create_raw_records_table(&connection)?;
insert_local_records(&connection, input)?;
connection.execute_batch(MATERIALIZE_SQL)?;
context_windows::materialize_context_windows(&connection)?;
Ok(())
}

327
codex-rs/analytics-materializer/src/lib_tests.rs
View File

@@ -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::<serde_json::Value>(&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<i64>>(2)?,
row.get::<_, Option<String>>(3)?,
))
})
.expect("query should run")
.collect::<duckdb::Result<Vec<_>>>()
.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<String> {
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<String> {
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()

166
codex-rs/analytics-materializer/src/synthetic_chat.rs Normal file
View File

@@ -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<String>,
pub(super) content: JsonValue,
pub(super) recipient: String,
pub(super) channel: Option<String>,
pub(super) end_turn: Option<bool>,
}
pub(super) fn push_synthetic_message(
messages: &mut Vec<JsonValue>,
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<String> {
match value {
JsonValue::Array(parts) => parts
.iter()
.map(render_content_part)
.collect::<Result<Vec<_>>>()
.map(|parts| {
parts
.into_iter()
.filter(|part| !part.is_empty())
.collect::<Vec<_>>()
.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<bool> {
(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<String, String>,
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<String> {
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<String> {
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)
})
}

425
codex-rs/analytics-materializer/src/synthetic_conversation.rs Normal file
View File

@@ -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<JsonValue> {
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<JsonValue>,
call_recipients: &mut HashMap<String, String>,
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<JsonValue>,
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<JsonValue>,
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<JsonValue>,
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<JsonValue>,
call_recipients: &mut HashMap<String, String>,
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<JsonValue>,
call_recipients: &mut HashMap<String, String>,
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<JsonValue>,
call_recipients: &HashMap<String, String>,
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<JsonValue>,
call_recipients: &mut HashMap<String, String>,
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<JsonValue>,
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<JsonValue>,
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<JsonValue>,
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<JsonValue>,
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<JsonValue>, 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<JsonValue>,
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<JsonValue>,
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<JsonValue>,
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),
},
);
}

9
codex-rs/analytics/src/client.rs
View File

@@ -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)]

2
codex-rs/analytics/src/client_tests.rs
View File

@@ -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"}));
}

8
codex-rs/analytics/src/local_responses.rs
View File

@@ -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,

2
codex-rs/analytics/src/local_responses_tests.rs
View File

@@ -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,

5
codex-rs/core/src/session/turn.rs
View File

@@ -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<ResponseItem>,
cancellation_token: CancellationToken,
) -> CodexResult<SamplingRequestResult> {
@@ -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<codex_extension_api::ExtensionData>,
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