diff --git a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json index 772a8a0a96..23b6ff042c 100644 --- a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json +++ b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json @@ -22338,9 +22338,25 @@ }, "ThreadItemEntry": { "properties": { + "completedAtMs": { + "description": "Unix timestamp (milliseconds) when the item completed, if recorded by the producer.", + "format": "int64", + "type": [ + "integer", + "null" + ] + }, "item": { "$ref": "#/definitions/v2/ThreadItem" }, + "startedAtMs": { + "description": "Unix timestamp (milliseconds) when the item started, if recorded by the producer.", + "format": "int64", + "type": [ + "integer", + "null" + ] + }, "turnId": { "description": "Turn containing this item.", "type": "string" diff --git a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json index ec8196f085..c1f737a88e 100644 --- a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json +++ b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json @@ -20052,9 +20052,25 @@ }, "ThreadItemEntry": { "properties": { + "completedAtMs": { + "description": "Unix timestamp (milliseconds) when the item completed, if recorded by the producer.", + "format": "int64", + "type": [ + "integer", + "null" + ] + }, "item": { "$ref": "#/definitions/ThreadItem" }, + "startedAtMs": { + "description": "Unix timestamp (milliseconds) when the item started, if recorded by the producer.", + "format": "int64", + "type": [ + "integer", + "null" + ] + }, "turnId": { "description": "Turn containing this item.", "type": "string" diff --git a/codex-rs/app-server-protocol/schema/json/v2/ThreadItemsListResponse.json b/codex-rs/app-server-protocol/schema/json/v2/ThreadItemsListResponse.json index 1389f19551..69549fe455 100644 --- a/codex-rs/app-server-protocol/schema/json/v2/ThreadItemsListResponse.json +++ b/codex-rs/app-server-protocol/schema/json/v2/ThreadItemsListResponse.json @@ -1631,9 +1631,25 @@ }, "ThreadItemEntry": { "properties": { + "completedAtMs": { + "description": "Unix timestamp (milliseconds) when the item completed, if recorded by the producer.", + "format": "int64", + "type": [ + "integer", + "null" + ] + }, "item": { "$ref": "#/definitions/ThreadItem" }, + "startedAtMs": { + "description": "Unix timestamp (milliseconds) when the item started, if recorded by the producer.", + "format": "int64", + "type": [ + "integer", + "null" + ] + }, "turnId": { "description": "Turn containing this item.", "type": "string" diff --git a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst index b01dad2e54..9bcc74d804 100644 Binary files a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst and b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst differ diff --git a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst index 3d990a9b7f..172ace1368 100644 Binary files a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst and b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst differ diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/ThreadItemEntry.ts b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadItemEntry.ts index c59564f265..5815a3afcf 100644 --- a/codex-rs/app-server-protocol/schema/typescript/v2/ThreadItemEntry.ts +++ b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadItemEntry.ts @@ -7,4 +7,12 @@ export type ThreadItemEntry = { /** * Turn containing this item. */ -turnId: string, item: ThreadItem, }; +turnId: string, item: ThreadItem, +/** + * Unix timestamp (milliseconds) when the item started, if recorded by the producer. + */ +startedAtMs: number | null, +/** + * Unix timestamp (milliseconds) when the item completed, if recorded by the producer. + */ +completedAtMs: number | null, }; diff --git a/codex-rs/app-server-protocol/src/protocol/v2/tests.rs b/codex-rs/app-server-protocol/src/protocol/v2/tests.rs index d0f9401d79..4cbe82ab4b 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2/tests.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2/tests.rs @@ -415,28 +415,43 @@ fn thread_items_list_round_trips() { "sortDirection": "asc", }) ); - let response = ThreadItemsListResponse { - data: vec![ThreadItemEntry { - turn_id: "turn_456".to_string(), - item: ThreadItem::ContextCompaction { - id: "item_1".to_string(), - }, - }], - next_cursor: None, - backwards_cursor: Some("cursor_0".to_string()), - }; - - assert_eq!( - serde_json::to_value(&response).expect("serialize response"), - json!({ - "data": [{ - "turnId": "turn_456", - "item": {"type": "contextCompaction", "id": "item_1"}, + for (started_at_ms, completed_at_ms) in [ + (Some(1_789_855_978_123), Some(1_789_855_979_456)), + (Some(1_789_855_978_123), None), + (Some(1_789_855_978_123), Some(1_789_855_978_123)), + (None, None), + ] { + let response = ThreadItemsListResponse { + data: vec![ThreadItemEntry { + turn_id: "turn_456".to_string(), + item: ThreadItem::ContextCompaction { + id: "item_1".to_string(), + }, + started_at_ms, + completed_at_ms, }], - "nextCursor": null, - "backwardsCursor": "cursor_0", - }) - ); + next_cursor: None, + backwards_cursor: Some("cursor_0".to_string()), + }; + let value = serde_json::to_value(&response).expect("serialize response"); + assert_eq!( + value, + json!({ + "data": [{ + "turnId": "turn_456", + "item": {"type": "contextCompaction", "id": "item_1"}, + "startedAtMs": started_at_ms, + "completedAtMs": completed_at_ms, + }], + "nextCursor": null, + "backwardsCursor": "cursor_0", + }) + ); + assert_eq!( + serde_json::from_value::(value).expect("deserialize response"), + response + ); + } let params_without_turn = ThreadItemsListParams { thread_id: "thr_123".to_string(), diff --git a/codex-rs/app-server-protocol/src/protocol/v2/thread.rs b/codex-rs/app-server-protocol/src/protocol/v2/thread.rs index 09732cfb43..a1376843a0 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2/thread.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2/thread.rs @@ -1759,6 +1759,12 @@ pub struct ThreadItemEntry { /// Turn containing this item. pub turn_id: String, pub item: ThreadItem, + /// Unix timestamp (milliseconds) when the item started, if recorded by the producer. + #[ts(type = "number | null")] + pub started_at_ms: Option, + /// Unix timestamp (milliseconds) when the item completed, if recorded by the producer. + #[ts(type = "number | null")] + pub completed_at_ms: Option, } #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] diff --git a/codex-rs/app-server/src/request_processors/thread_processor.rs b/codex-rs/app-server/src/request_processors/thread_processor.rs index b7affb2367..57c10fe833 100644 --- a/codex-rs/app-server/src/request_processors/thread_processor.rs +++ b/codex-rs/app-server/src/request_processors/thread_processor.rs @@ -3447,8 +3447,15 @@ impl ThreadRequestProcessor { .into_iter() .map(|stored_item| { let turn_id = stored_item.turn_id.clone(); + let started_at_ms = stored_item.started_at_ms; + let completed_at_ms = stored_item.completed_at_ms; let item = deserialize_stored_thread_item(stored_item)?; - Ok(ThreadItemEntry { turn_id, item }) + Ok(ThreadItemEntry { + turn_id, + item, + started_at_ms, + completed_at_ms, + }) }) .collect::, _>>()?; diff --git a/codex-rs/app-server/tests/suite/v2/thread_read.rs b/codex-rs/app-server/tests/suite/v2/thread_read.rs index 6897588e1c..d7d326f910 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_read.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_read.rs @@ -24,6 +24,7 @@ use codex_app_server_protocol::ThreadForkParams; use codex_app_server_protocol::ThreadForkResponse; use codex_app_server_protocol::ThreadHistoryMode; use codex_app_server_protocol::ThreadItem; +use codex_app_server_protocol::ThreadItemEntry; use codex_app_server_protocol::ThreadItemsListParams; use codex_app_server_protocol::ThreadItemsListResponse; use codex_app_server_protocol::ThreadListParams; @@ -2008,9 +2009,19 @@ async fn paginated_history_lists_and_legacy_reads_use_projected_turns_and_items( SortDirection::Asc, ) .await?; - assert_eq!(first_items_page.data.len(), 1); - assert_eq!(first_items_page.data[0].turn_id, "turn-1"); - assert_eq!(first_items_page.data[0].item.id(), "user-1"); + assert_eq!( + first_items_page.data, + vec![ThreadItemEntry { + turn_id: "turn-1".to_string(), + item: ThreadItem::UserMessage { + id: "user-1".to_string(), + client_id: None, + content: Vec::new(), + }, + started_at_ms: Some(0), + completed_at_ms: Some(1), + }] + ); let second_items_page = read_items_page( &mut mcp, thread_id, diff --git a/codex-rs/state/src/migrations_tests.rs b/codex-rs/state/src/migrations_tests.rs index 95cd6818c2..3a0b57da1f 100644 --- a/codex-rs/state/src/migrations_tests.rs +++ b/codex-rs/state/src/migrations_tests.rs @@ -546,6 +546,13 @@ async fn realtime_items_preserve_older_thread_history_writers() { ("older-writer-item".to_string(), "turn-1".to_string()), ] ); + let lifecycle_timestamps = sqlx::query_as::<_, (Option, Option)>( + "SELECT started_at_ms, completed_at_ms FROM thread_items ORDER BY rollout_ordinal", + ) + .fetch_all(&older_pool) + .await + .expect("old rows and older writers leave lifecycle timestamps unknown"); + assert_eq!(lifecycle_timestamps, vec![(None, None), (None, None)]); sqlx::query("DELETE FROM thread_history_projection_state WHERE thread_id = ?") .bind("thread-1") .execute(&older_pool) diff --git a/codex-rs/state/thread_history_migrations/0007_thread_item_lifecycle_timestamps.sql b/codex-rs/state/thread_history_migrations/0007_thread_item_lifecycle_timestamps.sql new file mode 100644 index 0000000000..6fe293575b --- /dev/null +++ b/codex-rs/state/thread_history_migrations/0007_thread_item_lifecycle_timestamps.sql @@ -0,0 +1,2 @@ +ALTER TABLE thread_items ADD COLUMN started_at_ms INTEGER; +ALTER TABLE thread_items ADD COLUMN completed_at_ms INTEGER; diff --git a/codex-rs/thread-store/src/local/thread_history.rs b/codex-rs/thread-store/src/local/thread_history.rs index 7b31841842..81364f5142 100644 --- a/codex-rs/thread-store/src/local/thread_history.rs +++ b/codex-rs/thread-store/src/local/thread_history.rs @@ -450,11 +450,15 @@ INSERT INTO thread_items ( rollout_ordinal, updated_at_ordinal, created_at_ms, + started_at_ms, + completed_at_ms, item_type, item_json -) VALUES (?, ?, ?, ?, ?, ?, json_extract(?, '$.type'), ?) +) VALUES (?, ?, ?, ?, ?, ?, ?, ?, json_extract(?, '$.type'), ?) ON CONFLICT(thread_id, turn_id, item_id) DO UPDATE SET updated_at_ordinal = excluded.updated_at_ordinal, + started_at_ms = COALESCE(thread_items.started_at_ms, excluded.started_at_ms), + completed_at_ms = COALESCE(thread_items.completed_at_ms, excluded.completed_at_ms), item_type = excluded.item_type, item_json = excluded.item_json "#, @@ -465,6 +469,8 @@ ON CONFLICT(thread_id, turn_id, item_id) DO UPDATE SET .bind(rollout_ordinal) .bind(rollout_ordinal) .bind(created_at_ms) + .bind(item.started_at_ms) + .bind(item.completed_at_ms) .bind(item_json.as_str()) .bind(item_json) .execute(&mut **transaction) diff --git a/codex-rs/thread-store/src/local/thread_history/read.rs b/codex-rs/thread-store/src/local/thread_history/read.rs index 7dd06c74cd..dad2a26467 100644 --- a/codex-rs/thread-store/src/local/thread_history/read.rs +++ b/codex-rs/thread-store/src/local/thread_history/read.rs @@ -69,12 +69,16 @@ pub(super) struct StoredSummaryColumns { summary_first_user_rollout_ordinal: Option, summary_first_user_updated_at_ordinal: Option, summary_first_user_created_at_ms: Option, + summary_first_user_started_at_ms: Option, + summary_first_user_completed_at_ms: Option, summary_first_user_item_json: Option, summary_final_agent_turn_id: Option, summary_final_agent_item_id: Option, summary_final_agent_rollout_ordinal: Option, summary_final_agent_updated_at_ordinal: Option, summary_final_agent_created_at_ms: Option, + summary_final_agent_started_at_ms: Option, + summary_final_agent_completed_at_ms: Option, summary_final_agent_item_json: Option, } @@ -84,6 +88,8 @@ struct StoredSummaryItemColumns { rollout_ordinal: Option, updated_at_ordinal: Option, created_at_ms: Option, + started_at_ms: Option, + completed_at_ms: Option, item_json: Option, } @@ -223,7 +229,7 @@ async fn load_inherited_summary_items( .transpose()?; let rows = sqlx::query( r#" -SELECT turn_id, item_id, updated_at_ordinal, created_at_ms, item_json +SELECT turn_id, item_id, updated_at_ordinal, created_at_ms, started_at_ms, completed_at_ms, item_json FROM thread_items WHERE thread_id = ? AND turn_id = ? @@ -322,6 +328,8 @@ impl StoredSummaryColumns { rollout_ordinal: self.summary_first_user_rollout_ordinal, updated_at_ordinal: self.summary_first_user_updated_at_ordinal, created_at_ms: self.summary_first_user_created_at_ms, + started_at_ms: self.summary_first_user_started_at_ms, + completed_at_ms: self.summary_first_user_completed_at_ms, item_json: self.summary_first_user_item_json, } .into_stored_item()?, @@ -331,6 +339,8 @@ impl StoredSummaryColumns { rollout_ordinal: self.summary_final_agent_rollout_ordinal, updated_at_ordinal: self.summary_final_agent_updated_at_ordinal, created_at_ms: self.summary_final_agent_created_at_ms, + started_at_ms: self.summary_final_agent_started_at_ms, + completed_at_ms: self.summary_final_agent_completed_at_ms, item_json: self.summary_final_agent_item_json, } .into_stored_item()?, @@ -373,6 +383,8 @@ impl StoredSummaryItemColumns { item_id, updated_at_ordinal: stored_updated_at_ordinal(updated_at_ordinal)?, created_at_ms, + started_at_ms: self.started_at_ms, + completed_at_ms: self.completed_at_ms, item_json: item_json.into_bytes(), }, ))) @@ -401,6 +413,8 @@ fn stored_thread_item(row: sqlx::sqlite::SqliteRow) -> ThreadStoreResult("item_json")?.into_bytes(), }) } diff --git a/codex-rs/thread-store/src/local/thread_history/read_tests.rs b/codex-rs/thread-store/src/local/thread_history/read_tests.rs index 06eaaca703..5d5ffb66ee 100644 --- a/codex-rs/thread-store/src/local/thread_history/read_tests.rs +++ b/codex-rs/thread-store/src/local/thread_history/read_tests.rs @@ -1526,6 +1526,8 @@ fn expected_item(turn_id: &str, item_id: &str, rollout_ordinal: u64) -> StoredTh item_id: item_id.to_string(), updated_at_ordinal: rollout_ordinal, created_at_ms: i64::try_from(rollout_ordinal).expect("fixture ordinal fits i64") * 1_000, + started_at_ms: None, + completed_at_ms: None, item_json: fixture_item(item_id).1.into_bytes(), } } diff --git a/codex-rs/thread-store/src/local/thread_history/segment_paging.rs b/codex-rs/thread-store/src/local/thread_history/segment_paging.rs index 45b087b675..9d11e31505 100644 --- a/codex-rs/thread-store/src/local/thread_history/segment_paging.rs +++ b/codex-rs/thread-store/src/local/thread_history/segment_paging.rs @@ -120,12 +120,16 @@ SELECT first_user.rollout_ordinal AS summary_first_user_rollout_ordinal, first_user.updated_at_ordinal AS summary_first_user_updated_at_ordinal, first_user.created_at_ms AS summary_first_user_created_at_ms, + first_user.started_at_ms AS summary_first_user_started_at_ms, + first_user.completed_at_ms AS summary_first_user_completed_at_ms, first_user.item_json AS summary_first_user_item_json, final_agent.turn_id AS summary_final_agent_turn_id, final_agent.item_id AS summary_final_agent_item_id, final_agent.rollout_ordinal AS summary_final_agent_rollout_ordinal, final_agent.updated_at_ordinal AS summary_final_agent_updated_at_ordinal, final_agent.created_at_ms AS summary_final_agent_created_at_ms, + final_agent.started_at_ms AS summary_final_agent_started_at_ms, + final_agent.completed_at_ms AS summary_final_agent_completed_at_ms, final_agent.item_json AS summary_final_agent_item_json FROM page_turns LEFT JOIN thread_items AS first_user @@ -245,7 +249,7 @@ pub(super) async fn page_item_rows( } let mut query = QueryBuilder::::new( r#" -SELECT turn_id, item_id, rollout_ordinal, updated_at_ordinal, created_at_ms, item_json +SELECT turn_id, item_id, rollout_ordinal, updated_at_ordinal, created_at_ms, started_at_ms, completed_at_ms, item_json FROM thread_items WHERE thread_id = "#, @@ -294,7 +298,7 @@ async fn page_updated_item_rows( )?; let mut query = QueryBuilder::::new( r#" -SELECT turn_id, item_id, updated_at_ordinal AS rollout_ordinal, updated_at_ordinal, created_at_ms, item_json +SELECT turn_id, item_id, updated_at_ordinal AS rollout_ordinal, updated_at_ordinal, created_at_ms, started_at_ms, completed_at_ms, item_json FROM thread_items WHERE thread_id = "#, diff --git a/codex-rs/thread-store/src/local/thread_history_materialization_tests.rs b/codex-rs/thread-store/src/local/thread_history_materialization_tests.rs index 55964eff96..d9034acd56 100644 --- a/codex-rs/thread-store/src/local/thread_history_materialization_tests.rs +++ b/codex-rs/thread-store/src/local/thread_history_materialization_tests.rs @@ -1384,6 +1384,57 @@ async fn subagent_prefix_advances_projection_without_materializing_history() { assert_eq!(projection_state(&pool, thread_id).await.1, 9); } +#[tokio::test] +async fn projection_preserves_exact_lifecycle_timestamps() { + let home = TempDir::new().expect("temp dir"); + let store = projection_store(home.path()).await; + let thread_id = ThreadId::default(); + create_paginated_thread(&store, thread_id).await; + let timestamps = [ + (Some(1_789_855_978_123), Some(1_789_855_979_456)), + (Some(1_789_855_978_123), Some(1_789_855_978_123)), + (None, Some(1_789_855_979_456)), + (None, None), + ]; + let mut items = vec![turn_started("turn-1")]; + for (index, (started_at_ms, completed_at_ms)) in timestamps.iter().enumerate() { + items.push(RolloutItem::EventMsg(EventMsg::ItemCompleted( + ItemCompletedEvent { + thread_id, + turn_id: "turn-1".to_string(), + item: TurnItem::UserMessage(UserMessageItem { + id: format!("user-{index}"), + client_id: None, + content: Vec::new(), + }), + started_at_ms: *started_at_ms, + completed_at_ms: completed_at_ms.unwrap_or_default(), + }, + ))); + } + store + .append_items(AppendThreadItemsParams { thread_id, items }) + .await + .expect("append lifecycle records"); + store + .shutdown_thread(thread_id) + .await + .expect("shutdown thread"); + let pool = codex_state::open_thread_history_db(&codex_state::SqliteConfig::new_for_testing( + home.path().abs(), + )) + .await + .expect("open thread history db"); + let actual = sqlx::query_as::<_, (Option, Option)>( + "SELECT started_at_ms, completed_at_ms FROM thread_items WHERE thread_id = ? ORDER BY rollout_ordinal", + ) + .bind(thread_id.to_string()) + .fetch_all(&pool) + .await + .expect("read persisted lifecycle timestamps"); + assert_eq!(actual, timestamps); +} + #[tokio::test] async fn unexpected_duplicate_item_completion_does_not_poison_projection() { let home = TempDir::new().expect("temp dir"); diff --git a/codex-rs/thread-store/src/types.rs b/codex-rs/thread-store/src/types.rs index 609e35f8d2..d6c0d40bb4 100644 --- a/codex-rs/thread-store/src/types.rs +++ b/codex-rs/thread-store/src/types.rs @@ -510,6 +510,10 @@ pub struct StoredThreadItem { pub updated_at_ordinal: u64, /// Unix timestamp (milliseconds) when this logical item was first projected. pub created_at_ms: i64, + /// Exact lifecycle start time in Unix milliseconds, when recorded by the producer. + pub started_at_ms: Option, + /// Exact lifecycle completion time in Unix milliseconds, when recorded by the producer. + pub completed_at_ms: Option, /// Serialized app-server ThreadItem snapshot. pub item_json: Vec, } diff --git a/codex-rs/tui/src/dynamic_tools.rs b/codex-rs/tui/src/dynamic_tools.rs index c1cea130ba..d9829d5548 100644 --- a/codex-rs/tui/src/dynamic_tools.rs +++ b/codex-rs/tui/src/dynamic_tools.rs @@ -854,6 +854,8 @@ async fn execute_inner( .map(|item| ThreadItemEntry { turn_id: turn.id.clone(), item, + started_at_ms: None, + completed_at_ms: None, }) .collect(), ) diff --git a/sdk/python/src/openai_codex/generated/v2_all.py b/sdk/python/src/openai_codex/generated/v2_all.py index e804eba3f3..278cbabada 100644 --- a/sdk/python/src/openai_codex/generated/v2_all.py +++ b/sdk/python/src/openai_codex/generated/v2_all.py @@ -11194,7 +11194,21 @@ class ThreadItemEntry(BaseModel): model_config = ConfigDict( populate_by_name=True, ) + completed_at_ms: Annotated[ + int | None, + Field( + alias="completedAtMs", + description="Unix timestamp (milliseconds) when the item completed, if recorded by the producer.", + ), + ] = None item: ThreadItem + started_at_ms: Annotated[ + int | None, + Field( + alias="startedAtMs", + description="Unix timestamp (milliseconds) when the item started, if recorded by the producer.", + ), + ] = None turn_id: Annotated[str, Field(alias="turnId", description="Turn containing this item.")]