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

Preserve paginated thread metadata across resumes (#35678)

Browse Source 
## Why

Paginated rollout history may contain only a bounded suffix. Deriving display
metadata from that suffix can replace the thread's original preview, title, and
first user message when the thread is read or resumed.

## What changed

- Use SQLite as the canonical source of display metadata for paginated thread
  reads, including reads by an explicit rollout path.
- Seed resume metadata tracking from the persisted values so appended history
  fills only missing fields instead of overwriting existing ones.
- Prefer an explicitly requested rollout path over a stale path stored in
  SQLite while retaining the canonical metadata.

## Testing

Added coverage for reads by ID and path, bounded-history resumes across app
server restarts, preservation of existing metadata, initialization of missing
fields, and stale persisted rollout paths.

GitOrigin-RevId: b393d0727afcba70fda7f50a6f2dcca0632745b4
This commit is contained in:
Owen Lin
2026-07-27 22:12:15 +00:00
committed by copyberry
parent bbeee9a406
commit 4d1f66bf81
6 changed files with 316 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server/tests/suite/v2/thread_read.rs
+1 -1
View File
@@ -1273,7 +1273,7 @@ async fn paginated_thread_name_set_is_reflected_in_read_list_and_metadata_resume
.await?;
// Set a user-facing thread title.
let new_name = "Saved user message";
let new_name = "Custom saved name";
let set_id = mcp
.send_thread_set_name_request(ThreadSetNameParams {
thread_id: conversation_id.clone(),
codex-rs/app-server/tests/suite/v2/thread_resume.rs
+84 -1
View File
@@ -77,6 +77,7 @@ use codex_protocol::models::ContentItem;
use codex_protocol::models::ResponseItem;
use codex_protocol::openai_models::ReasoningEffort;
use codex_protocol::protocol::AgentMessageEvent;
use codex_protocol::protocol::CompactedItem;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::ImageGenerationEndEvent;
use codex_protocol::protocol::McpInvocation;
@@ -135,7 +136,7 @@ const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs
const CODEX_5_2_INSTRUCTIONS_TEMPLATE_DEFAULT: &str = "You are Codex, a coding agent based on GPT-5. You and the user share the same workspace and collaborate to achieve the user's goals.";
#[tokio::test]
async fn thread_resume_paginated_metadata_only_uses_model_context() -> Result<()> {
async fn thread_resume_paginated_model_context_preserves_original_metadata() -> Result<()> {
let server = create_mock_responses_server_repeating_assistant("Done").await;
let codex_home = TempDir::new()?;
mock_responses_config(&server.uri()).write(codex_home.path())?;
@@ -147,6 +148,19 @@ async fn thread_resume_paginated_metadata_only_uses_model_context() -> Result<()
Some("mock_provider"),
/*git_info*/ None,
)?;
let path = rollout_path(codex_home.path(), "2025-01-05T12-00-00", &conversation_id);
append_rollout_item_to_path(
&path,
&RolloutItem::Compacted(CompactedItem {
message: "compacted history".to_string(),
replacement_history: Some(Vec::new()),
window_number: Some(1),
first_window_id: None,
previous_window_id: None,
window_id: None,
}),
)
.await?;
let mut primary = TestAppServer::builder()
.with_codex_home(codex_home.path())
@@ -166,7 +180,76 @@ async fn thread_resume_paginated_metadata_only_uses_model_context() -> Result<()
} = timeout(DEFAULT_READ_TIMEOUT, primary.read_response(resume_id)).await??;
assert_eq!(resumed.id, conversation_id);
assert_eq!(resumed.history_mode, ThreadHistoryMode::Paginated);
assert_eq!(resumed.preview, "Saved user message");
assert!(resumed.turns.is_empty());
timeout(
DEFAULT_READ_TIMEOUT,
primary.start_turn_and_wait_for_completion(TurnStartParams {
thread_id: conversation_id.clone(),
input: vec![UserInput::Text {
text: "bounded suffix user message".to_string(),
text_elements: Vec::new(),
}],
..Default::default()
}),
)
.await??;
timeout(DEFAULT_READ_TIMEOUT, primary.shutdown_gracefully()).await??;
let mut secondary = TestAppServer::builder()
.with_codex_home(codex_home.path())
.without_auto_env()
.build_initialized()
.await?;
let resume_id = secondary
.send_thread_resume_request(ThreadResumeParams {
thread_id: conversation_id.clone(),
path: Some(path),
exclude_turns: true,
..Default::default()
})
.await?;
let ThreadResumeResponse {
thread: resumed, ..
} = timeout(DEFAULT_READ_TIMEOUT, secondary.read_response(resume_id)).await??;
assert_eq!(resumed.preview, "Saved user message");
assert!(resumed.turns.is_empty());
timeout(
DEFAULT_READ_TIMEOUT,
secondary.start_turn_and_wait_for_completion(TurnStartParams {
thread_id: conversation_id.clone(),
input: vec![UserInput::Text {
text: "resumed user message".to_string(),
text_elements: Vec::new(),
}],
..Default::default()
}),
)
.await??;
let state_db = StateRuntime::init(
codex_state::SqliteConfig::new_for_testing(codex_home.path().abs()),
"mock_provider".into(),
)
.await?;
let metadata = state_db
.get_thread(ThreadId::from_string(&conversation_id)?)
.await?
.expect("thread metadata should exist");
assert_eq!(
(
metadata.preview.as_deref(),
metadata.title.as_str(),
metadata.first_user_message.as_deref(),
),
(
Some("Saved user message"),
"Saved user message",
Some("Saved user message"),
)
);
Ok(())
}
codex-rs/thread-store/src/live_thread.rs
+14 -1
View File
@@ -151,7 +151,20 @@ impl LiveThread {
let thread_id = params.thread_id;
let should_load_history = params.history.is_none();
let include_archived = params.include_archived;
let mut metadata_sync = ThreadMetadataSync::for_resume(&params);
let metadata = if history_mode == ThreadHistoryMode::Paginated
&& let Some(local_store) = thread_store.as_any().downcast_ref::<LocalThreadStore>()
&& let Some(state_db) = local_store.state_db().await
{
state_db
.get_thread(thread_id)
.await
.map_err(|err| ThreadStoreError::Internal {
message: format!("failed to read thread metadata for {thread_id}: {err}"),
})?
} else {
None
};
let mut metadata_sync = ThreadMetadataSync::for_resume(&params, metadata.as_ref());
thread_store.resume_thread(params).await?;
if should_load_history {
match thread_store
codex-rs/thread-store/src/local/mod.rs
+83
View File
@@ -674,6 +674,89 @@ mod tests {
assert_eq!(metadata.title, "observed append");
}
#[tokio::test]
async fn paginated_resume_prefers_explicit_rollout_path_over_stale_sqlite_path() {
let home = TempDir::new().expect("temp dir");
let config = test_config(home.path());
let runtime = codex_state::StateRuntime::init(
config.sqlite.clone(),
config.default_model_provider_id.clone(),
)
.await
.expect("state db should initialize");
let store = Arc::new(LocalThreadStore::new(config, Some(runtime.clone())));
let uuid = uuid::Uuid::from_u128(228);
let thread_id = ThreadId::from_string(&uuid.to_string()).expect("valid thread id");
let rollout_path = write_session_file_with_history_mode(
home.path(),
"2025-01-03T12-00-00",
uuid,
ThreadHistoryMode::Paginated,
)
.expect("paginated session file");
let stale_rollout_path = home.path().join("stale-rollout.jsonl");
tokio::fs::write(&stale_rollout_path, "malformed session metadata\n")
.await
.expect("write stale rollout");
let mut builder = codex_state::ThreadMetadataBuilder::new(
thread_id,
stale_rollout_path,
chrono::Utc::now(),
SessionSource::Cli,
);
builder.history_mode = ThreadHistoryMode::Paginated;
builder.cwd = home.path().to_path_buf();
let mut metadata = builder.build("test-provider");
metadata.preview = Some("original user message".to_string());
metadata.first_user_message = Some("original user message".to_string());
metadata.title = "original user message".to_string();
runtime
.upsert_thread(&metadata)
.await
.expect("update stale sqlite rollout path");
let resumed = LiveThread::resume(
store,
ThreadHistoryMode::Paginated,
ResumeThreadParams {
thread_id,
rollout_path: Some(rollout_path.clone()),
history: Some(Arc::new(vec![user_message_item("bounded suffix")])),
include_archived: false,
metadata: ThreadPersistenceMetadata {
cwd: Some(home.path().to_path_buf()),
model_provider: "test-provider".to_string(),
memory_mode: ThreadMemoryMode::Enabled,
},
},
)
.await
.expect("resume paginated thread from its requested rollout");
assert_eq!(
resumed.local_rollout_path().await.expect("live rollout"),
Some(rollout_path)
);
resumed.shutdown().await.expect("shutdown resumed writer");
let metadata = runtime
.get_thread(thread_id)
.await
.expect("sqlite metadata read")
.expect("sqlite metadata");
assert_eq!(
(
metadata.preview.as_deref(),
metadata.title.as_str(),
metadata.first_user_message.as_deref(),
),
(
Some("original user message"),
"original user message",
Some("original user message"),
)
);
}
#[tokio::test]
async fn live_thread_does_not_derive_metadata_from_inherited_items() {
let home = TempDir::new().expect("temp dir");
codex-rs/thread-store/src/local/read_thread.rs
+74 -25
View File
@@ -51,7 +51,10 @@ pub(super) async fn read_thread(
{
let metadata_sandbox_policy = metadata.sandbox_policy.clone();
let mut thread = stored_thread_from_sqlite_metadata(store, metadata).await?;
if !params.include_history
// Paginated history may contain only a suffix, so its display metadata lives in SQLite.
// Legacy display metadata remains rollout-derived.
if thread.history_mode == ThreadHistoryMode::Legacy
&& !params.include_history
&& let Some(rollout_path) = thread.rollout_path.clone()
&& let Ok(mut rollout_thread) = read_thread_from_rollout_path(store, rollout_path).await
&& rollout_thread.thread_id == thread_id
@@ -115,27 +118,21 @@ pub(super) async fn read_thread_by_rollout_path(
include_history: bool,
) -> ThreadStoreResult<StoredThread> {
let path = resolve_requested_rollout_path(store, rollout_path).await?;
let mut thread = read_thread_from_rollout_path(store, path).await?;
let mut thread = read_thread_from_rollout_path(store, path.clone()).await?;
if !include_archived && thread.archived_at.is_some() {
return Err(ThreadStoreError::InvalidRequest {
message: format!("thread {} is archived", thread.thread_id),
});
}
if let Some(metadata) = read_sqlite_metadata(store, thread.thread_id).await {
if let Some(mut metadata) = read_sqlite_metadata(store, thread.thread_id).await {
if thread.history_mode == ThreadHistoryMode::Paginated {
thread.name = sqlite_thread_name(&metadata);
}
thread.recency_at = metadata.recency_at;
thread.is_pinned = metadata.is_pinned;
thread.git_info = if thread.history_mode == ThreadHistoryMode::Paginated {
// A paginated rollout only has the initial Git tuple. Do not turn an explicit SQLite
// clear back into the stale rollout value while reading by path.
git_info_from_parts(
metadata.git_sha,
metadata.git_branch,
metadata.git_origin_url,
)
// Paginated display metadata lives in SQLite because rollout history may be partial.
metadata.rollout_path = path;
metadata.archived_at = thread.archived_at;
thread = stored_thread_from_sqlite_metadata(store, metadata).await?;
} else {
thread.recency_at = metadata.recency_at;
thread.is_pinned = metadata.is_pinned;
let (fallback_sha, fallback_branch, fallback_origin_url) = match thread.git_info.take()
{
Some(info) => (
@@ -145,12 +142,12 @@ pub(super) async fn read_thread_by_rollout_path(
),
None => (None, None, None),
};
git_info_from_parts(
thread.git_info = git_info_from_parts(
metadata.git_sha.or(fallback_sha),
metadata.git_branch.or(fallback_branch),
metadata.git_origin_url.or(fallback_origin_url),
)
};
);
}
}
reject_paginated_history(&thread, include_history)?;
attach_history_if_requested(&mut thread, include_history).await?;
@@ -528,9 +525,15 @@ mod tests {
use chrono::Utc;
use codex_protocol::ThreadId;
use codex_protocol::items::TurnItem;
use codex_protocol::items::UserMessageItem;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::ItemCompletedEvent;
use codex_protocol::protocol::RolloutItem;
use codex_protocol::protocol::SandboxPolicy;
use codex_protocol::protocol::SessionSource;
use codex_protocol::protocol::ThreadHistoryMode;
use codex_protocol::user_input::UserInput;
use codex_state::ThreadMetadataBuilder;
use pretty_assertions::assert_eq;
use tempfile::TempDir;
@@ -803,7 +806,7 @@ mod tests {
}
#[tokio::test]
async fn read_paginated_thread_uses_sqlite_name_over_legacy_compatibility() {
async fn read_paginated_thread_prefers_sqlite_display_metadata() {
let home = TempDir::new().expect("temp dir");
let config = test_config(home.path());
let uuid = Uuid::from_u128(228);
@@ -815,6 +818,20 @@ mod tests {
ThreadHistoryMode::Paginated,
)
.expect("session file");
codex_rollout::append_rollout_item_to_path(
&rollout_path,
&RolloutItem::EventMsg(EventMsg::ItemCompleted(ItemCompletedEvent {
thread_id,
turn_id: "turn-1".to_string(),
item: TurnItem::UserMessage(UserMessageItem::new(&[UserInput::Text {
text: "Rollout user message".to_string(),
text_elements: Vec::new(),
}])),
completed_at_ms: 0,
})),
)
.await
.expect("append rollout user message");
let runtime = codex_state::StateRuntime::init(
config.sqlite.clone(),
config.default_model_provider_id.clone(),
@@ -822,14 +839,21 @@ mod tests {
.await
.expect("state db should initialize");
let store = LocalThreadStore::new(config.clone(), Some(runtime.clone()));
let mut builder =
ThreadMetadataBuilder::new(thread_id, rollout_path, Utc::now(), SessionSource::Cli);
let mut builder = ThreadMetadataBuilder::new(
thread_id,
rollout_path.clone(),
Utc::now(),
SessionSource::Cli,
);
builder.history_mode = ThreadHistoryMode::Paginated;
builder.model_provider = Some(config.default_model_provider_id.clone());
builder.model_provider = Some("Canonical SQLite provider".to_string());
builder.cwd = home.path().to_path_buf();
let mut metadata = builder.build(config.default_model_provider_id.as_str());
metadata.name = Some("Canonical SQLite name".to_string());
metadata.title = "Stale title name".to_string();
metadata.preview = Some("Canonical SQLite preview".to_string());
metadata.first_user_message = Some("Canonical SQLite first user message".to_string());
metadata.title = "Canonical SQLite derived title".to_string();
metadata.model = Some("Canonical SQLite model".to_string());
runtime
.upsert_thread(&metadata)
.await
@@ -838,7 +862,7 @@ mod tests {
.await
.expect("append legacy thread name");
let thread = store
let thread_by_id = store
.read_thread(ReadThreadParams {
thread_id,
include_archived: false,
@@ -846,8 +870,33 @@ mod tests {
})
.await
.expect("read thread");
let thread_by_path = store
.read_thread_by_rollout_path(
rollout_path,
/*include_archived*/ false,
/*include_history*/ false,
)
.await
.expect("read thread by rollout path");
assert_eq!(thread.name.as_deref(), Some("Canonical SQLite name"));
for thread in [thread_by_id, thread_by_path] {
assert_eq!(
(
thread.name.as_deref(),
thread.preview.as_str(),
thread.first_user_message.as_deref(),
thread.model_provider.as_str(),
thread.model.as_deref(),
),
(
Some("Canonical SQLite name"),
"Canonical SQLite preview",
Some("Canonical SQLite first user message"),
"Canonical SQLite provider",
Some("Canonical SQLite model"),
)
);
}
}
#[tokio::test]
codex-rs/thread-store/src/thread_metadata_sync.rs
+60 -19
View File
@@ -16,6 +16,7 @@ use codex_protocol::protocol::ThreadMemoryMode;
use codex_protocol::protocol::UserMessageEvent;
use codex_protocol::protocol::strip_user_message_prefix;
use codex_protocol::protocol::user_message_preview;
use codex_state::ThreadMetadata;
use crate::CreateThreadParams;
use crate::GitInfoPatch;
@@ -90,7 +91,10 @@ impl ThreadMetadataSync {
}
}
pub(crate) fn for_resume(params: &ResumeThreadParams) -> Self {
pub(crate) fn for_resume(
params: &ResumeThreadParams,
metadata: Option<&ThreadMetadata>,
) -> Self {
let mut sync = Self {
thread_id: params.thread_id,
cwd_seen: params
@@ -98,9 +102,12 @@ impl ThreadMetadataSync {
.cwd
.as_ref()
.is_some_and(|cwd| !cwd.as_os_str().is_empty()),
preview_seen: false,
first_user_message_seen: false,
title_seen: false,
preview_seen: metadata
.and_then(|metadata| metadata.preview.as_deref())
.is_some_and(|preview| !preview.is_empty()),
first_user_message_seen: metadata
.is_some_and(|metadata| metadata.first_user_message.is_some()),
title_seen: metadata.is_some_and(|metadata| !metadata.title.is_empty()),
pending_update: None,
pending_update_generation: 0,
last_touch_persisted_at: None,
@@ -426,13 +433,13 @@ mod tests {
#[test]
fn resume_history_keeps_derived_metadata_pending_until_applied() {
let thread_id = ThreadId::new();
let mut sync = ThreadMetadataSync::for_resume(&resume_params(
let mut sync = resume_sync(
thread_id,
vec![
RolloutItem::SessionMeta(session_meta(thread_id)),
RolloutItem::EventMsg(EventMsg::UserMessage(user_message("hello metadata"))),
],
));
);
let update = sync.take_pending_update().expect("pending metadata update");
assert_eq!(
@@ -462,7 +469,7 @@ mod tests {
#[test]
fn goal_update_sets_preview_without_overriding_existing_preview() {
let thread_id = ThreadId::new();
let sync = ThreadMetadataSync::for_resume(&resume_params(
let sync = resume_sync(
thread_id,
vec![
RolloutItem::EventMsg(EventMsg::ThreadGoalUpdated(goal_update(
@@ -471,26 +478,26 @@ mod tests {
))),
RolloutItem::EventMsg(EventMsg::UserMessage(user_message("first user text"))),
],
));
);
let update = sync.take_pending_update().expect("pending metadata update");
assert_eq!(update.patch.preview.as_deref(), Some("ship the refactor"));
assert_eq!(update.patch.title.as_deref(), Some("first user text"));
assert_eq!(
update.patch.first_user_message.as_deref(),
Some("first user text")
);
assert_eq!(update.patch.title.as_deref(), Some("first user text"));
}
#[test]
fn later_user_messages_do_not_emit_existing_preview_fields() {
let thread_id = ThreadId::new();
let mut sync = ThreadMetadataSync::for_resume(&resume_params(
let mut sync = resume_sync(
thread_id,
vec![RolloutItem::EventMsg(EventMsg::UserMessage(user_message(
"first user text",
)))],
));
);
let pending = sync.take_pending_update().expect("pending resume metadata");
sync.mark_pending_update_applied(&pending);
@@ -509,7 +516,7 @@ mod tests {
#[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 mut sync = resume_sync(thread_id, Vec::new());
let update = sync
.observe_appended_items(&[RolloutItem::EventMsg(EventMsg::ItemCompleted(
ItemCompletedEvent {
@@ -536,7 +543,7 @@ mod tests {
#[test]
fn metadata_irrelevant_items_coalesce_updated_at_touches() {
let thread_id = ThreadId::new();
let mut sync = ThreadMetadataSync::for_resume(&resume_params(thread_id, Vec::new()));
let mut sync = resume_sync(thread_id, Vec::new());
let item = RolloutItem::Compacted(CompactedItem {
message: "compacted".to_string(),
replacement_history: None,
@@ -566,7 +573,7 @@ mod tests {
#[test]
fn turn_start_advances_recency_at_without_changing_updated_at_behavior() {
let thread_id = ThreadId::new();
let mut sync = ThreadMetadataSync::for_resume(&resume_params(thread_id, Vec::new()));
let mut sync = resume_sync(thread_id, Vec::new());
let update = sync
.observe_appended_items(&[RolloutItem::EventMsg(EventMsg::TurnStarted(
@@ -587,7 +594,7 @@ mod tests {
#[test]
fn thread_settings_applied_updates_live_metadata() {
let thread_id = ThreadId::new();
let mut sync = ThreadMetadataSync::for_resume(&resume_params(thread_id, Vec::new()));
let mut sync = resume_sync(thread_id, Vec::new());
let permission_profile = PermissionProfile::workspace_write();
let cwd = std::env::current_dir()
.expect("current directory")
@@ -655,13 +662,13 @@ mod tests {
#[test]
fn resume_history_waits_for_append_before_flushing_metadata() {
let thread_id = ThreadId::new();
let mut sync = ThreadMetadataSync::for_resume(&resume_params(
let mut sync = resume_sync(
thread_id,
vec![
RolloutItem::SessionMeta(session_meta(thread_id)),
RolloutItem::EventMsg(EventMsg::UserMessage(user_message("hello metadata"))),
],
));
);
assert!(
sync.take_pending_update_for_existing_history().is_none(),
@@ -687,13 +694,13 @@ mod tests {
branch: Some("stale-branch".to_string()),
repository_url: None,
});
let sync = ThreadMetadataSync::for_resume(&resume_params(
let sync = resume_sync(
thread_id,
vec![
RolloutItem::SessionMeta(meta),
RolloutItem::EventMsg(EventMsg::UserMessage(user_message("hello metadata"))),
],
));
);
let update = sync.take_pending_update().expect("pending metadata update");
assert_eq!(update.patch.git_info, None);
@@ -701,6 +708,40 @@ mod tests {
assert_eq!(update.patch.preview.as_deref(), Some("hello metadata"));
}
#[test]
fn resume_preserves_existing_display_metadata_and_initializes_missing_title() {
let thread_id = ThreadId::new();
let mut metadata = codex_state::ThreadMetadataBuilder::new(
thread_id,
Default::default(),
Utc::now(),
SessionSource::Exec,
)
.build("test-provider");
metadata.preview = Some("goal-first preview".to_string());
metadata.first_user_message = Some("[Image]".to_string());
let params = resume_params(
thread_id,
vec![RolloutItem::EventMsg(EventMsg::UserMessage(user_message(
"first textual user message",
)))],
);
let sync = ThreadMetadataSync::for_resume(&params, Some(&metadata));
let update = sync.take_pending_update().expect("pending metadata update");
assert_eq!(update.patch.preview, None);
assert_eq!(update.patch.first_user_message, None);
assert_eq!(
update.patch.title.as_deref(),
Some("first textual user message")
);
}
fn resume_sync(thread_id: ThreadId, history: Vec<RolloutItem>) -> ThreadMetadataSync {
ThreadMetadataSync::for_resume(&resume_params(thread_id, history), /*metadata*/ None)
}
fn resume_params(thread_id: ThreadId, history: Vec<RolloutItem>) -> ResumeThreadParams {
ResumeThreadParams {
thread_id,