Expose lifecycle timestamps on app-server thread items (#57172)

## What changed

- Add nullable `startedAtMs` and `completedAtMs` fields to `ThreadItem` in live notifications and turn history. Values are Unix milliseconds when recorded.
- Preserve recorded item times through history reconstruction, pagination, and resume, including command start times recorded during approval.
- Separate the Rust item payload into `ThreadItemPayload` while keeping the JSON payload flat, and update consumers, schemas, and generated TypeScript and Python types.

## Testing

Extend coverage for timestamp serialization, history reconstruction, paginated reads and resume, and persisted Python SDK items.

GitOrigin-RevId: ecbcee95af52db7c807b46a6e9deb30e5fba1f80
This commit is contained in:
Charlie Marsh
2026-10-26 18:13:00 +00:00
committed by copyberry
parent cf29971d93
commit a797dd7c6f
206 changed files with 4691 additions and 3089 deletions
@@ -4,7 +4,7 @@ use super::tools::McpHistory;
use crate::history_cell::ComputerActivityCell;
use crate::history_cell::HistoryCell;
use crate::history_cell::McpToolCallCell;
use codex_app_server_protocol::ThreadItem;
use codex_app_server_protocol::ThreadItemPayload;
use codex_app_server_protocol::Turn;
use std::sync::Arc;
@@ -40,7 +40,7 @@ pub(crate) fn older_computer_group(
let items = adjacent_items(older, newer, turns)?;
let older_items = items
.iter()
.filter(|item| !matches!(item, ThreadItem::Reasoning { .. }))
.filter(|entry| !matches!(&entry.payload, ThreadItemPayload::Reasoning { .. }))
.take(older.call_ids().count())
.cloned()
.collect::<Vec<_>>();
@@ -53,7 +53,7 @@ fn adjacent_items<'a>(
older: &ComputerActivityCell,
newer: &ComputerActivityCell,
turns: &'a [Turn],
) -> Option<&'a [ThreadItem]> {
) -> Option<&'a [codex_app_server_protocol::ThreadItem]> {
let ids = older.call_ids().chain(newer.call_ids()).collect::<Vec<_>>();
adjacent_activity_items(&ids, turns)
}
@@ -62,7 +62,7 @@ fn adjacent_items<'a>(
pub(super) fn adjacent_activity_items<'a>(
ids: &[&str],
turns: &'a [Turn],
) -> Option<&'a [ThreadItem]> {
) -> Option<&'a [codex_app_server_protocol::ThreadItem]> {
let first = *ids.first()?;
let (turn, start) = turns.iter().find_map(|turn| {
turn.items
@@ -71,9 +71,9 @@ pub(super) fn adjacent_activity_items<'a>(
.map(|start| (turn, start))
})?;
let mut matched = 0;
for (offset, item) in turn.items[start..].iter().enumerate() {
for (offset, item) in turn.item_payloads().skip(start).enumerate() {
if item.id() != ids[matched] {
if matches!(item, ThreadItem::Reasoning { .. }) {
if matches!(item, ThreadItemPayload::Reasoning { .. }) {
continue;
}
return None;
@@ -86,10 +86,13 @@ pub(super) fn adjacent_activity_items<'a>(
None
}
pub(super) fn completed_group(items: &[ThreadItem]) -> Option<ComputerActivityCell> {
pub(super) fn completed_group(
items: &[codex_app_server_protocol::ThreadItem],
) -> Option<ComputerActivityCell> {
let mut group = None;
for item in items {
if matches!(item, ThreadItem::Reasoning { .. }) {
for entry in items {
let item = &entry.payload;
if matches!(item, ThreadItemPayload::Reasoning { .. }) {
continue;
}
let call = McpHistory::from_item(item.clone())?;
@@ -5,12 +5,13 @@ use crate::thread_transcript::RawReasoningVisibility;
use crate::thread_transcript::TranscriptCells;
use crate::thread_transcript::thread_items_to_transcript_cells;
use codex_app_server_protocol::ThreadItem;
use codex_app_server_protocol::ThreadItemPayload;
use codex_app_server_protocol::TurnItemsView;
use codex_app_server_protocol::TurnStatus;
use codex_utils_absolute_path::AbsolutePathBuf;
use pretty_assertions::assert_eq;
fn computer(id: &str) -> ThreadItem {
fn computer(id: &str) -> ThreadItemPayload {
serde_json::from_value(serde_json::json!({
"type": "mcpToolCall", "id": id, "server": "cua_repl", "tool": "js",
"status": "completed", "arguments": {"title": format!("Inspect {id}"), "code": "await cua.getState()"},
@@ -20,11 +21,11 @@ fn computer(id: &str) -> ThreadItem {
})).unwrap()
}
fn turn(id: &str, items: Vec<ThreadItem>) -> Turn {
fn turn(id: &str, items: Vec<ThreadItemPayload>) -> Turn {
Turn {
id: id.to_string(),
root_turn_id: None,
items,
items: items.into_iter().map(ThreadItem::untimed).collect(),
items_view: TurnItemsView::Full,
status: TurnStatus::Completed,
error: None,
@@ -34,11 +35,11 @@ fn turn(id: &str, items: Vec<ThreadItem>) -> Turn {
}
}
fn project(items: &[ThreadItem]) -> TranscriptCells {
fn project(items: &[impl std::borrow::Borrow<ThreadItemPayload>]) -> TranscriptCells {
thread_items_to_transcript_cells(
/*thread_id*/ None,
&AbsolutePathBuf::current_dir().unwrap(),
items.to_vec(),
items.iter().map(|entry| entry.borrow().clone()),
RawReasoningVisibility::Hidden,
/*config*/ None,
)
@@ -50,14 +51,14 @@ fn every_page_split_retains_the_same_rich_computer_group() {
"turn",
vec![
computer("first"),
ThreadItem::Reasoning {
ThreadItemPayload::Reasoning {
id: "reasoning-first".to_owned(),
summary: vec!["Inspect the second screen".to_owned()],
content: Vec::new(),
},
computer("second"),
computer("third"),
ThreadItem::Reasoning {
ThreadItemPayload::Reasoning {
id: "reasoning-second".to_owned(),
summary: vec!["Confirm the final screen".to_owned()],
content: Vec::new(),
@@ -131,7 +132,7 @@ fn joins_respect_actual_turns_and_intervening_items() {
assert!(join_computer_groups(&older[0], &newer[0], &turns).is_none());
let grouped = project(&[
first.clone(),
ThreadItem::Sleep(codex_app_server_protocol::SleepItem {
ThreadItemPayload::Sleep(codex_app_server_protocol::SleepItem {
id: "sleep".into(),
duration_ms: 10,
}),
@@ -147,7 +148,7 @@ fn joins_respect_actual_turns_and_intervening_items() {
"turn",
vec![
first,
ThreadItem::Reasoning {
ThreadItemPayload::Reasoning {
id: "reasoning".to_string(),
summary: Vec::new(),
content: Vec::new(),
@@ -162,13 +163,13 @@ fn joins_respect_actual_turns_and_intervening_items() {
joined.transcript_lines(/*width*/ 80),
expected[0].transcript_lines(/*width*/ 80),
);
turns[0].items[1] = ThreadItem::AgentMessage {
turns[0].items[1] = ThreadItem::untimed(ThreadItemPayload::AgentMessage {
id: "message".to_string(),
text: "Visible boundary".to_string(),
phase: None,
memory_citation: None,
delivery: None,
questions: None,
};
});
assert!(join_computer_groups(&older[0], &newer[0], &turns).is_none());
}
@@ -3,7 +3,7 @@
use super::tools::CommandHistory;
use crate::exec_cell::ExecCell;
use crate::history_cell::HistoryCell;
use codex_app_server_protocol::ThreadItem;
use codex_app_server_protocol::ThreadItemPayload;
use codex_app_server_protocol::Turn;
use std::sync::Arc;
@@ -33,7 +33,7 @@ pub(crate) fn older_exploration_group(
let items = adjacent_items(older, newer, turns)?;
let older_items = items
.iter()
.filter(|item| !matches!(item, ThreadItem::Reasoning { .. }))
.filter(|entry| !matches!(&entry.payload, ThreadItemPayload::Reasoning { .. }))
.take(older.group.calls.len())
.cloned()
.collect::<Vec<_>>();
@@ -46,7 +46,7 @@ fn adjacent_items<'a>(
older: &ExecCell,
newer: &ExecCell,
turns: &'a [Turn],
) -> Option<&'a [ThreadItem]> {
) -> Option<&'a [codex_app_server_protocol::ThreadItem]> {
if !older.is_exploring_cell() || older.should_flush() || !newer.is_exploring_cell() {
return None;
}
@@ -58,10 +58,11 @@ fn adjacent_items<'a>(
super::computer_groups::adjacent_activity_items(&ids, turns)
}
pub(super) fn completed_group(items: &[ThreadItem]) -> Option<ExecCell> {
pub(super) fn completed_group(items: &[codex_app_server_protocol::ThreadItem]) -> Option<ExecCell> {
let mut group: Option<ExecCell> = None;
for item in items {
if matches!(item, ThreadItem::Reasoning { .. }) {
for entry in items {
let item = &entry.payload;
if matches!(item, ThreadItemPayload::Reasoning { .. }) {
continue;
}
let call = CommandHistory::from_item(item.clone())?.into_cell();
@@ -9,16 +9,16 @@ use crate::history_cell;
use crate::history_cell::PlainHistoryCell;
use crate::multi_agents;
use codex_app_server_protocol::PatchApplyStatus;
use codex_app_server_protocol::ThreadItem;
use codex_app_server_protocol::ThreadItemPayload;
use codex_app_server_protocol::WebSearchAction;
use codex_utils_absolute_path::AbsolutePathBuf;
use ratatui::style::Stylize as _;
use std::sync::Arc;
pub(super) fn cells(item: ThreadItem, cwd: &AbsolutePathBuf) -> TranscriptCells {
pub(super) fn cells(item: ThreadItemPayload, cwd: &AbsolutePathBuf) -> TranscriptCells {
let mut cells: TranscriptCells = Vec::new();
match item {
ThreadItem::FileChange {
ThreadItemPayload::FileChange {
id,
changes,
status,
@@ -50,15 +50,17 @@ pub(super) fn cells(item: ThreadItem, cwd: &AbsolutePathBuf) -> TranscriptCells
)));
}
},
ThreadItem::WebSearch(item) => cells.push(Arc::new(history_cell::new_web_search_call(
item.id,
item.query,
item.action.unwrap_or(WebSearchAction::Other),
))),
ThreadItem::ImageView { id, path } => {
ThreadItemPayload::WebSearch(item) => {
cells.push(Arc::new(history_cell::new_web_search_call(
item.id,
item.query,
item.action.unwrap_or(WebSearchAction::Other),
)))
}
ThreadItemPayload::ImageView { id, path } => {
cells.push(Arc::new(history_cell::new_view_image_tool_call(id, path)))
}
ThreadItem::ImageGeneration(item)
ThreadItemPayload::ImageGeneration(item)
if !matches!(item.status.as_str(), "completed" | "failed") =>
{
let status = if item.status.is_empty() {
@@ -71,7 +73,7 @@ pub(super) fn cells(item: ThreadItem, cwd: &AbsolutePathBuf) -> TranscriptCells
/*hint*/ None,
)));
}
ThreadItem::ImageGeneration(item) => {
ThreadItemPayload::ImageGeneration(item) => {
cells.push(Arc::new(history_cell::new_image_generation_call(
item.id,
&item.status,
@@ -79,7 +81,7 @@ pub(super) fn cells(item: ThreadItem, cwd: &AbsolutePathBuf) -> TranscriptCells
item.saved_path,
)));
}
item @ ThreadItem::CollabAgentToolCall { .. } => {
item @ ThreadItemPayload::CollabAgentToolCall { .. } => {
if let Some(cell) = multi_agents::tool_call_history_cell(
&item,
/*cached_spawn_request*/ None,
@@ -88,29 +90,29 @@ pub(super) fn cells(item: ThreadItem, cwd: &AbsolutePathBuf) -> TranscriptCells
cells.push(Arc::new(cell));
}
}
item @ ThreadItem::SubAgentActivity { .. } => {
item @ ThreadItemPayload::SubAgentActivity { .. } => {
if let Some(cell) = multi_agents::sub_agent_activity_history_cell(&item) {
cells.push(Arc::new(cell));
}
}
ThreadItem::EnteredReviewMode { review, .. } => {
ThreadItemPayload::EnteredReviewMode { review, .. } => {
cells.push(Arc::new(history_cell::new_review_status_line(format!(
">> Code review started: {review} <<"
))));
}
ThreadItem::ExitedReviewMode { review, .. } => {
ThreadItemPayload::ExitedReviewMode { review, .. } => {
cells.push(Arc::new(history_cell::new_review_status_line(format!(
"<< Code review finished: {review} >>"
))));
}
ThreadItem::ContextCompaction { .. } => {
ThreadItemPayload::ContextCompaction { .. } => {
cells.push(Arc::new(history_cell::new_info_event(
"Context compacted".to_string(),
/*hint*/ None,
)));
}
// These items do not have richer history-cell presentations yet.
ThreadItem::HookPrompt { fragments, .. } => {
ThreadItemPayload::HookPrompt { fragments, .. } => {
if !fragments.is_empty() {
cells.push(Arc::new(PlainHistoryCell::new(
fragments
@@ -126,19 +128,19 @@ pub(super) fn cells(item: ThreadItem, cwd: &AbsolutePathBuf) -> TranscriptCells
)));
}
}
item @ ThreadItem::DynamicToolCall { .. } => {
item @ ThreadItemPayload::DynamicToolCall { .. } => {
if let Some(cell) = history_cell::DynamicToolCallCell::from_item(item) {
cells.push(Arc::new(cell));
}
}
ThreadItem::UserMessage { .. }
| ThreadItem::AgentMessage { .. }
| ThreadItem::FunctionCallOutput { .. }
| ThreadItem::Plan { .. }
| ThreadItem::Reasoning { .. }
| ThreadItem::CommandExecution { .. }
| ThreadItem::McpToolCall { .. }
| ThreadItem::Sleep(_) => {}
ThreadItemPayload::UserMessage { .. }
| ThreadItemPayload::AgentMessage { .. }
| ThreadItemPayload::FunctionCallOutput { .. }
| ThreadItemPayload::Plan { .. }
| ThreadItemPayload::Reasoning { .. }
| ThreadItemPayload::CommandExecution { .. }
| ThreadItemPayload::McpToolCall { .. }
| ThreadItemPayload::Sleep(_) => {}
}
cells
}
@@ -12,7 +12,7 @@ use codex_app_server_protocol::ImageGenerationItem;
use codex_app_server_protocol::PatchApplyStatus;
use codex_app_server_protocol::PatchChangeKind;
use codex_app_server_protocol::SubAgentActivityKind;
use codex_app_server_protocol::ThreadItem;
use codex_app_server_protocol::ThreadItemPayload;
use codex_app_server_protocol::WebSearchAction;
use codex_app_server_protocol::WebSearchItem;
use codex_utils_path_uri::LegacyAppPathString;
@@ -68,7 +68,7 @@ fn completed_patch_restores_rich_diff_and_styles() {
);
changes[2].diff.push_str("\n\nMoved to: src/after.rs");
let actual = cells(
ThreadItem::FileChange {
ThreadItemPayload::FileChange {
id: "patch-1".to_string(),
changes,
status: PatchApplyStatus::Completed,
@@ -94,7 +94,7 @@ fn unfinished_and_rejected_patches_keep_their_outcome() {
.into_iter()
.flat_map(|status| {
cells(
ThreadItem::FileChange {
ThreadItemPayload::FileChange {
id: "patch-1".to_string(),
changes: vec![FileUpdateChange {
path: "main.rs".to_string(),
@@ -136,11 +136,11 @@ fn tool_and_notice_projection_uses_normal_transcript_presentation() {
generation_id: None,
};
let items = vec![
ThreadItem::EnteredReviewMode {
ThreadItemPayload::EnteredReviewMode {
id: "review-start".to_string(),
review: "current changes".to_string(),
},
ThreadItem::WebSearch(WebSearchItem {
ThreadItemPayload::WebSearch(WebSearchItem {
id: "search-1".to_string(),
query: "fallback query".to_string(),
action: Some(WebSearchAction::FindInPage {
@@ -149,16 +149,16 @@ fn tool_and_notice_projection_uses_normal_transcript_presentation() {
}),
results: None,
}),
ThreadItem::ImageView {
ThreadItemPayload::ImageView {
id: "view-1".to_string(),
path: LegacyAppPathString::from_string("diagram.png".to_string()),
},
ThreadItem::ImageGeneration(image.clone()),
ThreadItem::ImageGeneration(ImageGenerationItem {
ThreadItemPayload::ImageGeneration(image.clone()),
ThreadItemPayload::ImageGeneration(ImageGenerationItem {
status: "in_progress".into(),
..image
}),
ThreadItem::SubAgentActivity {
ThreadItemPayload::SubAgentActivity {
model: None,
reasoning_effort: None,
id: "agent-1".to_string(),
@@ -166,11 +166,11 @@ fn tool_and_notice_projection_uses_normal_transcript_presentation() {
agent_thread_id: "01912345-1234-7123-8123-123456789abc".to_string(),
agent_path: "/root/reviewer".to_string(),
},
ThreadItem::ExitedReviewMode {
ThreadItemPayload::ExitedReviewMode {
id: "review-end".to_string(),
review: "No findings".to_string(),
},
ThreadItem::ContextCompaction {
ThreadItemPayload::ContextCompaction {
id: "compact-1".to_string(),
},
];
+9 -9
View File
@@ -18,7 +18,7 @@ use codex_app_server_protocol::CollabAgentToolCallStatus;
use codex_app_server_protocol::CommandExecutionSource;
use codex_app_server_protocol::CommandExecutionStatus;
use codex_app_server_protocol::McpToolCallStatus;
use codex_app_server_protocol::ThreadItem;
use codex_app_server_protocol::ThreadItemPayload;
use codex_protocol::mcp::CallToolResult;
use codex_protocol::parse_command::ParsedCommand;
use ratatui::style::Stylize as _;
@@ -27,9 +27,9 @@ use ratatui::text::Line;
/// Preserve status and output when replay cannot reconstruct a rich completed tool cell.
/// Declined commands never ran, and historical terminal interactions do not carry stdin details.
/// Pending items retain their last known state without starting a running clock.
pub(crate) fn historical_tool_fallback(item: &ThreadItem) -> Option<PlainHistoryCell> {
pub(crate) fn historical_tool_fallback(item: &ThreadItemPayload) -> Option<PlainHistoryCell> {
let lines = match item {
ThreadItem::CommandExecution {
ThreadItemPayload::CommandExecution {
command,
source,
status,
@@ -73,7 +73,7 @@ pub(crate) fn historical_tool_fallback(item: &ThreadItem) -> Option<PlainHistory
}
lines
}
ThreadItem::McpToolCall {
ThreadItemPayload::McpToolCall {
server,
tool,
status,
@@ -88,7 +88,7 @@ pub(crate) fn historical_tool_fallback(item: &ThreadItem) -> Option<PlainHistory
};
vec![format!("mcp tool: {server}/{tool} · {status}").dim().into()]
}
ThreadItem::CollabAgentToolCall {
ThreadItemPayload::CollabAgentToolCall {
tool:
tool @ (CollabAgentTool::SpawnAgent
| CollabAgentTool::SendInput
@@ -128,8 +128,8 @@ pub(crate) struct CommandHistory {
}
impl CommandHistory {
pub(crate) fn from_item(item: ThreadItem) -> Option<Self> {
let ThreadItem::CommandExecution {
pub(crate) fn from_item(item: ThreadItemPayload) -> Option<Self> {
let ThreadItemPayload::CommandExecution {
id,
command,
source,
@@ -192,8 +192,8 @@ pub(crate) struct McpHistory {
}
impl McpHistory {
pub(crate) fn from_item(item: ThreadItem) -> Option<Self> {
let ThreadItem::McpToolCall {
pub(crate) fn from_item(item: ThreadItemPayload) -> Option<Self> {
let ThreadItemPayload::McpToolCall {
id,
server,
tool,
@@ -9,8 +9,8 @@ use codex_utils_path_uri::LegacyAppPathString;
use pretty_assertions::assert_eq;
use serde_json::json;
fn command_item(status: CommandExecutionStatus) -> ThreadItem {
ThreadItem::CommandExecution {
fn command_item(status: CommandExecutionStatus) -> ThreadItemPayload {
ThreadItemPayload::CommandExecution {
id: "command".to_string(),
plugin_id: None,
script_path: None,
@@ -35,8 +35,8 @@ fn command_item(status: CommandExecutionStatus) -> ThreadItem {
}
}
fn mcp_item(server: &str, id: &str) -> ThreadItem {
ThreadItem::McpToolCall {
fn mcp_item(server: &str, id: &str) -> ThreadItemPayload {
ThreadItemPayload::McpToolCall {
id: id.to_string(),
server: server.to_string(),
tool: "js".to_string(),
@@ -100,7 +100,7 @@ fn completed_tools_keep_compact_and_detailed_presentations() {
fn failed_commands_keep_failure_when_exit_code_is_missing_or_zero() {
for exit in [None, Some(0)] {
let mut item = command_item(CommandExecutionStatus::Failed);
if let ThreadItem::CommandExecution { exit_code, .. } = &mut item {
if let ThreadItemPayload::CommandExecution { exit_code, .. } = &mut item {
*exit_code = exit;
}
let command = CommandHistory::from_item(item).unwrap();
@@ -112,12 +112,12 @@ fn failed_commands_keep_failure_when_exit_code_is_missing_or_zero() {
fn incomplete_tool_payloads_preserve_last_known_status_and_output() {
let command = command_item(CommandExecutionStatus::InProgress);
let mut mcp = mcp_item("example", "pending");
if let ThreadItem::McpToolCall { status, result, .. } = &mut mcp {
if let ThreadItemPayload::McpToolCall { status, result, .. } = &mut mcp {
*status = McpToolCallStatus::InProgress;
*result = None;
}
let mut missing_result = mcp_item("example", "completed");
if let ThreadItem::McpToolCall { result, .. } = &mut missing_result {
if let ThreadItemPayload::McpToolCall { result, .. } = &mut missing_result {
*result = None;
}
let cwd = test_path_buf("/workspace").abs();
@@ -141,7 +141,7 @@ fn historical_command_fallbacks_preserve_status_output_and_group_boundaries() {
let cwd = test_path_buf("/workspace").abs();
let exploration = |call_id: &str| {
let mut item = command_item(CommandExecutionStatus::Completed);
if let ThreadItem::CommandExecution {
if let ThreadItemPayload::CommandExecution {
id,
command,
command_actions,
@@ -204,7 +204,7 @@ fn historical_command_fallbacks_preserve_status_output_and_group_boundaries() {
),
] {
let mut item = command_item(status);
if let ThreadItem::CommandExecution {
if let ThreadItemPayload::CommandExecution {
source: item_source,
command,
aggregated_output,
@@ -305,7 +305,7 @@ fn agent_tool_fallbacks_preserve_status_without_duplicating_v2_activity() {
| CollabAgentTool::ResumeAgent
| CollabAgentTool::Wait
);
let item = ThreadItem::CollabAgentToolCall {
let item = ThreadItemPayload::CollabAgentToolCall {
id: "pending-agent-call".to_string(),
tool,
status,