Extract persisted history types into a dedicated crate (#37871)

## What changed

- Add `codex-history` for model-history and persisted-rollout domain types, including `RolloutItem`, `RolloutLine`, `CompactedItem`, and initial/resumed history state.
- Re-export the persisted types from `codex-rollout` and update consumers to use the new crate boundary instead of `codex-protocol`.
- Preserve existing rollout serialization, including legacy numeric compacted-window IDs.

## Testing

- Add `codex-history` tests for rollout JSON round trips, compacted-history compatibility, persisted history modes, and multi-agent version selection.

GitOrigin-RevId: 944daa9297ddd231d3aebbdcb05fff4adf8b4e1b
This commit is contained in:
Adam Perry @ OpenAI authored and copyberry committed 2026-08-10 19:26:52 +00:00
1 parent a1c88e865d
commit 63002bdb26
127 files changed
+791 -673

No files matched your search

+344
View File
@@ -0,0 +1,344 @@
//! Model-history and persisted-rollout domain types.
use std::path::PathBuf;
use std::sync::Arc;
use codex_protocol::ThreadId;
use codex_protocol::capabilities::SelectedCapabilityRoot;
use codex_protocol::dynamic_tools::DynamicToolSpec;
use codex_protocol::models::BaseInstructions;
use codex_protocol::models::ContentItem;
use codex_protocol::models::ResponseItem;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::InterAgentCommunication;
use codex_protocol::protocol::MultiAgentVersion;
use codex_protocol::protocol::SessionMeta;
use codex_protocol::protocol::SessionMetaLine;
use codex_protocol::protocol::SessionSource;
use codex_protocol::protocol::ThreadHistoryMode;
use codex_protocol::protocol::ThreadSource;
use codex_protocol::protocol::TurnContextItem;
use codex_protocol::protocol::WorldStateItem;
use schemars::JsonSchema;
use serde::Deserialize;
use serde::Deserializer;
use serde::Serialize;
/// Persisted rollout item used by core history and rollout storage.
#[derive(Serialize, Deserialize, Debug, Clone, JsonSchema)]
#[serde(tag = "type", content = "payload", rename_all = "snake_case")]
pub enum RolloutItem {
SessionMeta(SessionMetaLine),
ResponseItem(ResponseItem),
InterAgentCommunication(InterAgentCommunication),
InterAgentCommunicationMetadata { trigger_turn: bool },
Compacted(CompactedItem),
TurnContext(TurnContextItem),
WorldState(WorldStateItem),
EventMsg(EventMsg),
}
#[derive(Serialize, Clone, Debug, PartialEq, JsonSchema)]
pub struct CompactedItem {
pub message: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub replacement_history: Option<Vec<ResponseItem>>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub window_number: Option<u64>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub first_window_id: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub previous_window_id: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub window_id: Option<String>,
}
impl From<CompactedItem> for ResponseItem {
fn from(value: CompactedItem) -> Self {
ResponseItem::Message {
id: None,
role: "assistant".to_string(),
content: vec![ContentItem::OutputText {
text: value.message,
}],
phase: None,
internal_chat_message_metadata_passthrough: None,
}
}
}
// Before window_number was introduced, the numeric window number was serialized as
// window_id. Accept that shape so existing rollouts remain resumable.
impl<'de> Deserialize<'de> for CompactedItem {
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
where
D: Deserializer<'de>,
{
let serialized = SerializedCompactedItem::deserialize(deserializer)?;
let mut window_number = serialized.window_number;
let window_id = match serialized.window_id {
Some(SerializedWindowId::Id(window_id)) => Some(window_id),
Some(SerializedWindowId::LegacyWindowNumber(legacy_window_number)) => {
window_number.get_or_insert(legacy_window_number);
None
}
None => None,
};
Ok(Self {
message: serialized.message,
replacement_history: serialized.replacement_history,
window_number,
first_window_id: serialized.first_window_id,
previous_window_id: serialized.previous_window_id,
window_id,
})
}
}
#[derive(Deserialize)]
struct SerializedCompactedItem {
message: String,
#[serde(default)]
replacement_history: Option<Vec<ResponseItem>>,
#[serde(default)]
window_number: Option<u64>,
#[serde(default)]
first_window_id: Option<String>,
#[serde(default)]
previous_window_id: Option<String>,
#[serde(default)]
window_id: Option<SerializedWindowId>,
}
#[derive(Deserialize)]
#[serde(untagged)]
enum SerializedWindowId {
Id(String),
LegacyWindowNumber(u64),
}
#[derive(Serialize, Deserialize, Clone, JsonSchema)]
pub struct RolloutLine {
pub timestamp: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub ordinal: Option<u64>,
#[serde(flatten)]
pub item: RolloutItem,
}
#[derive(Debug, Clone, Deserialize, Serialize)]
pub struct ResumedHistory {
pub conversation_id: ThreadId,
pub history: Arc<Vec<RolloutItem>>,
pub rollout_path: Option<PathBuf>,
}
#[derive(Debug, Clone, Deserialize, Serialize)]
pub enum InitialHistory {
New,
Cleared,
Resumed(ResumedHistory),
Forked(Vec<RolloutItem>),
}
impl InitialHistory {
pub fn scan_rollout_items(&self, mut predicate: impl FnMut(&RolloutItem) -> bool) -> bool {
match self {
Self::New | Self::Cleared => false,
Self::Resumed(resumed) => resumed.history.iter().any(&mut predicate),
Self::Forked(items) => items.iter().any(predicate),
}
}
pub fn forked_from_id(&self) -> Option<ThreadId> {
match self {
Self::New | Self::Cleared => None,
Self::Resumed(resumed) => resumed.history.iter().find_map(|item| match item {
RolloutItem::SessionMeta(meta_line) => meta_line.meta.forked_from_id,
_ => None,
}),
Self::Forked(items) => items.iter().find_map(|item| match item {
RolloutItem::SessionMeta(meta_line) => Some(meta_line.meta.id),
_ => None,
}),
}
}
pub fn session_cwd(&self) -> Option<PathBuf> {
match self {
Self::New | Self::Cleared => None,
Self::Resumed(resumed) => session_cwd_from_items(&resumed.history),
Self::Forked(items) => session_cwd_from_items(items),
}
}
pub fn get_rollout_items(&self) -> &[RolloutItem] {
match self {
Self::New | Self::Cleared => &[],
Self::Resumed(resumed) => &resumed.history,
Self::Forked(items) => items,
}
}
pub fn get_event_msgs(&self) -> Option<Vec<EventMsg>> {
match self {
Self::New | Self::Cleared => None,
Self::Resumed(resumed) => Some(
resumed
.history
.iter()
.filter_map(|item| match item {
RolloutItem::EventMsg(event) => Some(event.clone()),
_ => None,
})
.collect(),
),
Self::Forked(items) => Some(
items
.iter()
.filter_map(|item| match item {
RolloutItem::EventMsg(event) => Some(event.clone()),
_ => None,
})
.collect(),
),
}
}
pub fn get_base_instructions(&self) -> Option<BaseInstructions> {
match self {
Self::New | Self::Cleared => None,
Self::Resumed(resumed) => resumed.history.iter().find_map(|item| match item {
RolloutItem::SessionMeta(meta_line) => meta_line.meta.base_instructions.clone(),
_ => None,
}),
Self::Forked(items) => items.iter().find_map(|item| match item {
RolloutItem::SessionMeta(meta_line) => meta_line.meta.base_instructions.clone(),
_ => None,
}),
}
}
pub fn get_dynamic_tools(&self) -> Option<Vec<DynamicToolSpec>> {
match self {
Self::New | Self::Cleared => None,
Self::Resumed(resumed) => resumed.history.iter().find_map(|item| match item {
RolloutItem::SessionMeta(meta_line) => meta_line.meta.dynamic_tools.clone(),
_ => None,
}),
Self::Forked(items) => items.iter().find_map(|item| match item {
RolloutItem::SessionMeta(meta_line) => meta_line.meta.dynamic_tools.clone(),
_ => None,
}),
}
}
pub fn get_selected_capability_roots(&self) -> Vec<SelectedCapabilityRoot> {
self.get_session_meta()
.map(|meta| meta.selected_capability_roots.clone())
.unwrap_or_default()
}
pub fn get_multi_agent_version(&self) -> Option<MultiAgentVersion> {
match self {
Self::New | Self::Cleared => None,
Self::Resumed(resumed) => {
multi_agent_version_from_items(&resumed.history, Some(resumed.conversation_id))
}
Self::Forked(items) => multi_agent_version_from_items(items, /*thread_id*/ None),
}
}
pub fn get_history_mode(&self, default_history_mode: ThreadHistoryMode) -> ThreadHistoryMode {
match self {
Self::New | Self::Cleared => default_history_mode,
Self::Resumed(_) | Self::Forked(_) => self
.get_session_meta()
.map(|meta| meta.history_mode)
.unwrap_or(default_history_mode),
}
}
pub fn get_resumed_session_sources(&self) -> Option<(SessionSource, Option<ThreadSource>)> {
let meta = self.get_resumed_session_meta()?;
Some((meta.source.clone(), meta.thread_source.clone()))
}
pub fn get_resumed_thread_source(&self) -> Option<ThreadSource> {
self.get_resumed_session_meta()
.and_then(|meta| meta.thread_source.clone())
}
pub fn get_session_originator(&self) -> Option<String> {
self.get_session_meta()
.map(|meta| meta.originator.clone())
.filter(|originator| !originator.is_empty())
}
pub fn get_resumed_parent_thread_id(&self) -> Option<ThreadId> {
self.get_resumed_session_meta()
.and_then(|meta| meta.parent_thread_id)
}
fn get_session_meta(&self) -> Option<&SessionMeta> {
match self {
Self::New | Self::Cleared => None,
Self::Resumed(resumed) => resumed.history.iter().find_map(|item| match item {
RolloutItem::SessionMeta(meta_line) => Some(&meta_line.meta),
_ => None,
}),
Self::Forked(items) => items.iter().find_map(|item| match item {
RolloutItem::SessionMeta(meta_line) => Some(&meta_line.meta),
_ => None,
}),
}
}
fn get_resumed_session_meta(&self) -> Option<&SessionMeta> {
match self {
Self::New | Self::Cleared | Self::Forked(_) => None,
Self::Resumed(resumed) => resumed.history.iter().find_map(|item| match item {
RolloutItem::SessionMeta(meta_line) => Some(&meta_line.meta),
_ => None,
}),
}
}
}
fn session_cwd_from_items(items: &[RolloutItem]) -> Option<PathBuf> {
items.iter().find_map(|item| match item {
RolloutItem::SessionMeta(meta_line) => Some(meta_line.meta.cwd.clone()),
_ => None,
})
}
fn multi_agent_version_from_items(
items: &[RolloutItem],
thread_id: Option<ThreadId>,
) -> Option<MultiAgentVersion> {
let session_meta_version = items.iter().rev().find_map(|item| match item {
RolloutItem::SessionMeta(meta_line)
if thread_id.is_none_or(|thread_id| meta_line.meta.id == thread_id) =>
{
meta_line.meta.multi_agent_version
}
_ => None,
});
session_meta_version.or_else(|| {
items.iter().rev().find_map(|item| match item {
RolloutItem::TurnContext(turn_context) => turn_context.multi_agent_version,
RolloutItem::SessionMeta(_)
| RolloutItem::ResponseItem(_)
| RolloutItem::InterAgentCommunication(_)
| RolloutItem::InterAgentCommunicationMetadata { .. }
| RolloutItem::Compacted(_)
| RolloutItem::WorldState(_)
| RolloutItem::EventMsg(_) => None,
})
})
}
#[cfg(test)]
#[path = "tests.rs"]
mod tests;