codex: remove fork history source wrapper

This commit is contained in:
Ahmed Ibrahim
2026-06-01 13:15:00 -07:00
parent e02916c6ed
commit 384d0951d5
12 changed files with 66 additions and 127 deletions
@@ -384,7 +384,6 @@ use codex_protocol::protocol::ConversationStartParams;
use codex_protocol::protocol::ConversationStartTransport;
use codex_protocol::protocol::ConversationTextParams;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::ForkedHistory;
#[cfg(test)]
use codex_protocol::protocol::GitInfo as CoreGitInfo;
use codex_protocol::protocol::InitialHistory;
@@ -35,7 +35,6 @@ use codex_external_agent_sessions::PendingSessionImport;
use codex_external_agent_sessions::prepare_validated_session_imports;
use codex_external_agent_sessions::record_imported_session;
use codex_protocol::ThreadId;
use codex_protocol::protocol::ForkedHistory;
use codex_protocol::protocol::InitialHistory;
use codex_thread_store::ThreadMetadataPatch;
use std::collections::HashSet;
@@ -306,10 +305,7 @@ impl ExternalAgentConfigRequestProcessor {
.thread_manager
.start_thread_with_options(StartThreadOptions {
config,
initial_history: InitialHistory::Forked(ForkedHistory {
source_thread_id: None,
history: rollout_items,
}),
initial_history: InitialHistory::Forked(rollout_items),
session_source: None,
thread_source: None,
dynamic_tools: Vec::new(),
@@ -2501,7 +2501,7 @@ impl ThreadRequestProcessor {
let include_turns = !exclude_turns;
let (thread_history, resume_source_thread) = match if let Some(history) = history {
self.resume_thread_from_history(&thread_id, history.as_slice())
self.resume_thread_from_history(history.as_slice())
.await
.map(|thread_history| (thread_history, None))
} else {
@@ -2928,20 +2928,18 @@ impl ThreadRequestProcessor {
async fn resume_thread_from_history(
&self,
thread_id: &str,
history: &[ResponseItem],
) -> Result<InitialHistory, JSONRPCErrorError> {
if history.is_empty() {
return Err(invalid_request("history must not be empty"));
}
Ok(InitialHistory::Forked(ForkedHistory {
source_thread_id: ThreadId::from_string(thread_id).ok(),
history: history
Ok(InitialHistory::Forked(
history
.iter()
.cloned()
.map(RolloutItem::ResponseItem)
.collect(),
}))
))
}
async fn resume_thread_from_rollout(
@@ -3121,14 +3119,14 @@ impl ThreadRequestProcessor {
}
}
}
InitialHistory::Forked(forked) => {
InitialHistory::Forked(items) => {
let mut thread = build_thread_from_snapshot(
thread_id,
session_id.clone(),
&config_snapshot,
Some(rollout_path.into()),
);
thread.preview = preview_from_rollout_items(&forked.history);
thread.preview = preview_from_rollout_items(items);
Ok(thread)
}
InitialHistory::New | InitialHistory::Cleared => Err(format!(
+1 -5
View File
@@ -21,7 +21,6 @@ use codex_protocol::error::Result as CodexResult;
use codex_protocol::models::ContentItem;
use codex_protocol::models::MessagePhase;
use codex_protocol::models::ResponseItem;
use codex_protocol::protocol::ForkedHistory;
use codex_protocol::protocol::InitialHistory;
use codex_protocol::protocol::InterAgentCommunication;
use codex_protocol::protocol::MultiAgentVersion;
@@ -521,10 +520,7 @@ impl AgentControl {
state
.fork_thread_with_source(
config.clone(),
InitialHistory::Forked(ForkedHistory {
source_thread_id: Some(parent_thread_id),
history: forked_rollout_items,
}),
InitialHistory::Forked(forked_rollout_items),
self.clone(),
session_source,
/*thread_source*/ Some(ThreadSource::Subagent),
+2 -6
View File
@@ -17,7 +17,6 @@ use codex_protocol::openai_models::ReasoningEffort as ReasoningEffortConfig;
use codex_protocol::protocol::AskForApproval;
use codex_protocol::protocol::Event;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::ForkedHistory;
use codex_protocol::protocol::InitialHistory;
use codex_protocol::protocol::MultiAgentVersion;
use codex_protocol::protocol::Op;
@@ -225,10 +224,7 @@ impl GuardianReviewSession {
let prior_review_count = state.prior_review_count;
let last_reviewed_transcript_cursor = state.last_reviewed_transcript_cursor;
state.last_committed_fork_snapshot = Some(GuardianReviewForkSnapshot {
initial_history: InitialHistory::Forked(ForkedHistory {
source_thread_id: Some(self.codex.session.conversation_id),
history: items,
}),
initial_history: InitialHistory::Forked(items),
prior_review_count,
last_reviewed_transcript_cursor,
});
@@ -481,7 +477,7 @@ impl GuardianReviewSessionManager {
let state = trunk.state.lock().await;
let snapshot = state.last_committed_fork_snapshot.as_ref()?;
match &snapshot.initial_history {
InitialHistory::Forked(forked) => Some(forked.history.clone()),
InitialHistory::Forked(items) => Some(items.clone()),
InitialHistory::New | InitialHistory::Cleared | InitialHistory::Resumed(_) => None,
}
}
+5 -5
View File
@@ -1245,20 +1245,20 @@ impl Session {
let _ = self.flush_rollout().await;
}
}
InitialHistory::Forked(forked) => {
self.apply_rollout_reconstruction(&turn_context, &forked.history)
InitialHistory::Forked(rollout_items) => {
self.apply_rollout_reconstruction(&turn_context, &rollout_items)
.await;
// Seed usage info from the recorded rollout so UIs can show token counts
// immediately on resume/fork.
if let Some(info) = Self::last_token_info_from_rollout(&forked.history) {
if let Some(info) = Self::last_token_info_from_rollout(&rollout_items) {
let mut state = self.state.lock().await;
state.set_token_info(Some(info));
}
// If persisting, persist all rollout items as-is (the store filters).
if !forked.history.is_empty() {
self.persist_rollout_items(&forked.history).await;
if !rollout_items.is_empty() {
self.persist_rollout_items(&rollout_items).await;
}
// Forked threads should remain file-backed immediately after startup.
+2 -9
View File
@@ -109,7 +109,6 @@ use codex_protocol::protocol::CodexErrorInfo;
use codex_protocol::protocol::CompactedItem;
use codex_protocol::protocol::ConversationAudioParams;
use codex_protocol::protocol::CreditsSnapshot;
use codex_protocol::protocol::ForkedHistory;
use codex_protocol::protocol::GranularApprovalConfig;
use codex_protocol::protocol::InitialHistory;
use codex_protocol::protocol::InterAgentCommunication;
@@ -2258,10 +2257,7 @@ async fn record_initial_history_reconstructs_forked_transcript() {
let (rollout_items, expected) = sample_rollout(&session, &turn_context).await;
session
.record_initial_history(InitialHistory::Forked(ForkedHistory {
source_thread_id: None,
history: rollout_items,
}))
.record_initial_history(InitialHistory::Forked(rollout_items))
.await;
let history = session.state.lock().await.clone_history();
@@ -2518,10 +2514,7 @@ async fn record_initial_history_forked_hydrates_previous_turn_settings() {
];
session
.record_initial_history(InitialHistory::Forked(ForkedHistory {
source_thread_id: None,
history: rollout_items,
}))
.record_initial_history(InitialHistory::Forked(rollout_items))
.await;
let history = session.clone_history().await;
+8 -22
View File
@@ -37,7 +37,6 @@ use codex_protocol::error::Result as CodexResult;
use codex_protocol::openai_models::ModelPreset;
use codex_protocol::protocol::Event;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::ForkedHistory;
use codex_protocol::protocol::InitialHistory;
use codex_protocol::protocol::MultiAgentVersion;
use codex_protocol::protocol::Op;
@@ -1497,7 +1496,6 @@ fn truncate_before_nth_user_message(
n: usize,
snapshot_state: &SnapshotTurnState,
) -> InitialHistory {
let source_thread_id = history.source_thread_id();
let items: Vec<RolloutItem> = history.get_rollout_items();
let user_positions = truncation::user_message_positions_in_rollout(&items);
let rolled = if snapshot_state.ends_mid_turn && n >= user_positions.len() {
@@ -1516,10 +1514,7 @@ fn truncate_before_nth_user_message(
if rolled.is_empty() {
InitialHistory::New
} else {
InitialHistory::Forked(ForkedHistory {
source_thread_id,
history: rolled,
})
InitialHistory::Forked(rolled)
}
}
@@ -1598,10 +1593,7 @@ fn fork_history_from_snapshot(
InitialHistory::New => InitialHistory::New,
InitialHistory::Cleared => InitialHistory::Cleared,
InitialHistory::Forked(history) => InitialHistory::Forked(history),
InitialHistory::Resumed(resumed) => InitialHistory::Forked(ForkedHistory {
source_thread_id: Some(resumed.conversation_id),
history: resumed.history,
}),
InitialHistory::Resumed(resumed) => InitialHistory::Forked(resumed.history),
};
if snapshot_state.ends_mid_turn {
append_interrupted_boundary(
@@ -1638,27 +1630,21 @@ fn append_interrupted_boundary(
history.push(RolloutItem::ResponseItem(marker));
}
history.push(aborted_event);
InitialHistory::Forked(ForkedHistory {
source_thread_id: None,
history,
})
InitialHistory::Forked(history)
}
InitialHistory::Forked(mut forked) => {
InitialHistory::Forked(mut history) => {
if let Some(marker) = interrupted_turn_history_marker(interrupted_marker) {
forked.history.push(RolloutItem::ResponseItem(marker));
history.push(RolloutItem::ResponseItem(marker));
}
forked.history.push(aborted_event);
InitialHistory::Forked(forked)
history.push(aborted_event);
InitialHistory::Forked(history)
}
InitialHistory::Resumed(mut resumed) => {
if let Some(marker) = interrupted_turn_history_marker(interrupted_marker) {
resumed.history.push(RolloutItem::ResponseItem(marker));
}
resumed.history.push(aborted_event);
InitialHistory::Forked(ForkedHistory {
source_thread_id: Some(resumed.conversation_id),
history: resumed.history,
})
InitialHistory::Forked(resumed.history)
}
}
}
+17 -22
View File
@@ -33,13 +33,6 @@ use wiremock::MockServer;
const TEST_INSTALLATION_ID: &str = "11111111-1111-4111-8111-111111111111";
fn forked_history(history: Vec<RolloutItem>) -> InitialHistory {
InitialHistory::Forked(ForkedHistory {
source_thread_id: None,
history,
})
}
fn user_msg(text: &str) -> ResponseItem {
ResponseItem::Message {
id: None,
@@ -103,7 +96,7 @@ fn truncates_before_requested_user_message() {
.map(RolloutItem::ResponseItem)
.collect();
let truncated = truncate_before_nth_user_message(
forked_history(initial),
InitialHistory::Forked(initial),
/*n*/ 1,
&SnapshotTurnState {
ends_mid_turn: false,
@@ -128,7 +121,7 @@ fn truncates_before_requested_user_message() {
.map(RolloutItem::ResponseItem)
.collect();
let truncated2 = truncate_before_nth_user_message(
forked_history(initial2.clone()),
InitialHistory::Forked(initial2.clone()),
/*n*/ 2,
&SnapshotTurnState {
ends_mid_turn: false,
@@ -152,7 +145,7 @@ fn out_of_range_truncation_drops_only_unfinished_suffix_mid_turn() {
];
let truncated = truncate_before_nth_user_message(
forked_history(items.clone()),
InitialHistory::Forked(items.clone()),
usize::MAX,
&SnapshotTurnState {
ends_mid_turn: true,
@@ -203,7 +196,7 @@ fn out_of_range_truncation_drops_pre_user_active_turn_prefix() {
RolloutItem::ResponseItem(assistant_msg("partial")),
];
let snapshot_state = snapshot_turn_state(&forked_history(items.clone()));
let snapshot_state = snapshot_turn_state(&InitialHistory::Forked(items.clone()));
assert_eq!(
snapshot_state,
SnapshotTurnState {
@@ -214,7 +207,7 @@ fn out_of_range_truncation_drops_pre_user_active_turn_prefix() {
);
let truncated = truncate_before_nth_user_message(
forked_history(items.clone()),
InitialHistory::Forked(items.clone()),
usize::MAX,
&snapshot_state,
);
@@ -241,7 +234,7 @@ async fn ignores_session_prefix_messages_when_truncating() {
.collect();
let truncated = truncate_before_nth_user_message(
forked_history(rollout_items),
InitialHistory::Forked(rollout_items),
/*n*/ 1,
&SnapshotTurnState {
ends_mid_turn: false,
@@ -989,7 +982,8 @@ async fn new_uses_active_provider_for_model_refresh() {
#[test]
fn interrupted_fork_snapshot_appends_interrupt_boundary() {
let committed_history = forked_history(vec![RolloutItem::ResponseItem(user_msg("hello"))]);
let committed_history =
InitialHistory::Forked(vec![RolloutItem::ResponseItem(user_msg("hello"))]);
assert_eq!(
serde_json::to_value(
@@ -1038,7 +1032,8 @@ fn interrupted_fork_snapshot_appends_interrupt_boundary() {
#[test]
fn disabled_interrupted_fork_snapshot_appends_only_interrupt_event() {
let committed_history = forked_history(vec![RolloutItem::ResponseItem(user_msg("hello"))]);
let committed_history =
InitialHistory::Forked(vec![RolloutItem::ResponseItem(user_msg("hello"))]);
assert_eq!(
serde_json::to_value(
@@ -1085,7 +1080,7 @@ fn disabled_interrupted_fork_snapshot_appends_only_interrupt_event() {
#[test]
fn interrupted_snapshot_is_not_mid_turn() {
let interrupted_history = forked_history(vec![
let interrupted_history = InitialHistory::Forked(vec![
RolloutItem::ResponseItem(user_msg("hello")),
RolloutItem::ResponseItem(assistant_msg("partial")),
RolloutItem::ResponseItem(contextual_user_interrupted_marker()),
@@ -1127,7 +1122,7 @@ fn multi_agent_v2_interrupted_marker_uses_developer_input_message() {
#[test]
fn completed_legacy_event_history_is_not_mid_turn() {
let completed_history = forked_history(vec![
let completed_history = InitialHistory::Forked(vec![
RolloutItem::EventMsg(EventMsg::UserMessage(UserMessageEvent {
client_id: None,
message: "hello".to_string(),
@@ -1155,7 +1150,7 @@ fn completed_legacy_event_history_is_not_mid_turn() {
#[test]
fn mixed_response_and_legacy_user_event_history_is_mid_turn() {
let mixed_history = forked_history(vec![
let mixed_history = InitialHistory::Forked(vec![
RolloutItem::ResponseItem(user_msg("hello")),
RolloutItem::EventMsg(EventMsg::UserMessage(UserMessageEvent {
client_id: None,
@@ -1204,7 +1199,7 @@ async fn interrupted_fork_snapshot_does_not_synthesize_turn_id_for_legacy_histor
let source = manager
.resume_thread_with_history(
config.clone(),
forked_history(vec![
InitialHistory::Forked(vec![
RolloutItem::ResponseItem(user_msg("hello")),
RolloutItem::ResponseItem(assistant_msg("partial")),
]),
@@ -1312,7 +1307,7 @@ async fn interrupted_fork_snapshot_preserves_explicit_turn_id() {
let source = manager
.resume_thread_with_history(
config.clone(),
forked_history(vec![
InitialHistory::Forked(vec![
RolloutItem::EventMsg(EventMsg::TurnStarted(TurnStartedEvent {
turn_id: "turn-explicit".to_string(),
trace_id: None,
@@ -1410,7 +1405,7 @@ async fn interrupted_fork_snapshot_uses_persisted_mid_turn_history_without_live_
let source = manager
.resume_thread_with_history(
config.clone(),
forked_history(vec![
InitialHistory::Forked(vec![
RolloutItem::ResponseItem(user_msg("hello")),
RolloutItem::ResponseItem(assistant_msg("partial")),
]),
@@ -1553,7 +1548,7 @@ async fn resumed_thread_keeps_paused_goal_paused() -> anyhow::Result<()> {
let source = manager
.resume_thread_with_history(
config.clone(),
forked_history(vec![RolloutItem::ResponseItem(user_msg("keep working"))]),
InitialHistory::Forked(vec![RolloutItem::ResponseItem(user_msg("keep working"))]),
auth_manager.clone(),
/*persist_extended_history*/ false,
/*parent_trace*/ None,
@@ -40,7 +40,6 @@ use codex_protocol::protocol::FileSystemAccessMode;
use codex_protocol::protocol::FileSystemPath;
use codex_protocol::protocol::FileSystemSandboxEntry;
use codex_protocol::protocol::FileSystemSandboxPolicy;
use codex_protocol::protocol::ForkedHistory;
use codex_protocol::protocol::InitialHistory;
use codex_protocol::protocol::InterAgentCommunication;
use codex_protocol::protocol::NetworkSandboxPolicy;
@@ -2722,17 +2721,14 @@ async fn resume_agent_restores_closed_agent_and_accepts_send_input() {
let thread = manager
.resume_thread_with_history(
config.clone(),
InitialHistory::Forked(ForkedHistory {
source_thread_id: None,
history: vec![RolloutItem::ResponseItem(ResponseItem::Message {
id: None,
role: "user".to_string(),
content: vec![ContentItem::InputText {
text: "materialized".to_string(),
}],
phase: None,
})],
}),
InitialHistory::Forked(vec![RolloutItem::ResponseItem(ResponseItem::Message {
id: None,
role: "user".to_string(),
content: vec![ContentItem::InputText {
text: "materialized".to_string(),
}],
phase: None,
})]),
AuthManager::from_auth_for_testing(CodexAuth::from_api_key("dummy")),
/*persist_extended_history*/ false,
/*parent_trace*/ None,
@@ -15,7 +15,6 @@ use codex_protocol::protocol::ConversationStartTransport;
use codex_protocol::protocol::ConversationTextParams;
use codex_protocol::protocol::ErrorEvent;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::ForkedHistory;
use codex_protocol::protocol::InitialHistory;
use codex_protocol::protocol::Op;
use codex_protocol::protocol::RealtimeAudioFrame;
@@ -1885,10 +1884,7 @@ async fn conversation_startup_context_current_thread_selects_many_turns_by_budge
.thread_manager
.resume_thread_with_history(
test.config.clone(),
InitialHistory::Forked(ForkedHistory {
source_thread_id: None,
history,
}),
InitialHistory::Forked(history),
auth_manager_from_auth(CodexAuth::from_api_key("dummy")),
/*persist_extended_history*/ false,
/*parent_trace*/ None,
+15 -27
View File
@@ -2334,18 +2334,12 @@ pub struct ResumedHistory {
pub rollout_path: Option<PathBuf>,
}
#[derive(Debug, Clone, Deserialize, Serialize, JsonSchema, TS)]
pub struct ForkedHistory {
pub source_thread_id: Option<ThreadId>,
pub history: Vec<RolloutItem>,
}
#[derive(Debug, Clone, Deserialize, Serialize, JsonSchema, TS)]
pub enum InitialHistory {
New,
Cleared,
Resumed(ResumedHistory),
Forked(ForkedHistory),
Forked(Vec<RolloutItem>),
}
impl InitialHistory {
@@ -2353,7 +2347,7 @@ impl InitialHistory {
match self {
InitialHistory::New | InitialHistory::Cleared => false,
InitialHistory::Resumed(resumed) => resumed.history.iter().any(&mut predicate),
InitialHistory::Forked(forked) => forked.history.iter().any(predicate),
InitialHistory::Forked(items) => items.iter().any(predicate),
}
}
@@ -2366,15 +2360,10 @@ impl InitialHistory {
_ => None,
})
}
InitialHistory::Forked(forked) => forked.source_thread_id,
}
}
pub fn source_thread_id(&self) -> Option<ThreadId> {
match self {
InitialHistory::New | InitialHistory::Cleared => None,
InitialHistory::Resumed(resumed) => Some(resumed.conversation_id),
InitialHistory::Forked(forked) => forked.source_thread_id,
InitialHistory::Forked(items) => items.iter().find_map(|item| match item {
RolloutItem::SessionMeta(meta_line) => Some(meta_line.meta.id),
_ => None,
}),
}
}
@@ -2382,7 +2371,7 @@ impl InitialHistory {
match self {
InitialHistory::New | InitialHistory::Cleared => None,
InitialHistory::Resumed(resumed) => session_cwd_from_items(&resumed.history),
InitialHistory::Forked(forked) => session_cwd_from_items(&forked.history),
InitialHistory::Forked(items) => session_cwd_from_items(items),
}
}
@@ -2390,7 +2379,7 @@ impl InitialHistory {
match self {
InitialHistory::New | InitialHistory::Cleared => Vec::new(),
InitialHistory::Resumed(resumed) => resumed.history.clone(),
InitialHistory::Forked(forked) => forked.history.clone(),
InitialHistory::Forked(items) => items.clone(),
}
}
@@ -2407,9 +2396,8 @@ impl InitialHistory {
})
.collect(),
),
InitialHistory::Forked(forked) => Some(
forked
.history
InitialHistory::Forked(items) => Some(
items
.iter()
.filter_map(|ri| match ri {
RolloutItem::EventMsg(ev) => Some(ev.clone()),
@@ -2430,7 +2418,7 @@ impl InitialHistory {
_ => None,
})
}
InitialHistory::Forked(forked) => forked.history.iter().find_map(|item| match item {
InitialHistory::Forked(items) => items.iter().find_map(|item| match item {
RolloutItem::SessionMeta(meta_line) => meta_line.meta.base_instructions.clone(),
_ => None,
}),
@@ -2446,7 +2434,7 @@ impl InitialHistory {
_ => None,
})
}
InitialHistory::Forked(forked) => forked.history.iter().find_map(|item| match item {
InitialHistory::Forked(items) => items.iter().find_map(|item| match item {
RolloutItem::SessionMeta(meta_line) => meta_line.meta.dynamic_tools.clone(),
_ => None,
}),
@@ -2459,9 +2447,9 @@ impl InitialHistory {
InitialHistory::Resumed(resumed) => {
multi_agent_version_from_items(&resumed.history, Some(resumed.conversation_id))
}
InitialHistory::Forked(forked) => forked.source_thread_id.and_then(|thread_id| {
multi_agent_version_from_items(&forked.history, Some(thread_id))
}),
InitialHistory::Forked(items) => {
multi_agent_version_from_items(items, /*thread_id*/ None)
}
}
}