Skip to content

Reject unsupported history modes when loading rollouts #34344

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
copyberry merged 1 commit into from
Jul 20, 2026
+45 −132
Merged

Reject unsupported history modes when loading rollouts #34344

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
19 changes: 6 additions & 13 deletions codex-rs/app-server/src/request_processors/thread_processor.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1317,7 +1317,7 @@ impl ThreadRequestProcessor {

listener_task_context
.thread_watch_manager
.upsert_thread_silently(thread.clone())
.upsert_thread_silently(&thread.id)
.instrument(tracing::info_span!(
"app_server.thread_start.upsert_thread",
otel.name = "app_server.thread_start.upsert_thread",
Expand Down Expand Up @@ -2986,14 +2986,9 @@ impl ThreadRequestProcessor {
let mut raw_events_enabled = false;
if let Ok(thread) = self.thread_manager.get_thread(thread_id).await {
let config_snapshot = thread.config_snapshot().await;
let loaded_thread = build_thread_from_snapshot(
thread_id,
thread.session_configured().session_id.to_string(),
thread.multi_agent_version(),
&config_snapshot,
thread.rollout_path(),
);
self.thread_watch_manager.upsert_thread(loaded_thread).await;
self.thread_watch_manager
.upsert_thread(&thread_id.to_string())
.await;
if let Some(parent_thread_id) = config_snapshot.parent_thread_id {
raw_events_enabled = self
.thread_state_manager
Expand Down Expand Up @@ -3287,9 +3282,7 @@ impl ThreadRequestProcessor {
thread.turns = materialized_turns;
}

self.thread_watch_manager
.upsert_thread(thread.clone())
.await;
self.thread_watch_manager.upsert_thread(&thread.id).await;

let thread_status = self
.thread_watch_manager
Expand Down Expand Up @@ -4238,7 +4231,7 @@ impl ThreadRequestProcessor {
thread.thread_source = config_snapshot.thread_source.clone().map(Into::into);

self.thread_watch_manager
.upsert_thread_silently(thread.clone())
.upsert_thread_silently(&thread.id)
.await;

thread.status = resolve_thread_status(
Expand Down
2 changes: 1 addition & 1 deletion codex-rs/app-server/src/request_processors/turn_processor.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1332,7 +1332,7 @@ impl TurnRequestProcessor {
if let Some(mut thread) = stored_thread {
thread.session_id = review_thread.session_configured().session_id.to_string();
self.thread_watch_manager
.upsert_thread_silently(thread.clone())
.upsert_thread_silently(&thread.id)
.await;
thread.status = resolve_thread_status(
self.thread_watch_manager
Expand Down
112 changes: 16 additions & 96 deletions codex-rs/app-server/src/thread_status.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,6 @@ use crate::outgoing_message::OutgoingEnvelope;
use crate::outgoing_message::OutgoingMessage;
use crate::outgoing_message::OutgoingMessageSender;
use codex_app_server_protocol::ServerNotification;
use codex_app_server_protocol::Thread;
use codex_app_server_protocol::ThreadActiveFlag;
use codex_app_server_protocol::ThreadStatus;
use codex_app_server_protocol::ThreadStatusChangedNotification;
Expand Down Expand Up @@ -89,16 +88,18 @@ impl ThreadWatchManager {
}
}

pub(crate) async fn upsert_thread(&self, thread: Thread) {
pub(crate) async fn upsert_thread(&self, thread_id: &str) {
let thread_id = thread_id.to_string();
self.mutate_and_publish(move |state| {
state.upsert_thread(thread.id, /*emit_notification*/ true)
state.upsert_thread(thread_id, /*emit_notification*/ true)
})
.await;
}

pub(crate) async fn upsert_thread_silently(&self, thread: Thread) {
pub(crate) async fn upsert_thread_silently(&self, thread_id: &str) {
let thread_id = thread_id.to_string();
self.mutate_and_publish(move |state| {
state.upsert_thread(thread.id, /*emit_notification*/ false)
state.upsert_thread(thread_id, /*emit_notification*/ false)
})
.await;
}
Expand Down Expand Up @@ -453,8 +454,6 @@ fn loaded_thread_status(runtime: &RuntimeFacts) -> ThreadStatus {
#[cfg(test)]
mod tests {
use super::*;
use codex_utils_absolute_path::test_support::PathBufExt;
use codex_utils_absolute_path::test_support::test_path_buf;
use pretty_assertions::assert_eq;
use tokio::time::Duration;
use tokio::time::timeout;
Expand All @@ -477,12 +476,7 @@ mod tests {
#[tokio::test]
async fn tracks_non_interactive_thread_status() {
let manager = ThreadWatchManager::new();
manager
.upsert_thread(test_thread(
NON_INTERACTIVE_THREAD_ID,
codex_app_server_protocol::SessionSource::AppServer,
))
.await;
manager.upsert_thread(NON_INTERACTIVE_THREAD_ID).await;

manager.note_turn_started(NON_INTERACTIVE_THREAD_ID).await;

Expand All @@ -499,12 +493,7 @@ mod tests {
#[tokio::test]
async fn status_updates_track_single_thread() {
let manager = ThreadWatchManager::new();
manager
.upsert_thread(test_thread(
INTERACTIVE_THREAD_ID,
codex_app_server_protocol::SessionSource::Cli,
))
.await;
manager.upsert_thread(INTERACTIVE_THREAD_ID).await;

manager.note_turn_started(INTERACTIVE_THREAD_ID).await;
assert_eq!(
Expand Down Expand Up @@ -612,12 +601,7 @@ mod tests {
#[tokio::test]
async fn system_error_sets_idle_flag_until_next_turn() {
let manager = ThreadWatchManager::new();
manager
.upsert_thread(test_thread(
INTERACTIVE_THREAD_ID,
codex_app_server_protocol::SessionSource::Cli,
))
.await;
manager.upsert_thread(INTERACTIVE_THREAD_ID).await;

manager.note_turn_started(INTERACTIVE_THREAD_ID).await;
manager.note_system_error(INTERACTIVE_THREAD_ID).await;
Expand All @@ -643,12 +627,7 @@ mod tests {
#[tokio::test]
async fn shutdown_marks_thread_not_loaded() {
let manager = ThreadWatchManager::new();
manager
.upsert_thread(test_thread(
INTERACTIVE_THREAD_ID,
codex_app_server_protocol::SessionSource::Cli,
))
.await;
manager.upsert_thread(INTERACTIVE_THREAD_ID).await;

manager.note_turn_started(INTERACTIVE_THREAD_ID).await;
manager.note_thread_shutdown(INTERACTIVE_THREAD_ID).await;
Expand All @@ -664,12 +643,7 @@ mod tests {
#[tokio::test]
async fn loaded_statuses_default_to_not_loaded_for_untracked_threads() {
let manager = ThreadWatchManager::new();
manager
.upsert_thread(test_thread(
INTERACTIVE_THREAD_ID,
codex_app_server_protocol::SessionSource::Cli,
))
.await;
manager.upsert_thread(INTERACTIVE_THREAD_ID).await;
manager.note_turn_started(INTERACTIVE_THREAD_ID).await;

let statuses = manager
Expand All @@ -694,12 +668,7 @@ mod tests {
#[tokio::test]
async fn has_running_turns_tracks_runtime_running_flag_only() {
let manager = ThreadWatchManager::new();
manager
.upsert_thread(test_thread(
INTERACTIVE_THREAD_ID,
codex_app_server_protocol::SessionSource::Cli,
))
.await;
manager.upsert_thread(INTERACTIVE_THREAD_ID).await;

assert_eq!(manager.running_turn_count().await, 0);

Expand All @@ -725,12 +694,7 @@ mod tests {
codex_analytics::AnalyticsEventsClient::disabled(),
)));

manager
.upsert_thread(test_thread(
INTERACTIVE_THREAD_ID,
codex_app_server_protocol::SessionSource::Cli,
))
.await;
manager.upsert_thread(INTERACTIVE_THREAD_ID).await;
assert_eq!(
recv_status_changed_notification(&mut outgoing_rx).await,
ThreadStatusChangedNotification {
Expand Down Expand Up @@ -768,12 +732,7 @@ mod tests {
codex_analytics::AnalyticsEventsClient::disabled(),
)));

manager
.upsert_thread_silently(test_thread(
INTERACTIVE_THREAD_ID,
codex_app_server_protocol::SessionSource::Cli,
))
.await;
manager.upsert_thread_silently(INTERACTIVE_THREAD_ID).await;

assert_eq!(
manager
Expand Down Expand Up @@ -803,18 +762,8 @@ mod tests {
#[tokio::test]
async fn status_watchers_receive_only_their_thread_updates() {
let manager = ThreadWatchManager::new();
manager
.upsert_thread(test_thread(
INTERACTIVE_THREAD_ID,
codex_app_server_protocol::SessionSource::Cli,
))
.await;
manager
.upsert_thread(test_thread(
NON_INTERACTIVE_THREAD_ID,
codex_app_server_protocol::SessionSource::AppServer,
))
.await;
manager.upsert_thread(INTERACTIVE_THREAD_ID).await;
manager.upsert_thread(NON_INTERACTIVE_THREAD_ID).await;
let interactive_thread_id = ThreadId::from_string(INTERACTIVE_THREAD_ID)
.expect("interactive thread id should parse");
let non_interactive_thread_id = ThreadId::from_string(NON_INTERACTIVE_THREAD_ID)
Expand Down Expand Up @@ -885,33 +834,4 @@ mod tests {
};
notification
}

fn test_thread(thread_id: &str, source: codex_app_server_protocol::SessionSource) -> Thread {
Thread {
id: thread_id.to_string(),
extra: None,
session_id: thread_id.to_string(),
forked_from_id: None,
parent_thread_id: None,
preview: String::new(),
ephemeral: false,
history_mode: Default::default(),
model_provider: "mock-provider".to_string(),
created_at: 0,
updated_at: 0,
recency_at: Some(0),
status: ThreadStatus::NotLoaded,
path: None,
cwd: test_path_buf("/tmp").abs(),
cli_version: "test".to_string(),
agent_nickname: None,
agent_role: None,
source,
can_accept_direct_input: None,
thread_source: None,
git_info: None,
name: None,
turns: Vec::new(),
}
}
}
44 changes: 22 additions & 22 deletions codex-rs/rollout/src/recorder.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -971,43 +971,43 @@ impl RolloutRecorder {
continue;
}
saw_non_empty_line = true;
let mut v: Value = match serde_json::from_str(&line) {
Ok(v) => v,
let mut value: Value = match serde_json::from_str(&line) {
Ok(value) => value,
Err(e) => {
warn!("failed to parse line as JSON: {line:?}, error: {e}");
parse_errors = parse_errors.saturating_add(1);
continue;
}
};
if strip_legacy_ghost_snapshot_rollout_line(&mut v) {
if strip_legacy_ghost_snapshot_rollout_line(&mut value) {
trace!("skipping legacy ghost_snapshot rollout line");
continue;
}
if thread_id.is_none() {
// The first SessionMeta belongs to this rollout. Later SessionMeta lines
// can be copied from fork history, so only validate unknown history modes
// before we have parsed the rollout's own SessionMeta.
reject_unknown_thread_history_mode(&value)?;
}

// Parse the rollout line structure
match serde_json::from_value::<RolloutLine>(v.clone()) {
Ok(rollout_line) => {
let item = rollout_line.item;
// Use the FIRST SessionMeta encountered in the file as the canonical
// thread id and main session information. Keep all items intact.
if thread_id.is_none()
&& let RolloutItem::SessionMeta(session_meta_line) = &item
{
thread_id = Some(session_meta_line.meta.id);
}
items.push(item);
}
let rollout_line = match serde_json::from_value::<RolloutLine>(value) {
Ok(rollout_line) => rollout_line,
Err(e) => {
if thread_id.is_none() {
// The first SessionMeta belongs to this rollout. Later SessionMeta lines
// can be copied from fork history, so only validate unknown history modes
// before we have parsed the rollout's own SessionMeta.
reject_unknown_thread_history_mode(&v)?;
}
trace!("failed to parse rollout line: {e}");
parse_errors = parse_errors.saturating_add(1);
continue;
}
};

let item = rollout_line.item;
// Use the FIRST SessionMeta encountered in the file as the canonical
// thread id and main session information. Keep all items intact.
if thread_id.is_none()
&& let RolloutItem::SessionMeta(session_meta_line) = &item
{
thread_id = Some(session_meta_line.meta.id);
}
items.push(item);
}
if !saw_non_empty_line {
return Err(IoError::other("empty session file"));
Expand Down
Loading