openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-18 17:18:52 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(rollout): persist TurnItems for paginated thread rollouts (#30188)

Browse Source 
## Description

This PR makes new threads with `history_mode = "paginated"` persist
`ItemCompleted(item: <turn_item>)` in their rollout JSONL file.

Legacy threads keep persisting the existing legacy events. Because the
format is selected per thread, a rollout is either legacy or paginated;
we do not need to support mixed rollouts containing both
representations.

This PR depends on [#31473](https://github.com/openai/codex/pull/31473).

## Why

Paginated thread history needs stable turn/item IDs and completed item
snapshots so the later SQLite projector can materialize appended rollout
JSONL without rebuilding the whole thread.

Keeping the legacy persistence policy unchanged avoids changing
historical rollouts or the readers that still consume them.

## What changed

- Made rollout filtering history-mode aware. Paginated threads keep
completed canonical `ItemCompleted` events and drop their redundant
legacy projections; legacy threads keep the existing event set.
- Made forks inherit the source thread history mode, so copied legacy
history is never filtered as paginated.
- Made paginated threads assign IDs to locally-created response items
even when `Feature::ItemIds` is off, and reject streamed output items
that arrive without server IDs.
- Updated legacy turn replay, rollout list/search, and SQLite metadata
extraction to understand completed canonical user-message items.
This commit is contained in:
Owen Lin
2026-07-08 19:55:03 -07:00
committed by GitHub
parent 5892c7b69d
commit 2342b2c2a6
26 changed files with 497 additions and 220 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server/src/request_processors.rs
+3 -3
View File
@@ -438,8 +438,8 @@ use codex_protocol::protocol::SessionConfiguredEvent;
use codex_protocol::protocol::SessionMetaLine;
use codex_protocol::protocol::TurnEnvironmentSelection;
use codex_protocol::protocol::TurnEnvironmentSelections;
use codex_protocol::protocol::USER_MESSAGE_BEGIN;
use codex_protocol::protocol::W3cTraceContext;
use codex_protocol::protocol::strip_user_message_prefix;
use codex_protocol::user_input::MAX_USER_INPUT_TEXT_CHARS;
use codex_protocol::user_input::UserInput as CoreInputItem;
use codex_rmcp_client::perform_oauth_login_return_url_with_http_client;
@@ -625,10 +625,10 @@ pub(crate) use self::thread_summary::summary_to_thread;
pub(crate) use self::thread_summary::thread_settings_from_config_snapshot;
pub(crate) use self::thread_summary::thread_settings_from_core_snapshot;
pub(crate) fn build_api_turns_from_rollout_items(items: &[RolloutItem]) -> Vec<Turn> {
pub(crate) fn build_legacy_api_turns_from_rollout_items(items: &[RolloutItem]) -> Vec<Turn> {
let mut builder = ThreadHistoryBuilder::new();
for item in items {
if is_persisted_rollout_item(item) {
if is_persisted_rollout_item(item, codex_protocol::protocol::ThreadHistoryMode::Legacy) {
builder.handle_rollout_item(item);
}
}
codex-rs/app-server/src/request_processors/external_agent_session_import.rs
+1 -1
View File
@@ -231,7 +231,7 @@ impl ExternalAgentSessionImporter {
memory_mode,
},
};
rollout_items.retain(is_persisted_rollout_item);
rollout_items.retain(|item| is_persisted_rollout_item(item, ThreadHistoryMode::Legacy));
let title = title
.as_deref()
.and_then(codex_core::util::normalize_thread_name);
codex-rs/app-server/src/request_processors/thread_lifecycle.rs
+1 -1
View File
@@ -738,7 +738,7 @@ pub(crate) fn populate_thread_turns_from_history(
items: &[RolloutItem],
active_turn: Option<&Turn>,
) {
let mut turns = build_api_turns_from_rollout_items(items);
let mut turns = build_legacy_api_turns_from_rollout_items(items);
if let Some(active_turn) = active_turn {
merge_turn_history_with_active_turn(&mut turns, active_turn.clone());
}
codex-rs/app-server/src/request_processors/thread_processor.rs
+4 -7
View File
@@ -2339,7 +2339,7 @@ impl ThreadRequestProcessor {
let (mut thread, history) =
thread_from_stored_thread(stored_thread, fallback_provider, &self.config.cwd);
if include_turns && let Some(history) = history {
thread.turns = build_api_turns_from_rollout_items(&history.items);
thread.turns = build_legacy_api_turns_from_rollout_items(&history.items);
}
Ok(Some(thread))
}
@@ -2408,7 +2408,7 @@ impl ThreadRequestProcessor {
.load_history(/*include_archived*/ true)
.await
.map_err(|err| thread_read_history_load_error(thread_id, err))?;
thread.turns = build_api_turns_from_rollout_items(&history.items);
thread.turns = build_legacy_api_turns_from_rollout_items(&history.items);
}
Ok(())
@@ -4107,7 +4107,7 @@ fn reconstruct_thread_turns_for_turns_list(
|| active_turn
.as_ref()
.is_some_and(|turn| matches!(turn.status, TurnStatus::InProgress));
let mut turns = build_api_turns_from_rollout_items(items);
let mut turns = build_legacy_api_turns_from_rollout_items(items);
normalize_thread_turns_status(&mut turns, loaded_status, has_live_in_progress_turn);
if let Some(active_turn) = active_turn {
merge_turn_history_with_active_turn(&mut turns, active_turn);
@@ -4480,10 +4480,7 @@ fn preview_from_rollout_items(items: &[RolloutItem]) -> String {
},
_ => None,
})
.map(|preview| match preview.find(USER_MESSAGE_BEGIN) {
Some(idx) => preview[idx + USER_MESSAGE_BEGIN.len()..].trim().to_string(),
None => preview,
})
.map(|preview| strip_user_message_prefix(preview.as_str()).to_string())
.unwrap_or_default()
}
codex-rs/app-server/src/request_processors/thread_summary.rs
+1 -4
View File
@@ -97,10 +97,7 @@ fn extract_conversation_summary(
_ => None,
})?;
let preview = match preview.find(USER_MESSAGE_BEGIN) {
Some(idx) => preview[idx + USER_MESSAGE_BEGIN.len()..].trim(),
None => preview.as_str(),
};
let preview = strip_user_message_prefix(preview.as_str());
let timestamp = if session_meta.timestamp.is_empty() {
None
codex-rs/app-server/src/request_processors/thread_summary_tests.rs
+1
View File
@@ -1,6 +1,7 @@
use super::*;
use anyhow::Result;
use codex_protocol::protocol::USER_MESSAGE_BEGIN;
use pretty_assertions::assert_eq;
use serde_json::json;
use std::path::PathBuf;
codex-rs/core/src/session/mod.rs
+4 -4
View File
@@ -1379,7 +1379,7 @@ impl Session {
}
InitialHistory::Forked(mut rollout_items) => {
let turn_context = self.new_default_turn().await;
if turn_context.config.features.enabled(Feature::ItemIds) {
if turn_context.item_ids_enabled() {
for rollout_item in &mut rollout_items {
if let RolloutItem::ResponseItem(response_item) = rollout_item {
Self::assign_missing_response_item_id(response_item);
@@ -1962,7 +1962,7 @@ impl Session {
}
async fn send_event_raw_with_persistence(&self, event: Event, persist: bool) {
// Persist the event into rollout storage (the store filters as needed).
// Persist the event into rollout storage; the store applies its persistence policy.
if persist {
let rollout_items = vec![RolloutItem::EventMsg(event.msg.clone())];
self.persist_rollout_items(&rollout_items).await;
@@ -2776,7 +2776,7 @@ impl Session {
for item in items.to_mut() {
item.set_turn_id_if_missing(&turn_context.sub_id);
}
if turn_context.config.features.enabled(Feature::ItemIds) {
if turn_context.item_ids_enabled() {
Self::assign_missing_response_item_ids(items)
} else {
items
@@ -3035,7 +3035,7 @@ impl Session {
world_state_baseline: Option<Arc<WorldState>>,
compacted_item: CompactedItem,
) {
let items = if turn_context.config.features.enabled(Feature::ItemIds) {
let items = if turn_context.item_ids_enabled() {
Self::assign_missing_response_item_ids(Cow::Owned(items)).into_owned()
} else {
items
codex-rs/core/src/session/review.rs
+1
View File
@@ -120,6 +120,7 @@ pub(super) async fn spawn_review_thread(
reasoning_effort,
reasoning_summary,
session_source,
history_mode: parent_turn_context.history_mode,
parent_thread_id: parent_turn_context.parent_thread_id,
originator: parent_turn_context.originator.clone(),
environments: parent_turn_context.environments.clone(),
codex-rs/core/src/session/session.rs
+11 -2
View File
@@ -634,7 +634,12 @@ impl Session {
},
},
};
LiveThread::resume(Arc::clone(&thread_store), params).await?
LiveThread::resume(
Arc::clone(&thread_store),
session_configuration.history_mode,
params,
)
.await?
}
};
Ok(Some(live_thread))
@@ -1110,7 +1115,11 @@ impl Session {
config.features.enabled(Feature::EnableRequestCompression),
config.features.enabled(Feature::RuntimeMetrics),
Self::build_model_client_beta_features_header(config.as_ref()),
/*item_ids_enabled*/ config.features.enabled(Feature::ItemIds),
/*item_ids_enabled*/ config.features.enabled(Feature::ItemIds)
|| matches!(
session_configuration.history_mode,
ThreadHistoryMode::Paginated
),
/*concurrent_reasoning_summaries_enabled*/ config
.features
.enabled(Feature::ConcurrentReasoningSummaries),
codex-rs/core/src/session/tests.rs
+19
View File
@@ -67,6 +67,7 @@ use codex_protocol::permissions::FileSystemSandboxPolicy;
use codex_protocol::permissions::FileSystemSpecialPath;
use codex_protocol::protocol::NonSteerableTurnKind;
use codex_protocol::protocol::SandboxPolicy;
use codex_protocol::protocol::ThreadHistoryMode;
use codex_protocol::protocol::TurnEnvironmentSelections;
use codex_protocol::request_permissions::PermissionGrantScope;
use codex_protocol::request_permissions::RequestPermissionProfile;
@@ -257,6 +258,24 @@ fn assign_missing_response_item_ids_assigns_additional_tools_ids() {
assert!(items[0].id().is_some_and(|id| id.starts_with("at_")));
}
#[tokio::test]
async fn paginated_turn_context_assigns_missing_response_item_ids_without_feature() {
let (session, mut turn_context) = make_session_and_context().await;
turn_context.history_mode = ThreadHistoryMode::Paginated;
let response_item = user_message("hello");
let items = session.prepare_conversation_items_for_history(
&turn_context,
std::slice::from_ref(&response_item),
);
assert!(
items[0]
.id()
.is_some_and(|item_id| item_id.starts_with("msg_"))
);
}
fn assistant_message(text: &str) -> ResponseItem {
ResponseItem::Message {
id: None,
codex-rs/core/src/session/turn.rs
+23 -2
View File
@@ -1916,6 +1916,21 @@ async fn drain_in_flight(
Ok(())
}
fn assign_missing_streamed_response_item_id(
item: &mut ResponseItem,
active_item: Option<&TurnItem>,
) {
if item.id().is_some() {
return;
}
let active_item_id = active_item
.map(TurnItem::id)
.filter(|item_id| !item_id.is_empty());
item.set_id(active_item_id);
Session::assign_missing_response_item_id(item);
}
#[allow(clippy::too_many_arguments)]
#[instrument(level = "trace",
skip_all,
@@ -2030,7 +2045,10 @@ async fn try_run_sampling_request(
match event {
ResponseEvent::Created => {}
ResponseEvent::OutputItemDone(item) => {
ResponseEvent::OutputItemDone(mut item) => {
if turn_context.item_ids_enabled() {
assign_missing_streamed_response_item_id(&mut item, active_item.as_ref());
}
if let Some((_, mut consumer)) = active_tool_argument_diff_consumer.take()
&& let Ok(Some(event)) = consumer.finish()
{
@@ -2124,7 +2142,10 @@ async fn try_run_sampling_request(
});
}
}
ResponseEvent::OutputItemAdded(item) => {
ResponseEvent::OutputItemAdded(mut item) => {
if turn_context.item_ids_enabled() {
assign_missing_streamed_response_item_id(&mut item, /*active_item*/ None);
}
if let ResponseItem::CustomToolCall {
call_id,
name,
codex-rs/core/src/session/turn_context.rs
+9
View File
@@ -10,6 +10,7 @@ use codex_protocol::ThreadId;
use codex_protocol::models::AdditionalPermissionProfile;
use codex_protocol::openai_models::ModelInfo;
use codex_protocol::protocol::MultiAgentVersion;
use codex_protocol::protocol::ThreadHistoryMode;
use codex_protocol::protocol::TurnEnvironmentSelection;
use codex_sandboxing::compatibility_sandbox_policy_for_permission_profile;
use codex_sandboxing::policy_transforms::effective_file_system_sandbox_policy;
@@ -111,6 +112,7 @@ pub struct TurnContext {
pub(crate) reasoning_effort: Option<ReasoningEffortConfig>,
pub(crate) reasoning_summary: ReasoningSummaryConfig,
pub(crate) session_source: SessionSource,
pub(crate) history_mode: ThreadHistoryMode,
pub(crate) parent_thread_id: Option<ThreadId>,
pub(crate) originator: String,
pub(crate) environments: TurnEnvironmentSnapshot,
@@ -149,6 +151,11 @@ enum TurnMultiAgentRuntime {
}
impl TurnContext {
pub(crate) fn item_ids_enabled(&self) -> bool {
self.config.features.enabled(Feature::ItemIds)
|| matches!(self.history_mode, ThreadHistoryMode::Paginated)
}
pub(crate) fn permission_profile(&self) -> PermissionProfile {
self.permission_profile.clone()
}
@@ -265,6 +272,7 @@ impl TurnContext {
reasoning_effort,
reasoning_summary: self.reasoning_summary,
session_source: self.session_source.clone(),
history_mode: self.history_mode,
parent_thread_id: self.parent_thread_id,
originator: self.originator.clone(),
environments: self.environments.clone(),
@@ -544,6 +552,7 @@ impl Session {
reasoning_effort,
reasoning_summary,
session_source,
history_mode: session_configuration.history_mode,
parent_thread_id: session_configuration.parent_thread_id,
originator: session_configuration.originator.clone(),
environments,
codex-rs/core/tests/suite/items.rs
+70
View File
@@ -276,6 +276,76 @@ async fn reasoning_item_is_emitted() -> anyhow::Result<()> {
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn missing_streamed_reasoning_id_is_reused_for_completion() -> anyhow::Result<()> {
skip_if_no_network!(Ok(()));
let server = start_mock_server().await;
let TestCodex { codex, .. } = test_codex()
.with_config(|config| {
let _ = config.features.enable(Feature::ItemIds);
})
.build_with_auto_env(&server)
.await?;
let mut reasoning_added = ev_reasoning_item_added("unused", &[]);
reasoning_added["item"]
.as_object_mut()
.expect("reasoning item")
.remove("id");
let mut reasoning_done = ev_reasoning_item("unused", &["Consider inputs"], &[]);
reasoning_done["item"]
.as_object_mut()
.expect("reasoning item")
.remove("id");
let response_mock = mount_sse_once(
&server,
sse(vec![
ev_response_created("resp-1"),
reasoning_added,
reasoning_done,
ev_completed("resp-1"),
]),
)
.await;
codex
.submit(Op::UserInput {
items: vec![UserInput::Text {
text: "explain your reasoning".into(),
text_elements: Vec::new(),
}],
final_output_json_schema: None,
responsesapi_client_metadata: None,
additional_context: Default::default(),
thread_settings: Default::default(),
})
.await?;
let started_id = wait_for_event_match(&codex, |ev| match ev {
EventMsg::ItemStarted(ItemStartedEvent {
item: TurnItem::Reasoning(item),
..
}) => Some(item.id.clone()),
_ => None,
})
.await;
let completed_id = wait_for_event_match(&codex, |ev| match ev {
EventMsg::ItemCompleted(ItemCompletedEvent {
item: TurnItem::Reasoning(item),
..
}) => Some(item.id.clone()),
_ => None,
})
.await;
assert!(started_id.starts_with("rs_"));
assert_eq!(started_id, completed_id);
response_mock.single_request();
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn web_search_item_is_emitted() -> anyhow::Result<()> {
skip_if_no_network!(Ok(()));
codex-rs/protocol/src/legacy_events.rs
+7 -3
View File
@@ -75,10 +75,10 @@ impl ContextCompactionItem {
}
impl UserMessageItem {
pub fn as_legacy_event(&self) -> EventMsg {
pub fn as_legacy_user_message_event(&self) -> UserMessageEvent {
// Legacy user-message events flatten only text inputs into `message` and
// rebase text element ranges onto that concatenated text.
EventMsg::UserMessage(UserMessageEvent {
UserMessageEvent {
client_id: self.client_id.clone(),
message: self.message(),
images: Some(self.image_urls()),
@@ -86,7 +86,11 @@ impl UserMessageItem {
local_images: self.local_image_paths(),
local_image_details: self.local_image_details(),
text_elements: self.text_elements(),
})
}
}
pub fn as_legacy_event(&self) -> EventMsg {
EventMsg::UserMessage(self.as_legacy_user_message_event())
}
}
codex-rs/protocol/src/protocol.rs
+36 -10
View File
@@ -120,6 +120,14 @@ pub const CONTEXT_WINDOW_GUIDANCE_OPEN_TAG: &str = "<context_window_guidance>";
pub const CONTEXT_WINDOW_GUIDANCE_CLOSE_TAG: &str = "</context_window_guidance>";
pub const USER_MESSAGE_BEGIN: &str = "## My request for Codex:";
/// Removes the model-context prefix from a user message before displaying it.
pub fn strip_user_message_prefix(text: &str) -> &str {
match text.find(USER_MESSAGE_BEGIN) {
Some(idx) => text[idx + USER_MESSAGE_BEGIN.len()..].trim(),
None => text.trim(),
}
}
// TODO(anp): Replace `TurnEnvironmentSelection` with `PathUri` once path URIs carry environment
// identifiers.
#[derive(Debug, Clone, PartialEq)]
@@ -2294,6 +2302,23 @@ pub struct UserMessageEvent {
pub text_elements: Vec<crate::user_input::TextElement>,
}
/// Returns the user-facing preview text for a user message.
pub fn user_message_preview(user: &UserMessageEvent) -> Option<String> {
let message = strip_user_message_prefix(user.message.as_str());
if !message.is_empty() {
return Some(message.to_string());
}
if user
.images
.as_ref()
.is_some_and(|images| !images.is_empty())
|| !user.local_images.is_empty()
{
return Some("[Image]".to_string());
}
None
}
#[derive(Debug, Clone, Deserialize, Serialize, JsonSchema, TS)]
pub struct AgentReasoningEvent {
pub text: String,
@@ -2591,11 +2616,11 @@ impl InitialHistory {
pub fn get_history_mode(&self, default_history_mode: ThreadHistoryMode) -> ThreadHistoryMode {
match self {
InitialHistory::New | InitialHistory::Cleared | InitialHistory::Forked(_) => {
default_history_mode
}
InitialHistory::Resumed(_) => self
.get_resumed_session_meta()
InitialHistory::New | InitialHistory::Cleared => default_history_mode,
// Forks copy their source rollout items as-is, so they must keep
// the source format instead of converting to the destination default.
InitialHistory::Resumed(_) | InitialHistory::Forked(_) => self
.get_session_meta()
.map(|meta| meta.history_mode)
.unwrap_or(default_history_mode),
}
@@ -5782,13 +5807,13 @@ mod tests {
}
#[test]
fn resumed_history_uses_persisted_history_mode() -> Result<()> {
fn copied_history_uses_persisted_history_mode() -> Result<()> {
let thread_id = ThreadId::from_string("00000000-0000-0000-0000-000000000001")?;
let session_meta = RolloutItem::SessionMeta(SessionMetaLine {
meta: SessionMeta {
session_id: thread_id.into(),
id: thread_id,
history_mode: ThreadHistoryMode::Paginated,
history_mode: ThreadHistoryMode::Legacy,
..SessionMeta::default()
},
git: None,
@@ -5800,11 +5825,12 @@ mod tests {
});
assert_eq!(
history.get_history_mode(ThreadHistoryMode::Legacy),
ThreadHistoryMode::Paginated
history.get_history_mode(ThreadHistoryMode::Paginated),
ThreadHistoryMode::Legacy
);
assert_eq!(
InitialHistory::Forked(vec![session_meta]).get_history_mode(ThreadHistoryMode::Legacy),
InitialHistory::Forked(vec![session_meta])
.get_history_mode(ThreadHistoryMode::Paginated),
ThreadHistoryMode::Legacy
);
assert_eq!(
codex-rs/rollout/src/list.rs
+18 -25
View File
@@ -22,12 +22,13 @@ use crate::protocol::EventMsg;
use crate::state_db;
use codex_file_search as file_search;
use codex_protocol::ThreadId;
use codex_protocol::items::TurnItem;
use codex_protocol::protocol::RolloutItem;
use codex_protocol::protocol::RolloutLine;
use codex_protocol::protocol::SessionMetaLine;
use codex_protocol::protocol::SessionSource;
use codex_protocol::protocol::ThreadHistoryMode;
use codex_protocol::protocol::USER_MESSAGE_BEGIN;
use codex_protocol::protocol::user_message_preview;
use serde_json::Value;
/// Returned page of thread (thread) summaries.
@@ -1184,12 +1185,19 @@ async fn read_head_summary(path: &Path, head_limit: usize) -> io::Result<HeadTai
}
RolloutItem::EventMsg(ev) => {
if let Some(preview) = event_msg_preview(&ev) {
// Legacy rollouts persist UserMessage while paginated rollouts persist
// ItemCompleted(UserMessage), so summaries must recognize both formats.
let is_user_message = match &ev {
EventMsg::UserMessage(_) => true,
EventMsg::ItemCompleted(event) => {
matches!(event.item, TurnItem::UserMessage(_))
}
_ => false,
};
if summary.preview.is_none() {
summary.preview = Some(preview.clone());
}
if let EventMsg::UserMessage(_) = ev
&& summary.first_user_message.is_none()
{
if is_user_message && summary.first_user_message.is_none() {
summary.first_user_message = Some(preview);
}
}
@@ -1250,30 +1258,15 @@ pub async fn read_head_for_summary(path: &Path) -> io::Result<Vec<serde_json::Va
Ok(head)
}
fn strip_user_message_prefix(text: &str) -> &str {
match text.find(USER_MESSAGE_BEGIN) {
Some(idx) => text[idx + USER_MESSAGE_BEGIN.len()..].trim(),
None => text.trim(),
}
}
fn event_msg_preview(event: &EventMsg) -> Option<String> {
match event {
EventMsg::UserMessage(user) => {
let message = strip_user_message_prefix(user.message.as_str());
if !message.is_empty() {
return Some(message.to_string());
EventMsg::UserMessage(user) => user_message_preview(user),
EventMsg::ItemCompleted(event) => match &event.item {
TurnItem::UserMessage(user) => {
user_message_preview(&user.as_legacy_user_message_event())
}
if user
.images
.as_ref()
.is_some_and(|images| !images.is_empty())
|| !user.local_images.is_empty()
{
return Some("[Image]".to_string());
}
None
}
_ => None,
},
EventMsg::ThreadGoalUpdated(event) => {
let objective = event.goal.objective.trim();
(!objective.is_empty()).then(|| objective.to_string())
codex-rs/rollout/src/persistence_metrics.rs
+3 -1
View File
@@ -8,6 +8,7 @@ use codex_protocol::items::TurnItem;
use codex_protocol::models::ResponseItem;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::RolloutItem;
use codex_protocol::protocol::ThreadHistoryMode;
use crate::policy::is_persisted_rollout_item;
@@ -95,6 +96,7 @@ struct TurnMeasurementUpdate {
/// Measures logical JSON sizes while applying the shared rollout persistence policy once.
pub fn measure_and_filter_rollout_items(
items: &[RolloutItem],
history_mode: ThreadHistoryMode,
) -> (Vec<RolloutItem>, RolloutPersistenceBatchMeasurement) {
let mut persisted = Vec::new();
let mut measurement = RolloutPersistenceBatchMeasurement {
@@ -103,7 +105,7 @@ pub fn measure_and_filter_rollout_items(
};
for item in items {
let kept = is_persisted_rollout_item(item);
let kept = is_persisted_rollout_item(item, history_mode);
let decision = if kept {
PersistenceDecision::Kept
} else {
codex-rs/rollout/src/persistence_metrics_tests.rs
+94 -10
View File
@@ -1,11 +1,17 @@
use codex_protocol::ThreadId;
use codex_protocol::items::EnteredReviewModeItem;
use codex_protocol::items::ExitedReviewModeItem;
use codex_protocol::items::TurnItem;
use codex_protocol::items::UserMessageItem;
use codex_protocol::models::ContentItem;
use codex_protocol::models::ResponseItem;
use codex_protocol::protocol::EnteredReviewModeEvent;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::ExitedReviewModeEvent;
use codex_protocol::protocol::ItemCompletedEvent;
use codex_protocol::protocol::ReviewTarget;
use codex_protocol::protocol::RolloutItem;
use codex_protocol::protocol::ThreadHistoryMode;
use codex_protocol::protocol::TurnAbortReason;
use codex_protocol::protocol::TurnAbortedEvent;
use codex_protocol::protocol::TurnCompleteEvent;
@@ -65,7 +71,7 @@ fn update_for_batch(
state: &mut TurnMeasurementState,
items: &[RolloutItem],
) -> super::TurnMeasurementUpdate {
let (_, measurement) = measure_and_filter_rollout_items(items);
let (_, measurement) = measure_and_filter_rollout_items(items, ThreadHistoryMode::Legacy);
update_turn_measurements(state, items, &measurement)
}
@@ -100,7 +106,8 @@ fn mixed_batch_reports_exact_policy_counts_and_bytes() {
let dropped = RolloutItem::ResponseItem(ResponseItem::Other);
let items = vec![kept.clone(), dropped.clone()];
let (persisted, measurement) = measure_and_filter_rollout_items(&items);
let (persisted, measurement) =
measure_and_filter_rollout_items(&items, ThreadHistoryMode::Legacy);
let kept_bytes = serde_json::to_vec(&kept)
.expect("serialize kept item")
.len() as u64;
@@ -128,7 +135,8 @@ fn mixed_batch_reports_exact_policy_counts_and_bytes() {
#[test]
fn retained_items_are_byte_identical() {
let item = retained_message("a moderately sized payload");
let (persisted, measurement) = measure_and_filter_rollout_items(std::slice::from_ref(&item));
let (persisted, measurement) =
measure_and_filter_rollout_items(std::slice::from_ref(&item), ThreadHistoryMode::Legacy);
assert_eq!(
serde_json::to_vec(&persisted[0]).expect("serialize persisted item"),
@@ -155,8 +163,10 @@ fn turn_measurements_span_batches_and_include_items_before_start() {
retained_message("second response"),
turn_aborted("turn-2"),
];
let (_, first_expected) = measure_and_filter_rollout_items(&first_turn);
let (_, second_expected) = measure_and_filter_rollout_items(&second_turn);
let (_, first_expected) =
measure_and_filter_rollout_items(&first_turn, ThreadHistoryMode::Legacy);
let (_, second_expected) =
measure_and_filter_rollout_items(&second_turn, ThreadHistoryMode::Legacy);
let batches = [
first_turn[..1].to_vec(),
first_turn[1..3].to_vec(),
@@ -217,7 +227,8 @@ fn invalid_turn_boundaries_reset_partial_measurements() {
retained_message("retained turn"),
turn_complete("turn-2"),
];
let (_, expected) = measure_and_filter_rollout_items(&replacement[2..]);
let (_, expected) =
measure_and_filter_rollout_items(&replacement[2..], ThreadHistoryMode::Legacy);
let update = update_for_batch(&mut state, &replacement);
assert_eq!(
@@ -235,7 +246,7 @@ fn invalid_turn_boundaries_reset_partial_measurements() {
}
#[test]
fn filtered_item_completion_includes_its_nested_item_type() {
fn item_completion_persistence_depends_on_history_mode() {
let item = RolloutItem::EventMsg(EventMsg::ItemCompleted(ItemCompletedEvent {
thread_id: ThreadId::default(),
turn_id: "turn".to_string(),
@@ -247,14 +258,87 @@ fn filtered_item_completion_includes_its_nested_item_type() {
completed_at_ms: 0,
}));
let (_, measurement) = measure_and_filter_rollout_items(&[item]);
let (_, legacy_measurement) =
measure_and_filter_rollout_items(std::slice::from_ref(&item), ThreadHistoryMode::Legacy);
assert_eq!(
measurement.items[0].rollout_item_type,
legacy_measurement.items[0].rollout_item_type,
"event.item_completed.user_message"
);
assert_eq!(
measurement.items[0].decision,
legacy_measurement.items[0].decision,
super::PersistenceDecision::Dropped
);
let (persisted, paginated_measurement) =
measure_and_filter_rollout_items(std::slice::from_ref(&item), ThreadHistoryMode::Paginated);
assert_eq!(
serde_json::to_value(persisted).expect("serialize persisted items"),
serde_json::to_value([item]).expect("serialize expected items")
);
assert_eq!(
paginated_measurement.items[0].decision,
super::PersistenceDecision::Kept
);
}
#[test]
fn review_mode_persistence_depends_on_history_mode() {
let completed_items = vec![
RolloutItem::EventMsg(EventMsg::ItemCompleted(ItemCompletedEvent {
thread_id: ThreadId::default(),
turn_id: "turn".to_string(),
item: TurnItem::EnteredReviewMode(EnteredReviewModeItem {
id: "entered-review".to_string(),
target: ReviewTarget::Custom {
instructions: "review this".to_string(),
},
user_facing_hint: "Review requested.".to_string(),
}),
completed_at_ms: 0,
})),
RolloutItem::EventMsg(EventMsg::ItemCompleted(ItemCompletedEvent {
thread_id: ThreadId::default(),
turn_id: "turn".to_string(),
item: TurnItem::ExitedReviewMode(ExitedReviewModeItem {
id: "exited-review".to_string(),
review_output: None,
}),
completed_at_ms: 0,
})),
];
let legacy_events = vec![
RolloutItem::EventMsg(EventMsg::EnteredReviewMode(EnteredReviewModeEvent {
target: ReviewTarget::Custom {
instructions: "review this".to_string(),
},
user_facing_hint: Some("Review requested.".to_string()),
turn_id: Some("turn".to_string()),
item_id: Some("entered-review".to_string()),
})),
RolloutItem::EventMsg(EventMsg::ExitedReviewMode(ExitedReviewModeEvent {
turn_id: Some("turn".to_string()),
item_id: Some("exited-review".to_string()),
review_output: None,
})),
];
let items = completed_items
.iter()
.chain(&legacy_events)
.cloned()
.collect::<Vec<_>>();
let (persisted_legacy, _) = measure_and_filter_rollout_items(&items, ThreadHistoryMode::Legacy);
assert_eq!(
serde_json::to_value(persisted_legacy).expect("serialize persisted items"),
serde_json::to_value(legacy_events).expect("serialize expected items")
);
let (persisted_paginated, _) =
measure_and_filter_rollout_items(&items, ThreadHistoryMode::Paginated);
assert_eq!(
serde_json::to_value(persisted_paginated).expect("serialize persisted items"),
serde_json::to_value(completed_items).expect("serialize expected items")
);
}
codex-rs/rollout/src/policy.rs
+34 -28
View File
@@ -1,14 +1,16 @@
use crate::protocol::EventMsg;
use crate::protocol::RolloutItem;
use codex_protocol::items::TurnItem;
use codex_protocol::models::ResponseItem;
use codex_protocol::protocol::ThreadHistoryMode;
/// Whether a rollout `item` should be persisted in rollout files.
pub fn is_persisted_rollout_item(item: &RolloutItem) -> bool {
pub fn is_persisted_rollout_item(item: &RolloutItem, history_mode: ThreadHistoryMode) -> bool {
match item {
RolloutItem::ResponseItem(item) => should_persist_response_item(item),
RolloutItem::InterAgentCommunication(_)
| RolloutItem::InterAgentCommunicationMetadata { .. } => true,
RolloutItem::EventMsg(ev) => should_persist_event_msg(ev),
RolloutItem::EventMsg(ev) => should_persist_event_msg(ev, history_mode),
// Persist Codex executive markers so we can analyze flows (e.g., compaction, API turns).
RolloutItem::Compacted(_)
| RolloutItem::TurnContext(_)
@@ -17,11 +19,14 @@ pub fn is_persisted_rollout_item(item: &RolloutItem) -> bool {
}
}
/// Return the canonical rollout items that should be persisted for a live append.
pub fn persisted_rollout_items(items: &[RolloutItem]) -> Vec<RolloutItem> {
/// Return the rollout items that should be persisted for a live append.
pub fn persisted_rollout_items(
items: &[RolloutItem],
history_mode: ThreadHistoryMode,
) -> Vec<RolloutItem> {
let mut persisted = Vec::new();
for item in items {
if is_persisted_rollout_item(item) {
if is_persisted_rollout_item(item, history_mode) {
persisted.push(item.clone());
}
}
@@ -78,37 +83,38 @@ pub fn should_persist_response_item_for_memories(item: &ResponseItem) -> bool {
/// Whether an `EventMsg` should be persisted in rollout files.
#[inline]
pub fn should_persist_event_msg(ev: &EventMsg) -> bool {
pub fn should_persist_event_msg(ev: &EventMsg, history_mode: ThreadHistoryMode) -> bool {
match ev {
EventMsg::UserMessage(_)
| EventMsg::AgentMessage(_)
| EventMsg::AgentReasoning(_)
| EventMsg::AgentReasoningRawContent(_)
| EventMsg::PatchApplyEnd(_)
| EventMsg::TokenCount(_)
EventMsg::ItemCompleted(event) => {
// Paginated rollouts store TurnItems.
// Legacy rollouts keep only items with no raw ResponseItem or legacy equivalent.
matches!(history_mode, ThreadHistoryMode::Paginated)
|| matches!(event.item, TurnItem::Plan(_) | TurnItem::Sleep(_))
}
EventMsg::TokenCount(_)
| EventMsg::ThreadGoalUpdated(_)
| EventMsg::ContextCompacted(_)
| EventMsg::EnteredReviewMode(_)
| EventMsg::ExitedReviewMode(_)
| EventMsg::McpToolCallEnd(_)
| EventMsg::ThreadRolledBack(_)
| EventMsg::TurnAborted(_)
| EventMsg::TurnStarted(_)
| EventMsg::TurnComplete(_)
| EventMsg::ThreadSettingsApplied(_)
| EventMsg::ThreadSettingsApplied(_) => true,
// Only persist these legacy events when the thread's history mode is Legacy.
// New, paginated rollouts persist ItemCompleted events with TurnItems.
EventMsg::UserMessage(_)
| EventMsg::AgentMessage(_)
| EventMsg::AgentReasoning(_)
| EventMsg::AgentReasoningRawContent(_)
| EventMsg::EnteredReviewMode(_)
| EventMsg::ExitedReviewMode(_)
| EventMsg::PatchApplyEnd(_)
| EventMsg::ContextCompacted(_)
| EventMsg::McpToolCallEnd(_)
| EventMsg::WebSearchEnd(_)
| EventMsg::ImageGenerationEnd(_)
| EventMsg::SubAgentActivity(_) => true,
EventMsg::ItemCompleted(event) => {
// These items have no equivalent raw ResponseItem or legacy event,
// so persist their completion for replay without retaining every
// item lifecycle event.
matches!(
event.item,
codex_protocol::items::TurnItem::Plan(_)
| codex_protocol::items::TurnItem::Sleep(_)
)
}
| EventMsg::SubAgentActivity(_) => matches!(history_mode, ThreadHistoryMode::Legacy),
// Transient, non-durable events.
EventMsg::Error(_)
| EventMsg::GuardianAssessment(_)
| EventMsg::ExecCommandEnd(_)
codex-rs/rollout/src/search.rs
+1 -8
View File
@@ -9,7 +9,7 @@ use codex_protocol::models::ResponseItem;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::RolloutItem;
use codex_protocol::protocol::RolloutLine;
use codex_protocol::protocol::USER_MESSAGE_BEGIN;
use codex_protocol::protocol::strip_user_message_prefix;
use regex::Regex;
use regex::RegexBuilder;
use tokio::process::Command;
@@ -297,13 +297,6 @@ fn content_item_text(item: &ContentItem) -> Option<&str> {
}
}
fn strip_user_message_prefix(text: &str) -> &str {
match text.find(USER_MESSAGE_BEGIN) {
Some(idx) => text[idx + USER_MESSAGE_BEGIN.len()..].trim(),
None => text.trim(),
}
}
fn excerpt_around_match(text: &str, search_term: &Regex) -> Option<String> {
let normalized = normalize_preview_text(text);
let matched = search_term.find(normalized.as_str())?;
codex-rs/state/src/extract.rs
+56 -44
View File
@@ -1,16 +1,16 @@
use crate::model::ThreadMetadata;
use codex_protocol::items::TurnItem;
use codex_protocol::models::ResponseItem;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::RolloutItem;
use codex_protocol::protocol::SessionMetaLine;
use codex_protocol::protocol::TurnContextItem;
use codex_protocol::protocol::USER_MESSAGE_BEGIN;
use codex_protocol::protocol::UserMessageEvent;
use codex_protocol::protocol::strip_user_message_prefix;
use codex_protocol::protocol::user_message_preview;
use serde::Serialize;
use serde_json::Value;
const IMAGE_ONLY_USER_MESSAGE_PLACEHOLDER: &str = "[Image]";
/// Apply a rollout item to the metadata structure.
pub fn apply_rollout_item(
metadata: &mut ThreadMetadata,
@@ -39,6 +39,11 @@ pub fn rollout_item_affects_thread_metadata(item: &RolloutItem) -> bool {
RolloutItem::EventMsg(
EventMsg::TokenCount(_) | EventMsg::UserMessage(_) | EventMsg::ThreadGoalUpdated(_),
) => true,
RolloutItem::EventMsg(EventMsg::ItemCompleted(event))
if matches!(event.item, TurnItem::UserMessage(_)) =>
{
true
}
RolloutItem::EventMsg(_)
| RolloutItem::ResponseItem(_)
| RolloutItem::InterAgentCommunication(_)
@@ -96,16 +101,11 @@ fn apply_event_msg(metadata: &mut ThreadMetadata, event: &EventMsg) {
}
}
EventMsg::UserMessage(user) => {
let preview = user_message_preview(user);
if metadata.first_user_message.is_none() {
metadata.first_user_message = preview.clone();
}
set_preview_if_empty(metadata, preview);
if metadata.title.is_empty() {
let title = strip_user_message_prefix(user.message.as_str());
if !title.is_empty() {
metadata.title = title.to_string();
}
apply_user_message(metadata, user);
}
EventMsg::ItemCompleted(event) => {
if let TurnItem::UserMessage(user) = &event.item {
apply_user_message(metadata, &user.as_legacy_user_message_event());
}
}
EventMsg::ThreadGoalUpdated(event) => {
@@ -120,35 +120,26 @@ fn apply_event_msg(metadata: &mut ThreadMetadata, event: &EventMsg) {
fn apply_response_item(_metadata: &mut ThreadMetadata, _item: &ResponseItem) {}
fn apply_user_message(metadata: &mut ThreadMetadata, user: &UserMessageEvent) {
let preview = user_message_preview(user);
if metadata.first_user_message.is_none() {
metadata.first_user_message = preview.clone();
}
set_preview_if_empty(metadata, preview);
if metadata.title.is_empty() {
let title = strip_user_message_prefix(user.message.as_str());
if !title.is_empty() {
metadata.title = title.to_string();
}
}
}
fn set_preview_if_empty(metadata: &mut ThreadMetadata, preview: Option<String>) {
if metadata.preview.is_none() {
metadata.preview = preview;
}
}
fn strip_user_message_prefix(text: &str) -> &str {
match text.find(USER_MESSAGE_BEGIN) {
Some(idx) => text[idx + USER_MESSAGE_BEGIN.len()..].trim(),
None => text.trim(),
}
}
fn user_message_preview(user: &UserMessageEvent) -> Option<String> {
let message = strip_user_message_prefix(user.message.as_str());
if !message.is_empty() {
return Some(message.to_string());
}
if user
.images
.as_ref()
.is_some_and(|images| !images.is_empty())
|| !user.local_images.is_empty()
{
return Some(IMAGE_ONLY_USER_MESSAGE_PLACEHOLDER.to_string());
}
None
}
pub(crate) fn enum_to_string<T: Serialize>(value: &T) -> String {
match serde_json::to_value(value) {
Ok(Value::String(s)) => s,
@@ -164,12 +155,15 @@ mod tests {
use chrono::DateTime;
use chrono::Utc;
use codex_protocol::ThreadId;
use codex_protocol::items::TurnItem;
use codex_protocol::items::UserMessageItem;
use codex_protocol::models::ContentItem;
use codex_protocol::models::PermissionProfile;
use codex_protocol::models::ResponseItem;
use codex_protocol::openai_models::ReasoningEffort;
use codex_protocol::protocol::AskForApproval;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::ItemCompletedEvent;
use codex_protocol::protocol::RolloutItem;
use codex_protocol::protocol::SandboxPolicy;
use codex_protocol::protocol::SessionMeta;
@@ -182,6 +176,7 @@ mod tests {
use codex_protocol::protocol::TurnContextItem;
use codex_protocol::protocol::USER_MESSAGE_BEGIN;
use codex_protocol::protocol::UserMessageEvent;
use codex_protocol::user_input::UserInput;
use pretty_assertions::assert_eq;
use std::path::PathBuf;
@@ -229,6 +224,29 @@ mod tests {
assert_eq!(metadata.title, "actual user request");
}
#[test]
fn completed_user_message_items_set_title_and_first_user_message() {
let mut metadata = metadata_for_test();
let item = RolloutItem::EventMsg(EventMsg::ItemCompleted(ItemCompletedEvent {
thread_id: ThreadId::default(),
turn_id: "turn-1".to_string(),
item: TurnItem::UserMessage(UserMessageItem::new(&[UserInput::Text {
text: format!("{USER_MESSAGE_BEGIN} actual user request"),
text_elements: Vec::new(),
}])),
completed_at_ms: 0,
}));
apply_rollout_item(&mut metadata, &item, "test-provider");
assert_eq!(
metadata.first_user_message.as_deref(),
Some("actual user request")
);
assert_eq!(metadata.preview.as_deref(), Some("actual user request"));
assert_eq!(metadata.title, "actual user request");
}
#[test]
fn event_msg_image_only_user_message_sets_image_placeholder_preview() {
let mut metadata = metadata_for_test();
@@ -243,14 +261,8 @@ mod tests {
apply_rollout_item(&mut metadata, &item, "test-provider");
assert_eq!(
metadata.first_user_message.as_deref(),
Some(super::IMAGE_ONLY_USER_MESSAGE_PLACEHOLDER)
);
assert_eq!(
metadata.preview.as_deref(),
Some(super::IMAGE_ONLY_USER_MESSAGE_PLACEHOLDER)
);
assert_eq!(metadata.first_user_message.as_deref(), Some("[Image]"));
assert_eq!(metadata.preview.as_deref(), Some("[Image]"));
assert_eq!(metadata.title, "");
}
codex-rs/thread-store/src/in_memory.rs
+7 -3
View File
@@ -482,17 +482,21 @@ impl InMemoryThreadStore {
}
async fn append_items(&self, params: AppendThreadItemsParams) -> ThreadStoreResult<()> {
let canonical_items = persisted_rollout_items(params.items.as_slice());
if canonical_items.is_empty() {
if params.items.is_empty() {
return Ok(());
}
let mut state = self.state.lock().await;
let history_mode = history_mode_from_state(&state, params.thread_id);
let persisted_items = persisted_rollout_items(params.items.as_slice(), history_mode);
if persisted_items.is_empty() {
return Ok(());
}
state.calls.append_items += 1;
state
.histories
.entry(params.thread_id)
.or_default()
.extend(canonical_items);
.extend(persisted_items);
Ok(())
}
codex-rs/thread-store/src/live_thread.rs
+19 -11
View File
@@ -3,6 +3,7 @@ use std::sync::Arc;
use codex_protocol::ThreadId;
use codex_protocol::protocol::RolloutItem;
use codex_protocol::protocol::ThreadHistoryMode;
use codex_protocol::protocol::ThreadMemoryMode;
use codex_rollout::RolloutPersistenceTelemetry;
use codex_rollout::measure_and_filter_rollout_items;
@@ -32,6 +33,7 @@ use crate::thread_metadata_sync::ThreadMetadataSync;
#[derive(Clone)]
pub struct LiveThread {
thread_id: ThreadId,
history_mode: ThreadHistoryMode,
thread_store: Arc<dyn ThreadStore>,
metadata_sync: Arc<Mutex<ThreadMetadataSync>>,
persistence_telemetry: RolloutPersistenceTelemetry,
@@ -92,10 +94,12 @@ impl LiveThread {
params: CreateThreadParams,
) -> ThreadStoreResult<Self> {
let thread_id = params.thread_id;
let history_mode = params.history_mode;
let metadata_sync = ThreadMetadataSync::for_create(&params).await;
thread_store.create_thread(params).await?;
Ok(Self {
thread_id,
history_mode,
thread_store,
metadata_sync: Arc::new(Mutex::new(metadata_sync)),
persistence_telemetry: RolloutPersistenceTelemetry::new(thread_id),
@@ -104,6 +108,7 @@ impl LiveThread {
pub async fn resume(
thread_store: Arc<dyn ThreadStore>,
history_mode: ThreadHistoryMode,
params: ResumeThreadParams,
) -> ThreadStoreResult<Self> {
let thread_id = params.thread_id;
@@ -132,6 +137,7 @@ impl LiveThread {
}
Ok(Self {
thread_id,
history_mode,
thread_store,
metadata_sync: Arc::new(Mutex::new(metadata_sync)),
persistence_telemetry: RolloutPersistenceTelemetry::new(thread_id),
@@ -141,36 +147,38 @@ impl LiveThread {
#[tracing::instrument(
level = "trace",
skip_all,
fields(item_count = items.len())
fields(item_count = raw_items.len())
)]
pub async fn append_items(&self, items: &[RolloutItem]) -> ThreadStoreResult<()> {
pub async fn append_items(&self, raw_items: &[RolloutItem]) -> ThreadStoreResult<()> {
// Empty appends are intentionally ignored rather than represented as zero-sized batches.
if items.is_empty() {
if raw_items.is_empty() {
return Ok(());
}
let (canonical_items, measurement) = if self.persistence_telemetry.is_enabled() {
let (canonical_items, measurement) = measure_and_filter_rollout_items(items);
(canonical_items, Some(measurement))
let (items, measurement) = if self.persistence_telemetry.is_enabled() {
let (items, measurement) =
measure_and_filter_rollout_items(raw_items, self.history_mode);
(items, Some(measurement))
} else {
(persisted_rollout_items(items), None)
(persisted_rollout_items(raw_items, self.history_mode), None)
};
self.thread_store
.append_items(AppendThreadItemsParams {
thread_id: self.thread_id,
items: items.to_vec(),
items: raw_items.to_vec(),
})
.await?;
if let Some(measurement) = measurement.as_ref() {
self.persistence_telemetry.record_batch(items, measurement);
self.persistence_telemetry
.record_batch(raw_items, measurement);
}
if canonical_items.is_empty() {
if items.is_empty() {
return Ok(());
}
let update = self
.metadata_sync
.lock()
.await
.observe_appended_items(canonical_items.as_slice());
.observe_appended_items(items.as_slice());
if let Some(update) = update {
self.thread_store
.update_thread_metadata(UpdateThreadMetadataParams {
codex-rs/thread-store/src/local/live_writer.rs
+6 -3
View File
@@ -1,6 +1,7 @@
use std::path::PathBuf;
use codex_protocol::ThreadId;
use codex_protocol::protocol::ThreadHistoryMode;
use codex_protocol::protocol::ThreadMemoryMode;
use codex_rollout::RolloutConfig;
use codex_rollout::RolloutRecorder;
@@ -115,13 +116,15 @@ pub(super) async fn append_items(
store: &LocalThreadStore,
params: AppendThreadItemsParams,
) -> ThreadStoreResult<()> {
let canonical_items = persisted_rollout_items(params.items.as_slice());
if canonical_items.is_empty() {
// LocalThreadStore rejects paginated threads before opening a writer.
let persisted_items =
persisted_rollout_items(params.items.as_slice(), ThreadHistoryMode::Legacy);
if persisted_items.is_empty() {
return Ok(());
}
let recorder = store.live_recorder(params.thread_id).await?;
recorder
.record_canonical_items(canonical_items.as_slice())
.record_canonical_items(persisted_items.as_slice())
.await
.map_err(thread_store_io_error)?;
// LiveThread applies metadata immediately after append_items returns. Wait for the local
codex-rs/thread-store/src/local/mod.rs
+2
View File
@@ -694,6 +694,7 @@ mod tests {
write_session_file(home.path(), "2025-01-03T17-00-00", uuid).expect("session file");
let live_thread = LiveThread::resume(
store,
ThreadHistoryMode::Legacy,
ResumeThreadParams {
thread_id,
rollout_path: Some(rollout_path),
@@ -748,6 +749,7 @@ mod tests {
.expect("external session file");
let live_thread = LiveThread::resume(
store,
ThreadHistoryMode::Legacy,
ResumeThreadParams {
thread_id,
rollout_path: Some(rollout_path),
codex-rs/thread-store/src/thread_metadata_sync.rs
+66 -50
View File
@@ -1,5 +1,3 @@
#[cfg(test)]
use std::sync::Arc;
use std::time::Duration;
use std::time::Instant;
@@ -9,29 +7,27 @@ use chrono::Utc;
use codex_git_utils::collect_git_info;
use codex_git_utils::get_git_repo_root;
use codex_protocol::ThreadId;
use codex_protocol::items::TurnItem;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::GitInfo;
use codex_protocol::protocol::RolloutItem;
use codex_protocol::protocol::ThreadMemoryMode;
use codex_protocol::protocol::USER_MESSAGE_BEGIN;
use codex_protocol::protocol::UserMessageEvent;
use codex_protocol::protocol::strip_user_message_prefix;
use codex_protocol::protocol::user_message_preview;
use crate::CreateThreadParams;
use crate::GitInfoPatch;
use crate::ResumeThreadParams;
use crate::ThreadMetadataPatch;
const IMAGE_ONLY_USER_MESSAGE_PLACEHOLDER: &str = "[Image]";
#[cfg(not(test))]
const THREAD_UPDATED_AT_TOUCH_INTERVAL: Duration = Duration::from_secs(5);
#[cfg(test)]
const THREAD_UPDATED_AT_TOUCH_INTERVAL: Duration = Duration::from_millis(50);
/// Live-thread helper that derives metadata updates from canonical rollout items.
/// Live-thread helper that derives metadata updates from appended rollout items.
///
/// Stores receive raw history plus explicit metadata patches. This helper keeps append-derived
/// metadata observation in the live layer without owning persistence-policy filtering or making
/// `append_items` infer metadata inside a `ThreadStore` implementation.
/// Stores receive raw rollout items plus explicit metadata patches. This helper
/// keeps append-derived metadata observation in the live layer without owning persistence-policy
/// filtering or making `append_items` infer metadata inside a `ThreadStore` implementation.
pub(crate) struct ThreadMetadataSync {
thread_id: ThreadId,
cwd_seen: bool,
@@ -249,22 +245,14 @@ impl ThreadMetadataSync {
update.permission_profile = Some(turn_ctx.permission_profile());
}
RolloutItem::EventMsg(EventMsg::UserMessage(user)) => {
if let Some(preview) = user_message_preview(user) {
if !self.first_user_message_seen {
self.first_user_message_seen = true;
update.first_user_message = Some(preview.clone());
}
if !self.preview_seen {
self.preview_seen = true;
update.preview = Some(preview);
}
}
if !self.title_seen {
let title = strip_user_message_prefix(user.message.as_str());
if !title.is_empty() {
self.title_seen = true;
update.title = Some(title.to_string());
}
self.observe_user_message(user, &mut update);
}
RolloutItem::EventMsg(EventMsg::ItemCompleted(event)) => {
if let TurnItem::UserMessage(user) = &event.item {
self.observe_user_message(
&user.as_legacy_user_message_event(),
&mut update,
);
}
}
RolloutItem::EventMsg(EventMsg::TokenCount(token_count)) => {
@@ -293,6 +281,26 @@ impl ThreadMetadataSync {
Some(update)
}
fn observe_user_message(&mut self, user: &UserMessageEvent, update: &mut ThreadMetadataPatch) {
if let Some(preview) = user_message_preview(user) {
if !self.first_user_message_seen {
self.first_user_message_seen = true;
update.first_user_message = Some(preview.clone());
}
if !self.preview_seen {
self.preview_seen = true;
update.preview = Some(preview);
}
}
if !self.title_seen {
let title = strip_user_message_prefix(user.message.as_str());
if !title.is_empty() {
self.title_seen = true;
update.title = Some(title.to_string());
}
}
}
fn merge_pending_update(&mut self, update: Option<ThreadMetadataPatch>) {
let Some(update) = update else {
return;
@@ -323,29 +331,6 @@ fn parse_session_timestamp(value: &str) -> Option<DateTime<Utc>> {
.ok()
}
fn strip_user_message_prefix(text: &str) -> &str {
match text.find(USER_MESSAGE_BEGIN) {
Some(idx) => text[idx + USER_MESSAGE_BEGIN.len()..].trim(),
None => text.trim(),
}
}
fn user_message_preview(user: &UserMessageEvent) -> Option<String> {
let message = strip_user_message_prefix(user.message.as_str());
if !message.is_empty() {
return Some(message.to_string());
}
if user
.images
.as_ref()
.is_some_and(|images| !images.is_empty())
|| !user.local_images.is_empty()
{
return Some(IMAGE_ONLY_USER_MESSAGE_PLACEHOLDER.to_string());
}
None
}
fn thread_updated_at_touch() -> ThreadMetadataPatch {
ThreadMetadataPatch {
updated_at: Some(Utc::now()),
@@ -387,7 +372,11 @@ fn git_info_patch_from_observation(git_info: GitInfo) -> GitInfoPatch {
#[cfg(test)]
mod tests {
use std::sync::Arc;
use codex_protocol::items::UserMessageItem;
use codex_protocol::protocol::CompactedItem;
use codex_protocol::protocol::ItemCompletedEvent;
use codex_protocol::protocol::SessionMeta;
use codex_protocol::protocol::SessionMetaLine;
use codex_protocol::protocol::SessionSource;
@@ -396,6 +385,7 @@ mod tests {
use codex_protocol::protocol::ThreadGoalUpdatedEvent;
use codex_protocol::protocol::TurnStartedEvent;
use codex_protocol::protocol::UserMessageEvent;
use codex_protocol::user_input::UserInput;
use pretty_assertions::assert_eq;
use super::*;
@@ -484,6 +474,32 @@ mod tests {
assert!(update.patch.updated_at.is_some());
}
#[test]
fn completed_user_message_items_emit_metadata_fields() {
let thread_id = ThreadId::new();
let mut sync = ThreadMetadataSync::for_resume(&resume_params(thread_id, Vec::new()));
let update = sync
.observe_appended_items(&[RolloutItem::EventMsg(EventMsg::ItemCompleted(
ItemCompletedEvent {
thread_id,
turn_id: "turn-1".to_string(),
item: TurnItem::UserMessage(UserMessageItem::new(&[UserInput::Text {
text: "first user text".to_string(),
text_elements: Vec::new(),
}])),
completed_at_ms: 0,
},
))])
.expect("completed user message metadata update");
assert_eq!(update.patch.preview.as_deref(), Some("first user text"));
assert_eq!(update.patch.title.as_deref(), Some("first user text"));
assert_eq!(
update.patch.first_user_message.as_deref(),
Some("first user text")
);
}
#[test]
fn metadata_irrelevant_items_coalesce_updated_at_touches() {
let thread_id = ThreadId::new();