Skip to content

[codex] Store compact window id in rollout #27264

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
pakrym-oai merged 3 commits into from
Jun 10, 2026
Merged

[codex] Store compact window id in rollout #27264

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
f99614b
Store compact window id in rollout
pakrym-oai Jun 9, 2026
39d4f7c
codex: fix CI failure on PR #27264
pakrym-oai Jun 10, 2026
124b7a6
codex: fix CI failure on PR #27264
pakrym-oai Jun 10, 2026
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
1 change: 1 addition & 0 deletions codex-rs/app-server-protocol/src/protocol/thread_history.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2939,6 +2939,7 @@ mod tests {
RolloutItem::Compacted(CompactedItem {
message: String::new(),
replacement_history: None,
window_id: None,
}),
RolloutItem::EventMsg(EventMsg::TurnComplete(TurnCompleteEvent {
turn_id: "turn-compact".into(),
Expand Down
1 change: 1 addition & 0 deletions codex-rs/core/src/agent/control_tests.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1097,6 +1097,7 @@ async fn spawn_agent_fork_strips_parent_usage_hints_from_compacted_history() {
RolloutItem::Compacted(CompactedItem {
message: String::new(),
replacement_history: Some(replacement_history),
window_id: None,
}),
RolloutItem::TurnContext(turn_context.to_turn_context_item()),
RolloutItem::ResponseItem(spawn_agent_call(&parent_spawn_call_id)),
Expand Down
66 changes: 31 additions & 35 deletions codex-rs/core/src/client.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,6 @@ use std::sync::Arc;
use std::sync::Mutex as StdMutex;
use std::sync::OnceLock;
use std::sync::atomic::AtomicBool;
use std::sync::atomic::AtomicU64;
use std::sync::atomic::Ordering;

use codex_api::ApiError;
Expand Down Expand Up @@ -172,7 +171,6 @@ pub(crate) struct CompactConversationRequestSettings {
struct ModelClientState {
session_id: SessionId,
thread_id: ThreadId,
window_generation: AtomicU64,
installation_id: String,
provider: SharedModelProvider,
auth_env_telemetry: AuthEnvTelemetry,
Expand Down Expand Up @@ -345,7 +343,6 @@ impl ModelClient {
state: Arc::new(ModelClientState {
session_id,
thread_id,
window_generation: AtomicU64::new(0),
installation_id,
provider: model_provider,
auth_env_telemetry,
Expand Down Expand Up @@ -394,24 +391,6 @@ impl ModelClient {
self.state.provider.auth_manager()
}

pub(crate) fn set_window_generation(&self, window_generation: u64) {
self.state
.window_generation
.store(window_generation, Ordering::Relaxed);
self.store_cached_websocket_session(WebsocketSession::default());
}

pub(crate) fn advance_window_generation(&self) {
self.state.window_generation.fetch_add(1, Ordering::Relaxed);
self.store_cached_websocket_session(WebsocketSession::default());
}

pub(crate) fn current_window_id(&self) -> String {
let thread_id = self.state.thread_id;
let window_generation = self.state.window_generation.load(Ordering::Relaxed);
format!("{thread_id}:{window_generation}")
}

fn take_cached_websocket_session(&self) -> WebsocketSession {
let mut cached_websocket_session = self
.state
Expand Down Expand Up @@ -457,13 +436,15 @@ impl ModelClient {
///
/// The model selection and telemetry context are passed explicitly to keep `ModelClient`
/// session-scoped.
#[allow(clippy::too_many_arguments)]
pub(crate) async fn compact_conversation_history(
&self,
prompt: &Prompt,
model_info: &ModelInfo,
settings: CompactConversationRequestSettings,
session_telemetry: &SessionTelemetry,
compaction_trace: &CompactionTraceContext,
window_id: &str,
turn_metadata_header: Option<&str>,
) -> Result<Vec<ResponseItem>> {
if prompt.input.is_empty() {
Expand All @@ -488,6 +469,7 @@ impl ModelClient {
settings.effort,
settings.summary,
settings.service_tier,
window_id,
)?;
let ResponsesApiRequest {
model,
Expand Down Expand Up @@ -522,7 +504,7 @@ impl ModelClient {
/*turn_state*/ None,
parse_turn_metadata_header(turn_metadata_header).as_ref(),
));
extra_headers.extend(self.build_responses_identity_headers());
extra_headers.extend(self.build_responses_identity_headers(Some(window_id)));
extra_headers.extend(build_session_headers(
Some(self.state.session_id.to_string()),
Some(self.state.thread_id.to_string()),
Expand Down Expand Up @@ -644,21 +626,24 @@ impl ModelClient {
extra_headers
}

fn build_responses_identity_headers(&self) -> ApiHeaderMap {
fn build_responses_identity_headers(&self, window_id: Option<&str>) -> ApiHeaderMap {
let mut extra_headers = self.build_subagent_headers();
if let Some(parent_thread_id) = parent_thread_id_header_value(self.state.parent_thread_id)
&& let Ok(val) = HeaderValue::from_str(&parent_thread_id)
{
extra_headers.insert(X_CODEX_PARENT_THREAD_ID_HEADER, val);
}
if let Ok(val) = HeaderValue::from_str(&self.current_window_id()) {
if let Some(window_id) = window_id
&& let Ok(val) = HeaderValue::from_str(window_id)
{
extra_headers.insert(X_CODEX_WINDOW_ID_HEADER, val);
}
extra_headers
}

fn build_ws_client_metadata(
&self,
window_id: &str,
turn_metadata_header: Option<&str>,
use_responses_lite: bool,
) -> HashMap<String, String> {
Expand All @@ -667,10 +652,7 @@ impl ModelClient {
X_CODEX_INSTALLATION_ID_HEADER.to_string(),
self.state.installation_id.clone(),
);
client_metadata.insert(
X_CODEX_WINDOW_ID_HEADER.to_string(),
self.current_window_id(),
);
client_metadata.insert(X_CODEX_WINDOW_ID_HEADER.to_string(), window_id.to_string());
if let Some(subagent) = subagent_header_value(&self.state.session_source) {
client_metadata.insert(X_OPENAI_SUBAGENT_HEADER.to_string(), subagent);
}
Expand Down Expand Up @@ -752,6 +734,7 @@ impl ModelClient {
}
}

#[allow(clippy::too_many_arguments)]
fn build_responses_request(
&self,
provider: &codex_api::Provider,
Expand All @@ -760,6 +743,7 @@ impl ModelClient {
effort: Option<ReasoningEffortConfig>,
summary: ReasoningSummaryConfig,
service_tier: Option<String>,
window_id: &str,
) -> Result<ResponsesApiRequest> {
let instructions = &prompt.base_instructions.text;
let input = prompt.get_formatted_input();
Expand Down Expand Up @@ -807,10 +791,7 @@ impl ModelClient {
X_CODEX_INSTALLATION_ID_HEADER.to_string(),
self.state.installation_id.clone(),
),
(
X_CODEX_WINDOW_ID_HEADER.to_string(),
self.current_window_id(),
),
(X_CODEX_WINDOW_ID_HEADER.to_string(), window_id.to_string()),
])),
};
Ok(request)
Expand Down Expand Up @@ -955,7 +936,7 @@ impl ModelClient {
headers.insert("x-client-request-id", header_value);
}
headers.extend(build_session_headers(Some(session_id), Some(thread_id)));
headers.extend(self.build_responses_identity_headers());
headers.extend(self.build_responses_identity_headers(/*window_id*/ None));
if let Some(header_value) = self.generate_attestation_header_for().await {
headers.insert(X_OAI_ATTESTATION_HEADER, header_value);
}
Expand Down Expand Up @@ -998,6 +979,7 @@ impl ModelClientSession {
/// regardless of transport choice.
async fn build_responses_options(
&self,
window_id: &str,
turn_metadata_header: Option<&str>,
compression: Compression,
use_responses_lite: bool,
Expand All @@ -1015,7 +997,10 @@ impl ModelClientSession {
Some(&self.turn_state),
turn_metadata_header.as_ref(),
);
headers.extend(self.client.build_responses_identity_headers());
headers.extend(
self.client
.build_responses_identity_headers(Some(window_id)),
);
if let Some(header_value) = self.client.generate_attestation_header_for().await {
headers.insert(X_OAI_ATTESTATION_HEADER, header_value);
}
Expand Down Expand Up @@ -1115,7 +1100,6 @@ impl ModelClientSession {
pub async fn preconnect_websocket(
&mut self,
session_telemetry: &SessionTelemetry,
_model_info: &ModelInfo,
) -> std::result::Result<(), ApiError> {
if !self.client.responses_websocket_enabled() {
return Ok(());
Expand Down Expand Up @@ -1257,6 +1241,7 @@ impl ModelClientSession {
)]
async fn stream_responses_api(
&self,
window_id: &str,
prompt: &Prompt,
model_info: &ModelInfo,
session_telemetry: &SessionTelemetry,
Expand Down Expand Up @@ -1288,6 +1273,7 @@ impl ModelClientSession {
let compression = self.responses_request_compression(client_setup.auth.as_ref());
let mut options = self
.build_responses_options(
window_id,
turn_metadata_header,
compression,
model_info.use_responses_lite,
Expand All @@ -1301,6 +1287,7 @@ impl ModelClientSession {
effort.clone(),
summary,
service_tier.clone(),
window_id,
)?;
let inference_trace_attempt = inference_trace.start_attempt();
inference_trace_attempt.add_request_headers(&mut options.extra_headers);
Expand Down Expand Up @@ -1374,6 +1361,7 @@ impl ModelClientSession {
)]
async fn stream_responses_websocket(
&mut self,
window_id: &str,
prompt: &Prompt,
model_info: &ModelInfo,
session_telemetry: &SessionTelemetry,
Expand Down Expand Up @@ -1402,6 +1390,7 @@ impl ModelClientSession {

let options = self
.build_responses_options(
window_id,
turn_metadata_header,
compression,
model_info.use_responses_lite,
Expand All @@ -1414,10 +1403,12 @@ impl ModelClientSession {
effort.clone(),
summary,
service_tier.clone(),
window_id,
)?;
let mut ws_payload = ResponseCreateWsRequest {
client_metadata: response_create_client_metadata(
Some(self.client.build_ws_client_metadata(
window_id,
turn_metadata_header,
model_info.use_responses_lite,
)),
Expand Down Expand Up @@ -1553,6 +1544,7 @@ impl ModelClientSession {
#[allow(clippy::too_many_arguments)]
pub async fn prewarm_websocket(
&mut self,
window_id: &str,
prompt: &Prompt,
model_info: &ModelInfo,
session_telemetry: &SessionTelemetry,
Expand All @@ -1571,6 +1563,7 @@ impl ModelClientSession {
let disabled_trace = InferenceTraceContext::disabled();
match self
.stream_responses_websocket(
window_id,
prompt,
model_info,
session_telemetry,
Expand Down Expand Up @@ -1614,6 +1607,7 @@ impl ModelClientSession {
/// branches.
pub async fn stream(
&mut self,
window_id: &str,
prompt: &Prompt,
model_info: &ModelInfo,
session_telemetry: &SessionTelemetry,
Expand All @@ -1630,6 +1624,7 @@ impl ModelClientSession {
let request_trace = current_span_w3c_trace_context();
match self
.stream_responses_websocket(
window_id,
prompt,
model_info,
session_telemetry,
Expand All @@ -1651,6 +1646,7 @@ impl ModelClientSession {
}

self.stream_responses_api(
window_id,
prompt,
model_info,
session_telemetry,
Expand Down
6 changes: 3 additions & 3 deletions codex-rs/core/src/client_tests.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -291,13 +291,13 @@ fn build_ws_client_metadata_includes_window_lineage_and_turn_metadata() {
Some(parent_thread_id),
);

client.advance_window_generation();

let thread_id = client.state.thread_id;
let window_id = format!("{thread_id}:1");
let client_metadata = client.build_ws_client_metadata(
&window_id,
Some(r#"{"turn_id":"turn-123"}"#),
/*use_responses_lite*/ false,
);
let thread_id = client.state.thread_id;
assert_eq!(
client_metadata,
std::collections::HashMap::from([
Expand Down
6 changes: 5 additions & 1 deletion codex-rs/core/src/compact.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -228,14 +228,15 @@ async fn run_compact_task_inner_impl(
personality: turn_context.personality,
..Default::default()
};
let window_id = sess.services.model_client.current_window_id();
let window_id = sess.current_window_id().await;
let turn_metadata_header = turn_context
.turn_metadata_state
.current_header_value_for_compaction(&window_id, compaction_metadata);
let attempt_result = drain_to_completed(
&sess,
turn_context.as_ref(),
&mut client_session,
&window_id,
turn_metadata_header.as_deref(),
&prompt,
)
Expand Down Expand Up @@ -309,6 +310,7 @@ async fn run_compact_task_inner_impl(
let compacted_item = CompactedItem {
message: summary_text.clone(),
replacement_history: Some(new_history.clone()),
window_id: None,
};
sess.replace_compacted_history(new_history, reference_context_item, compacted_item)
.await;
Expand Down Expand Up @@ -579,11 +581,13 @@ async fn drain_to_completed(
sess: &Session,
turn_context: &TurnContext,
client_session: &mut ModelClientSession,
window_id: &str,
turn_metadata_header: Option<&str>,
prompt: &Prompt,
) -> CodexResult<()> {
let mut stream = client_session
.stream(
window_id,
prompt,
&turn_context.model_info,
&turn_context.session_telemetry,
Expand Down
4 changes: 3 additions & 1 deletion codex-rs/core/src/compact_remote.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -224,7 +224,7 @@ async fn run_remote_compact_task_inner_impl(
output_schema: None,
output_schema_strict: true,
};
let window_id = sess.services.model_client.current_window_id();
let window_id = sess.current_window_id().await;
let turn_metadata_header = turn_context
.turn_metadata_state
.current_header_value_for_compaction(&window_id, compaction_metadata);
Expand All @@ -245,6 +245,7 @@ async fn run_remote_compact_task_inner_impl(
},
&turn_context.session_telemetry,
&compaction_trace,
&window_id,
turn_metadata_header.as_deref(),
)
.await?;
Expand All @@ -263,6 +264,7 @@ async fn run_remote_compact_task_inner_impl(
let compacted_item = CompactedItem {
message: String::new(),
replacement_history: Some(new_history.clone()),
window_id: None,
};
// Install is the semantic boundary where the compact endpoint's output becomes live
// thread history. Keep it distinct from the later inference request so the reducer can
Expand Down
Loading
Loading