core: persist initial context window metadata

Record the initial AutoCompactWindowIds window_id in session_meta.context_window so JSONL tailers can observe the first context window before any compaction item exists. Thread creation now requires an initial_window_id string, while rollout JSONL keeps that value under the context_window namespace without writing derived chain fields for the initial window.

Resume reconstruction derives window 0 state from that initial_window_id when no compaction window metadata exists. Fork reconstruction ignores copied parent session_meta.context_window values so the fork keeps its own fresh initial window id, and legacy compactions without window_number continue to use compaction-count fallback accounting.

Verification:
- just test -p codex-protocol
- just test -p codex-rollout recorder_materializes_on_flush_with_pending_items
- just test -p codex-core reconstruct_history_restores_initial_window_from_session_meta
- just test -p codex-core reconstruct_history_prefers_compacted_window_over_session_meta
- just test -p codex-core reconstruct_history_preserves_legacy_compaction_count_with_session_meta_window
- just test -p codex-core record_initial_history_forked_ignores_parent_session_meta_window
- just test -p codex-thread-store
- just test -p codex-state
- just test -p codex-app-server thread_read_returns_summary_without_turns
- just test -p codex-rollout persistence_metrics
- just fix -p codex-protocol -p codex-rollout -p codex-thread-store -p codex-core -p codex-app-server
- just fix -p codex-core -p codex-rollout
This commit is contained in:
Michael Bolin
2026-06-23 13:43:59 -07:00
parent 9fe689783d
commit 1ca31ab972
33 changed files with 354 additions and 17 deletions
@@ -215,6 +215,7 @@ impl ExternalAgentSessionImporter {
},
dynamic_tools: Vec::new(),
multi_agent_version: Some(MultiAgentVersion::V1),
initial_window_id: uuid::Uuid::now_v7().to_string(),
metadata: ThreadPersistenceMetadata {
cwd: Some(cwd.clone()),
model_provider: model_provider.clone(),
@@ -202,6 +202,7 @@ fn create_fake_rollout_with_source_and_parent_thread_id(
dynamic_tools: None,
memory_mode: None,
multi_agent_version: None,
context_window: None,
};
let payload = serde_json::to_value(SessionMetaLine {
meta,
@@ -289,6 +290,7 @@ pub fn create_fake_rollout_with_text_elements(
dynamic_tools: None,
memory_mode: None,
multi_agent_version: None,
context_window: None,
};
let payload = serde_json::to_value(SessionMetaLine {
meta,
@@ -131,6 +131,7 @@ async fn get_conversation_summary_by_thread_id_reads_pathless_store_thread() ->
base_instructions: BaseInstructions::default(),
dynamic_tools: Vec::new(),
multi_agent_version: None,
initial_window_id: Uuid::now_v7().to_string(),
metadata: ThreadPersistenceMetadata {
cwd: None,
model_provider: "test-provider".to_string(),
@@ -157,6 +157,7 @@ async fn thread_delete_with_non_local_thread_store_does_not_create_local_persist
base_instructions: BaseInstructions::default(),
dynamic_tools: Vec::new(),
multi_agent_version: None,
initial_window_id: Uuid::now_v7().to_string(),
metadata: ThreadPersistenceMetadata {
cwd: Some(codex_home.path().to_path_buf()),
model_provider: "mock_provider".to_string(),
@@ -1367,6 +1367,7 @@ async fn seed_pathless_store_thread(
base_instructions: BaseInstructions::default(),
dynamic_tools: Vec::new(),
multi_agent_version: None,
initial_window_id: Uuid::now_v7().to_string(),
metadata: ThreadPersistenceMetadata {
cwd: None,
model_provider: "test-provider".to_string(),
@@ -2072,6 +2072,7 @@ stream_max_retries = 0
dynamic_tools: None,
memory_mode: None,
multi_agent_version: None,
context_window: None,
};
std::fs::write(
&rollout_path,
@@ -218,6 +218,7 @@ async fn thread_unarchive_preserves_pathless_store_metadata() -> Result<()> {
base_instructions: BaseInstructions::default(),
dynamic_tools: Vec::new(),
multi_agent_version: None,
initial_window_id: Uuid::now_v7().to_string(),
metadata: ThreadPersistenceMetadata {
cwd: None,
model_provider: "test-provider".to_string(),
@@ -61,6 +61,7 @@ async fn write_rollout_with_user_event(dir: &Path, thread_id: ThreadId) -> io::R
dynamic_tools: None,
memory_mode: None,
multi_agent_version: None,
context_window: None,
},
git: None,
};
+7 -2
View File
@@ -10,6 +10,7 @@ use tracing::Instrument;
use tracing::debug_span;
use tracing::info_span;
use super::rollout_reconstruction::SessionMetaWindowRestore;
use crate::session::SteerInputError;
use crate::session::TurnInput;
use crate::session::session::Session;
@@ -526,8 +527,12 @@ pub async fn thread_rollback(sess: &Arc<Session>, sub_id: String, num_turns: u32
.into_iter()
.chain(std::iter::once(RolloutItem::EventMsg(rollback_msg.clone())))
.collect::<Vec<_>>();
sess.apply_rollout_reconstruction(turn_context.as_ref(), replay_items.as_slice())
.await;
sess.apply_rollout_reconstruction(
turn_context.as_ref(),
replay_items.as_slice(),
SessionMetaWindowRestore::Restore,
)
.await;
sess.services
.agent_control
.rollout_budget()
+18 -4
View File
@@ -236,6 +236,7 @@ pub(crate) use self::input_queue::InputQueueActivity;
pub(crate) use self::input_queue::TurnInput;
pub(crate) use self::input_queue::TurnInputQueue;
use self::review::spawn_review_thread;
use self::rollout_reconstruction::SessionMetaWindowRestore;
use self::session::AppServerClientMetadata;
use self::session::Session;
use self::session::SessionConfiguration;
@@ -1290,7 +1291,11 @@ impl Session {
let turn_context = self.new_default_turn().await;
let rollout_items = resumed_history.history;
let previous_turn_settings = self
.apply_rollout_reconstruction(&turn_context, &rollout_items)
.apply_rollout_reconstruction(
&turn_context,
&rollout_items,
SessionMetaWindowRestore::Restore,
)
.await;
// If resuming, warn when the last recorded model differs from the current one.
@@ -1328,8 +1333,12 @@ impl Session {
}
InitialHistory::Forked(rollout_items) => {
let turn_context = self.new_default_turn().await;
self.apply_rollout_reconstruction(&turn_context, &rollout_items)
.await;
self.apply_rollout_reconstruction(
&turn_context,
&rollout_items,
SessionMetaWindowRestore::Ignore,
)
.await;
// Seed usage info from the recorded rollout so UIs can show token counts
// immediately on resume/fork.
@@ -1366,6 +1375,7 @@ impl Session {
&self,
turn_context: &TurnContext,
rollout_items: &[RolloutItem],
session_meta_window_restore: SessionMetaWindowRestore,
) -> Option<PreviousTurnSettings> {
let rollout_reconstruction::RolloutReconstruction {
mut history,
@@ -1376,7 +1386,11 @@ impl Session {
previous_window_id,
window_id,
} = self
.reconstruct_history_from_rollout(turn_context, rollout_items)
.reconstruct_history_from_rollout_with_session_meta_window_restore(
turn_context,
rollout_items,
session_meta_window_restore,
)
.await;
// Keep the recorded rollout unchanged. Prepare its reconstructed history before
// installing it, so legacy images are processed once for this resume or fork and
@@ -1,5 +1,6 @@
use super::*;
use crate::context_manager::is_user_turn_boundary;
use codex_protocol::protocol::SessionContextWindow;
use uuid::Uuid;
// Return value of `Session::reconstruct_history_from_rollout`, bundling the rebuilt history with
@@ -23,6 +24,29 @@ struct ReconstructedWindow {
id: Option<Uuid>,
}
/// Controls whether reconstruction may seed the active context-window ids from
/// `SessionMeta.context_window`.
///
/// This only applies to the session-level initial-window fallback. Window ids
/// recorded on `CompactedItem`s are still replayed whenever present, because
/// they describe compaction boundaries that are part of the reconstructed
/// history.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(super) enum SessionMetaWindowRestore {
/// Use the current thread's session metadata as an initial-window fallback.
///
/// This is appropriate for resume and rollback replay, where the rollout
/// items belong to the thread whose runtime state is being reconstructed.
Restore,
/// Ignore session metadata while reconstructing copied rollout history.
///
/// This is appropriate for forks, where source rollout items may contain a
/// parent thread's initial window id but the fork already has its own fresh
/// initial window id. Compacted window metadata in the copied history still
/// remains eligible for replay.
Ignore,
}
#[derive(Debug, Default)]
enum TurnReferenceContextItem {
/// No `TurnContextItem` has been seen for this replay span yet.
@@ -103,16 +127,52 @@ fn finalize_active_segment<'a>(
}
impl Session {
#[cfg(test)]
pub(super) async fn reconstruct_history_from_rollout(
&self,
turn_context: &TurnContext,
rollout_items: &[RolloutItem],
) -> RolloutReconstruction {
self.reconstruct_history_from_rollout_with_session_meta_window_restore(
turn_context,
rollout_items,
SessionMetaWindowRestore::Restore,
)
.await
}
pub(super) async fn reconstruct_history_from_rollout_with_session_meta_window_restore(
&self,
turn_context: &TurnContext,
rollout_items: &[RolloutItem],
session_meta_window_restore: SessionMetaWindowRestore,
) -> RolloutReconstruction {
// Replay metadata should already match the shape of the future lazy reverse loader, even
// while history materialization still uses an eager bridge. Scan newest-to-oldest,
// stopping once a surviving replacement-history checkpoint and the required resume metadata
// are both known; then replay only the buffered surviving tail forward to preserve exact
// history semantics.
let has_legacy_compaction_without_window_number =
rollout_items.iter().any(|item| {
matches!(item, RolloutItem::Compacted(compacted) if compacted.window_number.is_none())
});
let initial_window = match session_meta_window_restore {
SessionMetaWindowRestore::Restore => {
if has_legacy_compaction_without_window_number {
None
} else {
rollout_items.iter().find_map(|item| match item {
RolloutItem::SessionMeta(session_meta) => session_meta
.meta
.context_window
.as_ref()
.and_then(reconstructed_window_from_session_context_window),
_ => None,
})
}
}
SessionMetaWindowRestore::Ignore => None,
};
let mut base_replacement_history: Option<&[ResponseItem]> = None;
let mut previous_turn_settings = None;
let mut reference_context_item = TurnReferenceContextItem::NeverSet;
@@ -348,7 +408,7 @@ impl Session {
reference_context_item
};
let window = window.unwrap_or(ReconstructedWindow {
let window = window.or(initial_window).unwrap_or(ReconstructedWindow {
number: fallback_window_number,
first_id: None,
previous_id: None,
@@ -371,3 +431,15 @@ fn parse_uuid_v7(value: &str) -> Option<Uuid> {
.ok()
.filter(|uuid| uuid.get_version_num() == 7)
}
fn reconstructed_window_from_session_context_window(
context_window: &SessionContextWindow,
) -> Option<ReconstructedWindow> {
let id = parse_uuid_v7(&context_window.window_id)?;
Some(ReconstructedWindow {
number: 0,
first_id: Some(id),
previous_id: None,
id: Some(id),
})
}
@@ -9,8 +9,12 @@ use codex_protocol::protocol::CompactedItem;
use codex_protocol::protocol::InitialHistory;
use codex_protocol::protocol::InterAgentCommunication;
use codex_protocol::protocol::ResumedHistory;
use codex_protocol::protocol::SessionContextWindow;
use codex_protocol::protocol::SessionMeta;
use codex_protocol::protocol::SessionMetaLine;
use pretty_assertions::assert_eq;
use std::path::PathBuf;
use uuid::Uuid;
fn user_message(text: &str) -> ResponseItem {
ResponseItem::Message {
@@ -909,6 +913,116 @@ async fn record_initial_history_resumed_does_not_seed_reference_context_item_aft
assert!(session.reference_context_item().await.is_none());
}
#[tokio::test]
async fn reconstruct_history_restores_initial_window_from_session_meta() {
let (session, turn_context) = make_session_and_context().await;
let thread_id = ThreadId::default();
let initial_window_id = Uuid::now_v7();
let rollout_items = vec![RolloutItem::SessionMeta(SessionMetaLine {
meta: SessionMeta {
session_id: thread_id.into(),
id: thread_id,
context_window: Some(SessionContextWindow {
window_id: initial_window_id.to_string(),
}),
..SessionMeta::default()
},
git: None,
})];
let reconstructed = session
.reconstruct_history_from_rollout(&turn_context, &rollout_items)
.await;
assert_eq!(reconstructed.window_number, 0);
assert_eq!(reconstructed.first_window_id, Some(initial_window_id));
assert_eq!(reconstructed.previous_window_id, None);
assert_eq!(reconstructed.window_id, Some(initial_window_id));
}
#[tokio::test]
async fn reconstruct_history_prefers_compacted_window_over_session_meta() {
let (session, turn_context) = make_session_and_context().await;
let thread_id = ThreadId::default();
let initial_window_id = Uuid::now_v7();
let compacted_first_window_id = Uuid::now_v7();
let compacted_previous_window_id = Uuid::now_v7();
let compacted_window_id = Uuid::now_v7();
let rollout_items = vec![
RolloutItem::SessionMeta(SessionMetaLine {
meta: SessionMeta {
session_id: thread_id.into(),
id: thread_id,
context_window: Some(SessionContextWindow {
window_id: initial_window_id.to_string(),
}),
..SessionMeta::default()
},
git: None,
}),
RolloutItem::Compacted(CompactedItem {
message: String::new(),
replacement_history: Some(Vec::new()),
window_number: Some(2),
first_window_id: Some(compacted_first_window_id.to_string()),
previous_window_id: Some(compacted_previous_window_id.to_string()),
window_id: Some(compacted_window_id.to_string()),
}),
];
let reconstructed = session
.reconstruct_history_from_rollout(&turn_context, &rollout_items)
.await;
assert_eq!(reconstructed.window_number, 2);
assert_eq!(
reconstructed.first_window_id,
Some(compacted_first_window_id)
);
assert_eq!(
reconstructed.previous_window_id,
Some(compacted_previous_window_id)
);
assert_eq!(reconstructed.window_id, Some(compacted_window_id));
}
#[tokio::test]
async fn reconstruct_history_preserves_legacy_compaction_count_with_session_meta_window() {
let (session, turn_context) = make_session_and_context().await;
let thread_id = ThreadId::default();
let initial_window_id = Uuid::now_v7();
let rollout_items = vec![
RolloutItem::SessionMeta(SessionMetaLine {
meta: SessionMeta {
session_id: thread_id.into(),
id: thread_id,
context_window: Some(SessionContextWindow {
window_id: initial_window_id.to_string(),
}),
..SessionMeta::default()
},
git: None,
}),
RolloutItem::Compacted(CompactedItem {
message: "legacy summary".to_string(),
replacement_history: None,
window_number: None,
first_window_id: None,
previous_window_id: None,
window_id: None,
}),
];
let reconstructed = session
.reconstruct_history_from_rollout(&turn_context, &rollout_items)
.await;
assert_eq!(reconstructed.window_number, 1);
assert_eq!(reconstructed.first_window_id, None);
assert_eq!(reconstructed.previous_window_id, None);
assert_eq!(reconstructed.window_id, None);
}
#[tokio::test]
async fn reconstruct_history_legacy_compaction_without_replacement_history_does_not_inject_current_initial_context()
{
+8 -1
View File
@@ -546,6 +546,7 @@ impl Session {
SessionId::from(thread_id)
}
});
let initial_auto_compact_window_ids = AutoCompactWindowIds::new_initial();
let agent_control = agent_control.with_session_id(
session_id,
config
@@ -585,6 +586,9 @@ impl Session {
},
dynamic_tools: session_configuration.dynamic_tools.clone(),
multi_agent_version: initial_multi_agent_version,
initial_window_id: initial_auto_compact_window_ids
.window_id
.to_string(),
metadata: ThreadPersistenceMetadata {
cwd: Some(config.cwd.to_path_buf()),
model_provider: config.model_provider_id.clone(),
@@ -898,7 +902,10 @@ impl Session {
session_configuration.thread_name = thread_name.clone();
validate_config_lock_if_configured(&session_configuration).await?;
export_config_lock_if_configured(&session_configuration, thread_id).await?;
let state = SessionState::new(session_configuration.clone());
let state = SessionState::new_with_auto_compact_window_ids(
session_configuration.clone(),
initial_auto_compact_window_ids,
);
let managed_network_requirements_configured = config
.config_layer_stack
.requirements_toml()
+30
View File
@@ -127,6 +127,7 @@ use codex_protocol::protocol::RealtimeVoice;
use codex_protocol::protocol::RealtimeVoicesList;
use codex_protocol::protocol::ResumedHistory;
use codex_protocol::protocol::RolloutItem;
use codex_protocol::protocol::SessionContextWindow;
use codex_protocol::protocol::SessionMeta;
use codex_protocol::protocol::SessionMetaLine;
use codex_protocol::protocol::SkillScope;
@@ -2607,6 +2608,33 @@ async fn record_initial_history_reconstructs_forked_transcript() {
assert_eq!(expected, history.raw_items());
}
#[tokio::test]
async fn record_initial_history_forked_ignores_parent_session_meta_window() {
let (session, _turn_context) = make_session_and_context().await;
let fork_initial_ids = session.state.lock().await.auto_compact_window_ids();
let parent_thread_id = ThreadId::default();
let parent_initial_window_id = Uuid::now_v7();
let rollout_items = vec![RolloutItem::SessionMeta(SessionMetaLine {
meta: SessionMeta {
session_id: parent_thread_id.into(),
id: parent_thread_id,
context_window: Some(SessionContextWindow {
window_id: parent_initial_window_id.to_string(),
}),
..SessionMeta::default()
},
git: None,
})];
session
.record_initial_history(InitialHistory::Forked(rollout_items))
.await;
let state = session.state.lock().await;
assert_eq!(state.auto_compact_window_number(), 0);
assert_eq!(state.auto_compact_window_ids(), fork_initial_ids);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn session_configured_reports_permission_profile_for_external_sandbox() -> anyhow::Result<()>
{
@@ -3835,6 +3863,7 @@ async fn attach_thread_persistence(session: &mut Session) -> PathBuf {
base_instructions: BaseInstructions::default(),
dynamic_tools: Vec::new(),
multi_agent_version: None,
initial_window_id: Uuid::now_v7().to_string(),
metadata: ThreadPersistenceMetadata {
cwd: Some(config.cwd.to_path_buf()),
model_provider: config.model_provider_id.clone(),
@@ -6684,6 +6713,7 @@ async fn shutdown_complete_does_not_append_to_thread_store_after_shutdown() {
base_instructions: BaseInstructions::default(),
dynamic_tools: Vec::new(),
multi_agent_version: None,
initial_window_id: Uuid::now_v7().to_string(),
metadata: ThreadPersistenceMetadata {
cwd: Some(config.cwd.to_path_buf()),
model_provider: config.model_provider_id.clone(),
+17 -6
View File
@@ -8,6 +8,17 @@ pub(crate) struct AutoCompactWindowIds {
pub(crate) window_id: Uuid,
}
impl AutoCompactWindowIds {
pub(crate) fn new_initial() -> Self {
let window_id = Uuid::now_v7();
Self {
first_window_id: window_id,
previous_window_id: None,
window_id,
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) struct AutoCompactWindowSnapshot {
pub(crate) prefill_input_tokens: Option<i64>,
@@ -34,15 +45,15 @@ pub(super) struct AutoCompactWindow {
}
impl AutoCompactWindow {
#[cfg(test)]
pub(super) fn new() -> Self {
let window_id = Uuid::now_v7();
Self::new_with_ids(AutoCompactWindowIds::new_initial())
}
pub(super) fn new_with_ids(ids: AutoCompactWindowIds) -> Self {
Self {
window_number: 0,
ids: AutoCompactWindowIds {
first_window_id: window_id,
previous_window_id: None,
window_id,
},
ids,
new_context_window_requested: false,
prefill_input_tokens: None,
token_budget_reminder_delivered: false,
+12 -1
View File
@@ -47,7 +47,18 @@ pub(crate) struct SessionState {
impl SessionState {
/// Create a new session state mirroring previous `State::default()` semantics.
#[cfg(test)]
pub(crate) fn new(session_configuration: SessionConfiguration) -> Self {
Self::new_with_auto_compact_window_ids(
session_configuration,
AutoCompactWindowIds::new_initial(),
)
}
pub(crate) fn new_with_auto_compact_window_ids(
session_configuration: SessionConfiguration,
auto_compact_window_ids: AutoCompactWindowIds,
) -> Self {
let history = ContextManager::new();
Self {
session_configuration,
@@ -57,7 +68,7 @@ impl SessionState {
mcp_dependency_prompted: HashSet::new(),
additional_context: AdditionalContextStore::default(),
previous_turn_settings: None,
auto_compact_window: AutoCompactWindow::new(),
auto_compact_window: AutoCompactWindow::new_with_ids(auto_compact_window_ids),
startup_prewarm: None,
current_time_reminder: CurrentTimeReminderState::default(),
active_connector_selection: HashSet::new(),
@@ -77,6 +77,7 @@ async fn write_rollout_with_user_event(dir: &Path, thread_id: ThreadId) -> io::R
dynamic_tools: None,
memory_mode: None,
multi_agent_version: None,
context_window: None,
},
git: None,
};
@@ -128,6 +129,7 @@ async fn write_rollout_with_meta_only(dir: &Path, thread_id: ThreadId) -> io::Re
dynamic_tools: None,
memory_mode: None,
multi_agent_version: None,
context_window: None,
},
git: None,
};
@@ -374,6 +374,7 @@ async fn backfill_scans_existing_rollouts() -> Result<()> {
dynamic_tools: None,
memory_mode: None,
multi_agent_version: None,
context_window: None,
},
git: None,
};
+16
View File
@@ -2925,6 +2925,18 @@ pub enum MultiAgentVersion {
V2,
}
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq, JsonSchema, TS)]
pub struct SessionContextWindow {
/// UUIDv7 identity of this context window.
pub window_id: String,
}
impl SessionContextWindow {
pub fn new(window_id: String) -> Self {
Self { window_id }
}
}
/// SessionMeta contains session-level data that doesn't correspond to a specific turn.
///
/// NOTE: There used to be an `instructions` field here, which stored user_instructions, but we
@@ -2971,6 +2983,9 @@ pub struct SessionMeta {
pub memory_mode: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub multi_agent_version: Option<MultiAgentVersion>,
/// Initial context-window identity for consumers that tail rollout JSONL before compaction.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub context_window: Option<SessionContextWindow>,
}
impl Default for SessionMeta {
@@ -2995,6 +3010,7 @@ impl Default for SessionMeta {
dynamic_tools: None,
memory_mode: None,
multi_agent_version: None,
context_window: None,
}
}
}
@@ -474,6 +474,7 @@ fn write_rollout(path: &std::path::Path, thread_id: ThreadId, message: &str) ->
dynamic_tools: None,
memory_mode: None,
multi_agent_version: None,
context_window: None,
},
git: None,
};
+3
View File
@@ -51,6 +51,7 @@ async fn extract_metadata_from_rollout_uses_session_meta() {
dynamic_tools: None,
memory_mode: None,
multi_agent_version: None,
context_window: None,
};
let session_meta_line = SessionMetaLine {
meta: session_meta,
@@ -107,6 +108,7 @@ async fn extract_metadata_from_rollout_returns_latest_memory_mode() {
dynamic_tools: None,
memory_mode: None,
multi_agent_version: None,
context_window: None,
};
let polluted_meta = SessionMeta {
memory_mode: Some("polluted".to_string()),
@@ -375,6 +377,7 @@ fn write_rollout_in_sessions_with_cwd(
dynamic_tools: None,
memory_mode: None,
multi_agent_version: None,
context_window: None,
};
let session_meta_line = SessionMetaLine {
meta: session_meta,
@@ -255,6 +255,7 @@ fn turn_item_type(item: &TurnItem) -> &'static str {
fn response_item_type(item: &ResponseItem) -> &'static str {
match item {
ResponseItem::AdditionalTools { .. } => "response.additional_tools",
ResponseItem::Message { .. } => "response.message",
ResponseItem::AgentMessage { .. } => "response.agent_message",
ResponseItem::Reasoning { .. } => "response.reasoning",
+16
View File
@@ -57,6 +57,7 @@ use codex_protocol::protocol::MultiAgentVersion;
use codex_protocol::protocol::ResumedHistory;
use codex_protocol::protocol::RolloutItem;
use codex_protocol::protocol::RolloutLine;
use codex_protocol::protocol::SessionContextWindow;
use codex_protocol::protocol::SessionMeta;
use codex_protocol::protocol::SessionMetaLine;
use codex_protocol::protocol::SessionSource;
@@ -91,6 +92,7 @@ pub enum RolloutRecorderParams {
base_instructions: BaseInstructions,
dynamic_tools: Vec<DynamicToolSpec>,
multi_agent_version: Option<MultiAgentVersion>,
initial_window_id: Option<String>,
},
Resume {
path: PathBuf,
@@ -178,6 +180,7 @@ impl RolloutRecorderParams {
base_instructions,
dynamic_tools,
multi_agent_version: None,
initial_window_id: None,
}
}
@@ -202,6 +205,17 @@ impl RolloutRecorderParams {
self
}
pub fn with_initial_window_id(mut self, initial_window_id: String) -> Self {
if let Self::Create {
initial_window_id: window_id,
..
} = &mut self
{
*window_id = Some(initial_window_id);
}
self
}
pub fn resume(path: PathBuf) -> Self {
Self::Resume { path }
}
@@ -715,6 +729,7 @@ impl RolloutRecorder {
base_instructions,
dynamic_tools,
multi_agent_version,
initial_window_id,
} => {
let log_file_info = precompute_log_file_info(config, conversation_id)?;
let path = log_file_info.path.clone();
@@ -752,6 +767,7 @@ impl RolloutRecorder {
},
memory_mode: (!config.generate_memories()).then_some("disabled".to_string()),
multi_agent_version,
context_window: initial_window_id.map(SessionContextWindow::new),
};
(None, Some(log_file_info), path, Some(session_meta))
+11 -1
View File
@@ -103,6 +103,7 @@ async fn state_db_init_backfills_before_returning() -> anyhow::Result<()> {
dynamic_tools: None,
memory_mode: None,
multi_agent_version: None,
context_window: None,
},
git: None,
};
@@ -375,6 +376,7 @@ async fn recorder_materializes_on_flush_with_pending_items() -> std::io::Result<
let config = test_config(home.path());
let session_id = SessionId::default();
let thread_id = ThreadId::new();
let initial_window_id = Uuid::now_v7().to_string();
let recorder = RolloutRecorder::new(
&config,
RolloutRecorderParams::new(
@@ -386,7 +388,8 @@ async fn recorder_materializes_on_flush_with_pending_items() -> std::io::Result<
BaseInstructions::default(),
Vec::new(),
)
.with_session_id(session_id),
.with_session_id(session_id)
.with_initial_window_id(initial_window_id.clone()),
)
.await?;
@@ -437,6 +440,13 @@ async fn recorder_materializes_on_flush_with_pending_items() -> std::io::Result<
panic!("expected session metadata in rollout");
};
assert_eq!(session_meta.meta.session_id, session_id);
assert_eq!(
session_meta
.meta
.context_window
.map(|window| window.window_id),
Some(initial_window_id)
);
let buffered_idx = text
.find("buffered-event")
.expect("buffered event in rollout");
@@ -43,6 +43,7 @@ fn write_rollout_with_metadata(path: &Path, thread_id: ThreadId) -> std::io::Res
dynamic_tools: None,
memory_mode: None,
multi_agent_version: None,
context_window: None,
},
git: None,
}),
+1
View File
@@ -176,6 +176,7 @@ fn write_rollout_with_user_message(
dynamic_tools: None,
memory_mode: None,
multi_agent_version: None,
context_window: None,
},
git: None,
}),
+1
View File
@@ -1290,6 +1290,7 @@ async fn test_updated_at_uses_file_mtime() -> Result<()> {
dynamic_tools: None,
memory_mode: None,
multi_agent_version: None,
context_window: None,
},
git: None,
}),
+2
View File
@@ -341,6 +341,7 @@ mod tests {
dynamic_tools: None,
memory_mode: None,
multi_agent_version: None,
context_window: None,
},
git: None,
}),
@@ -532,6 +533,7 @@ mod tests {
dynamic_tools: None,
memory_mode: None,
multi_agent_version: None,
context_window: None,
},
git: None,
}),
+2
View File
@@ -1975,6 +1975,7 @@ mod tests {
dynamic_tools: None,
memory_mode: Some("polluted".to_string()),
multi_agent_version: None,
context_window: None,
},
git: None,
})];
@@ -2037,6 +2038,7 @@ mod tests {
dynamic_tools: None,
memory_mode: None,
multi_agent_version: None,
context_window: None,
},
git: Some(GitInfo {
commit_hash: Some(codex_git_utils::GitSha::new("rollout-sha")),
+3
View File
@@ -10,6 +10,7 @@ use codex_protocol::ThreadId;
use codex_protocol::models::PermissionProfile;
use codex_protocol::protocol::AskForApproval;
use codex_protocol::protocol::RolloutItem;
use codex_protocol::protocol::SessionContextWindow;
use codex_protocol::protocol::SessionMeta;
use codex_protocol::protocol::SessionMetaLine;
use codex_protocol::protocol::ThreadMemoryMode;
@@ -120,6 +121,7 @@ mod tests {
base_instructions: BaseInstructions::default(),
dynamic_tools: Vec::new(),
multi_agent_version: None,
initial_window_id: uuid::Uuid::now_v7().to_string(),
metadata: ThreadPersistenceMetadata {
cwd: None,
model_provider: "test-provider".to_string(),
@@ -248,6 +250,7 @@ impl InMemoryThreadStore {
memory_mode: matches!(params.metadata.memory_mode, ThreadMemoryMode::Disabled)
.then_some("disabled".to_string()),
multi_agent_version: params.multi_agent_version,
context_window: Some(SessionContextWindow::new(params.initial_window_id.clone())),
..SessionMeta::default()
};
state
@@ -37,7 +37,8 @@ pub(super) async fn create_thread(
params.dynamic_tools,
)
.with_session_id(params.session_id)
.with_multi_agent_version(params.multi_agent_version),
.with_multi_agent_version(params.multi_agent_version)
.with_initial_window_id(params.initial_window_id),
)
.await
.map_err(|err| ThreadStoreError::Internal {
+1
View File
@@ -1133,6 +1133,7 @@ mod tests {
base_instructions: BaseInstructions::default(),
dynamic_tools: Vec::new(),
multi_agent_version: None,
initial_window_id: uuid::Uuid::now_v7().to_string(),
metadata: thread_metadata(),
}
}
+2
View File
@@ -85,6 +85,8 @@ pub struct CreateThreadParams {
pub dynamic_tools: Vec<DynamicToolSpec>,
/// Multi-agent runtime selected when the thread was created.
pub multi_agent_version: Option<MultiAgentVersion>,
/// Initial context-window identity captured when the thread was created.
pub initial_window_id: String,
/// Metadata captured for the newly created thread.
pub metadata: ThreadPersistenceMetadata,
}