Preserve and expose thread item lifecycle timestamps (#47114)

## What changed

- Persist producer-recorded item start and completion times in thread history, preserving existing non-null values on updates.
- Return nullable `startedAtMs` and `completedAtMs` Unix millisecond timestamps from `thread/items/list`, and update the generated schemas and TypeScript and Python types.
- Leave timestamps unknown for existing rows and older writers.

## Testing

Add coverage for exact timestamp persistence, protocol round trips with present and missing timestamps, paginated item responses, and compatibility with older database writers.

GitOrigin-RevId: 894853481085782ff54db1704799e41446d15e51
This commit is contained in:
chiam-oai
2026-09-21 22:23:26 +00:00
committed by copyberry
parent 23ae65bb06
commit 772abc9425
20 changed files with 231 additions and 30 deletions
@@ -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"
@@ -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"
@@ -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"
@@ -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, };
@@ -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::<ThreadItemsListResponse>(value).expect("deserialize response"),
response
);
}
let params_without_turn = ThreadItemsListParams {
thread_id: "thr_123".to_string(),
@@ -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<i64>,
/// Unix timestamp (milliseconds) when the item completed, if recorded by the producer.
#[ts(type = "number | null")]
pub completed_at_ms: Option<i64>,
}
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
@@ -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::<Result<Vec<_>, _>>()?;
@@ -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,
+7
View File
@@ -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<i64>, Option<i64>)>(
"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)
@@ -0,0 +1,2 @@
ALTER TABLE thread_items ADD COLUMN started_at_ms INTEGER;
ALTER TABLE thread_items ADD COLUMN completed_at_ms INTEGER;
@@ -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)
@@ -69,12 +69,16 @@ pub(super) struct StoredSummaryColumns {
summary_first_user_rollout_ordinal: Option<i64>,
summary_first_user_updated_at_ordinal: Option<i64>,
summary_first_user_created_at_ms: Option<i64>,
summary_first_user_started_at_ms: Option<i64>,
summary_first_user_completed_at_ms: Option<i64>,
summary_first_user_item_json: Option<String>,
summary_final_agent_turn_id: Option<String>,
summary_final_agent_item_id: Option<String>,
summary_final_agent_rollout_ordinal: Option<i64>,
summary_final_agent_updated_at_ordinal: Option<i64>,
summary_final_agent_created_at_ms: Option<i64>,
summary_final_agent_started_at_ms: Option<i64>,
summary_final_agent_completed_at_ms: Option<i64>,
summary_final_agent_item_json: Option<String>,
}
@@ -84,6 +88,8 @@ struct StoredSummaryItemColumns {
rollout_ordinal: Option<i64>,
updated_at_ordinal: Option<i64>,
created_at_ms: Option<i64>,
started_at_ms: Option<i64>,
completed_at_ms: Option<i64>,
item_json: Option<String>,
}
@@ -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<StoredT
item_id: row.try_get("item_id")?,
updated_at_ordinal,
created_at_ms: row.try_get("created_at_ms")?,
started_at_ms: row.try_get("started_at_ms")?,
completed_at_ms: row.try_get("completed_at_ms")?,
item_json: row.try_get::<String, _>("item_json")?.into_bytes(),
})
}
@@ -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(),
}
}
@@ -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::<Sqlite>::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::<Sqlite>::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 =
"#,
@@ -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<i64>, Option<i64>)>(
"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");
+4
View File
@@ -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<i64>,
/// Exact lifecycle completion time in Unix milliseconds, when recorded by the producer.
pub completed_at_ms: Option<i64>,
/// Serialized app-server ThreadItem snapshot.
pub item_json: Vec<u8>,
}
+2
View File
@@ -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(),
)
@@ -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.")]