diff --git a/src/config.rs b/src/config.rs index ee1ff80a..5d8f6fd5 100644 --- a/src/config.rs +++ b/src/config.rs @@ -944,7 +944,7 @@ pub fn codex_transport() -> CodexTransport { { return transport; } - if codex_request_compression() { + if codex_request_compression() && !codex_previous_response_id() { return CodexTransport::Http; } CodexTransport::Auto @@ -1589,6 +1589,10 @@ mod tests { .unwrap(); assert!(codex_request_compression()); assert_eq!(codex_transport(), CodexTransport::Http); + { + let _prev_resp_env = EnvGuard::set("CCP_CODEX_PREVIOUS_RESPONSE_ID", "1"); + assert_eq!(codex_transport(), CodexTransport::Auto); + } { let _transport_env = EnvGuard::set("CCP_CODEX_TRANSPORT", "auto"); assert_eq!(codex_transport(), CodexTransport::Auto); diff --git a/src/providers/codex/compaction.rs b/src/providers/codex/compaction.rs index 12033d02..3209a0fc 100644 --- a/src/providers/codex/compaction.rs +++ b/src/providers/codex/compaction.rs @@ -12,10 +12,8 @@ use super::translate::request::{ }; const RETAINED_MESSAGE_TOKEN_BUDGET: u64 = 20_000; -const STATE_TTL_MS: u64 = 30 * 60 * 1_000; +const STATE_TTL_MS: u64 = 24 * 60 * 60 * 1_000; const MAX_STATES: usize = 1_000; -const MAX_STATE_BYTES: usize = 4 * 1024 * 1024; -const MAX_TOTAL_STATE_BYTES: usize = 20_000_000; const MIN_PORTABLE_SUMMARY_BYTES: usize = 32; #[derive(Debug)] @@ -58,7 +56,6 @@ pub struct CompactionReplay { #[derive(Default)] struct CompactionRegistry { states: HashMap, - total_bytes: usize, } static REGISTRY: Mutex> = Mutex::new(None); @@ -128,11 +125,6 @@ pub fn store_compaction( state.native_history = native_history; state.phase = CompactionPhase::Unconfirmed; state.updated_at = now; - if state_size(session_id, state) > MAX_STATE_BYTES { - registry.states.remove(session_id); - update_total_bytes(registry); - return false; - } evict_states(registry, now); registry .states @@ -168,16 +160,10 @@ pub fn activate_compaction( } if state.model != model || !matches!(state.phase, CompactionPhase::Unconfirmed) { registry.states.remove(session_id); - update_total_bytes(registry); return false; } state.phase = CompactionPhase::Anchored { portable_summary }; state.updated_at = now; - if state_size(session_id, state) > MAX_STATE_BYTES { - registry.states.remove(session_id); - update_total_bytes(registry); - return false; - } evict_states(registry, now); registry.states.contains_key(session_id) } @@ -197,7 +183,6 @@ pub fn apply_compaction_replay( } if state.model != request.model { registry.states.remove(session_id); - update_total_bytes(registry); return None; } let CompactionPhase::Anchored { portable_summary } = &state.phase else { @@ -208,12 +193,10 @@ pub fn apply_compaction_replay( let summary_item = conversation.first()?; let Some(text) = message_text(summary_item) else { registry.states.remove(session_id); - update_total_bytes(registry); return None; }; if text.match_indices(portable_summary).count() != 1 { registry.states.remove(session_id); - update_total_bytes(registry); return None; } if conversation.len() == 1 { @@ -227,11 +210,6 @@ pub fn apply_compaction_replay( .chain(state.native_history.iter().cloned()) .chain(conversation[1..].iter().cloned()) .collect(); - if serialized_size(&replay.input) > MAX_STATE_BYTES { - registry.states.remove(session_id); - update_total_bytes(registry); - return None; - } state.updated_at = now; Some(CompactionReplay { request: replay, @@ -253,7 +231,6 @@ pub fn abort_compaction_attempt(session_id: Option<&str>, attempt: Option Vec { let mut input = input.to_vec(); - let remove_empty_message = if let Some(ResponsesInputItem::Message { role, content }) = - input.last_mut() + let target_idx = input.iter().rposition(|item| { + !matches!( + item, + ResponsesInputItem::Message { role, .. } if role == "developer" + ) + }); + if let Some(idx) = target_idx + && let Some(ResponsesInputItem::Message { role, content }) = input.get_mut(idx) && role == "user" { + let before_len = content.len(); content.retain(|part| { !matches!( part, @@ -288,12 +271,17 @@ fn without_compaction_instruction(input: &[ResponsesInputItem]) -> Vec String { text } -fn serialized_size(items: &[ResponsesInputItem]) -> usize { - serde_json::to_vec(items).map_or(usize::MAX, |value| value.len()) -} - -fn state_size(session_id: &str, state: &CompactionState) -> usize { - let summary_len = match &state.phase { - CompactionPhase::Preparing | CompactionPhase::Unconfirmed => 0, - CompactionPhase::Anchored { portable_summary } => portable_summary.len(), - }; - session_id.len() + state.model.len() + summary_len + serialized_size(&state.native_history) -} - fn now_ms() -> u64 { SystemTime::now() .duration_since(UNIX_EPOCH) @@ -520,20 +496,11 @@ fn now_ms() -> u64 { .as_millis() as u64 } -fn update_total_bytes(registry: &mut CompactionRegistry) { - registry.total_bytes = registry - .states - .iter() - .map(|(session_id, state)| state_size(session_id, state)) - .sum(); -} - fn evict_states(registry: &mut CompactionRegistry, now: u64) { registry .states .retain(|_, state| now.saturating_sub(state.updated_at) <= STATE_TTL_MS); - update_total_bytes(registry); - while registry.states.len() > MAX_STATES || registry.total_bytes > MAX_TOTAL_STATE_BYTES { + while registry.states.len() > MAX_STATES { let oldest = registry .states .iter() @@ -543,7 +510,6 @@ fn evict_states(registry: &mut CompactionRegistry, now: u64) { break; }; registry.states.remove(&oldest); - update_total_bytes(registry); } } @@ -974,4 +940,62 @@ mod tests { assert!(text.len() <= RETAINED_MESSAGE_TOKEN_BUDGET as usize * 4); assert!(std::str::from_utf8(text.as_bytes()).is_ok()); } + + #[test] + fn large_post_compaction_transcripts_beyond_4mb_preserve_replay() { + let _guard = TEST_REGISTRY_LOCK.lock().unwrap(); + clear_all_compactions_for_tests(); + let attempt = stored_compaction( + "session-large", + vec![ResponsesInputItem::Compaction { + encrypted_content: "opaque-compacted".to_string(), + }], + ); + assert!(activate_compaction( + Some("session-large"), + Some(attempt), + "gpt-5.6-sol", + &output(SUMMARY), + )); + + let large_encrypted = "x".repeat(5 * 1024 * 1024); + let next = request(json!([ + {"type":"message","role":"user","content":[{"type":"input_text","text":SUMMARY}]}, + {"type":"reasoning","id":"rs_1","encrypted_content":large_encrypted,"summary":[]}, + {"type":"message","role":"user","content":[{"type":"input_text","text":"continue"}]} + ])); + let replay = apply_compaction_replay(Some("session-large"), &next) + .expect("compaction replay should survive large post-compaction transcripts") + .request; + assert!(matches!( + replay.input[0], + ResponsesInputItem::Compaction { ref encrypted_content } + if encrypted_content == "opaque-compacted" + )); + } + + #[test] + fn without_compaction_instruction_strips_trailing_hook_developer_message() { + let input: Vec = serde_json::from_value(json!([ + {"type":"message","role":"user","content":[{"type":"input_text","text":"prior turn"}]}, + { + "type":"message", + "role":"user", + "content":[{ + "type":"input_text", + "text":"CRITICAL: Respond with TEXT ONLY. Do NOT call any tools.\n\nYour task is to create a detailed summary of the conversation so far." + }] + }, + { + "type":"message", + "role":"developer", + "content":[{"type":"input_text","text":"UserPromptSubmit hook output"}] + } + ])) + .unwrap(); + + let stripped = without_compaction_instruction(&input); + assert_eq!(stripped.len(), 1); + assert_eq!(message_text(&stripped[0]).as_deref(), Some("prior turn")); + } } diff --git a/src/providers/codex/continuation.rs b/src/providers/codex/continuation.rs index 6c645031..69d98da4 100644 --- a/src/providers/codex/continuation.rs +++ b/src/providers/codex/continuation.rs @@ -2,14 +2,14 @@ use std::collections::HashMap; use std::sync::Mutex; use std::sync::atomic::{AtomicU64, Ordering}; +use sha2::{Digest, Sha256}; + use crate::request_identity::ConversationIdentity; use super::translate::request::{ResponsesInputItem, ResponsesRequest}; const TTL_MS: u64 = 30 * 60 * 1000; const MAX_STATES: usize = 10_000; -const MAX_OWNER_TRANSCRIPT_BYTES: u64 = 2_000_000; -const MAX_TOTAL_TRANSCRIPT_BYTES: u64 = 20_000_000; #[derive(Clone)] struct ContinuationState { @@ -17,7 +17,6 @@ struct ContinuationState { socket_id: u64, prompt_signature: String, transcript: Vec, - transcript_bytes: u64, updated_at: u64, } @@ -30,7 +29,6 @@ struct OwnerState { #[derive(Default)] struct ContinuationRegistry { owners: HashMap, - total_transcript_bytes: u64, } static REGISTRY: Mutex> = Mutex::new(None); @@ -154,6 +152,60 @@ pub(crate) fn continuation_candidate_for_owner( continuation_candidate_inner(owner, body, enabled, "missing_identity") } +/// Derives a conversation owner when `x-claude-code-agent-id` is absent +/// (`ConversationIdentity::Main(session_id)`) by hashing `prompt_signature` +/// (including leading developer envelope items for Responses Lite) plus the +/// first user message in `body.input`. Explicit `ConversationIdentity::Agent` +/// headers are preserved unchanged. +pub(crate) fn resolve_continuation_owner( + owner: Option<&ConversationIdentity>, + body: &ResponsesRequest, +) -> Option { + match owner { + Some(ConversationIdentity::Main(session_id)) => Some(ConversationIdentity::Agent( + session_id.clone(), + synthetic_subagent_key(body), + )), + Some(ConversationIdentity::Agent(session_id, agent_id)) => Some( + ConversationIdentity::Agent(session_id.clone(), agent_id.clone()), + ), + None => None, + } +} + +pub(crate) fn synthetic_subagent_key(body: &ResponsesRequest) -> String { + let mut hasher = Sha256::new(); + hasher.update(prompt_signature(body).as_bytes()); + hasher.update(b"\0"); + for item in body + .input + .iter() + .take_while(|item| is_developer_envelope_item(item)) + { + let value = serde_json::to_value(item).unwrap_or_default(); + hasher.update(stable_json(&value).as_bytes()); + hasher.update(b"\0"); + } + hasher.update(b"\0"); + if let Some(first_user) = body + .input + .iter() + .find(|item| matches!(item, ResponsesInputItem::Message { role, .. } if role == "user")) + { + let value = serde_json::to_value(first_user).unwrap_or_default(); + hasher.update(stable_json(&value).as_bytes()); + } + format!("synthetic,{:x}", hasher.finalize()) +} + +fn is_developer_envelope_item(item: &ResponsesInputItem) -> bool { + match item { + ResponsesInputItem::AdditionalTools { .. } => true, + ResponsesInputItem::Message { role, .. } => role == "developer", + _ => false, + } +} + fn continuation_candidate_inner( owner: Option<&ConversationIdentity>, body: &ResponsesRequest, @@ -196,11 +248,6 @@ fn continuation_candidate_inner( let existing = registry.owners.remove(owner); let superseded_turn = existing.is_some(); let state = existing.and_then(|owner| owner.continuation); - if let Some(state) = &state { - registry.total_transcript_bytes = registry - .total_transcript_bytes - .saturating_sub(state.transcript_bytes); - } registry.owners.insert( owner.clone(), OwnerState { @@ -348,20 +395,11 @@ pub(crate) fn record_continuation_for_owner( let mut transcript: Vec = request_body.input.clone(); transcript.extend_from_slice(output_items); - let transcript_json = serde_json::to_string(&transcript).unwrap_or_default(); - let transcript_bytes = transcript_json.len() as u64; - - if transcript_bytes > MAX_OWNER_TRANSCRIPT_BYTES { - abort_continuation_inner(Some(owner), Some(turn_id)); - return; - } - let state = ContinuationState { response_id, socket_id, prompt_signature: prompt_signature(request_body), transcript, - transcript_bytes, updated_at: now_ms(), }; @@ -374,13 +412,8 @@ pub(crate) fn record_continuation_for_owner( }; if owner_state.current_turn != turn_id { return; - } - if let Some(existing) = owner_state.continuation.replace(state) { - registry.total_transcript_bytes = registry - .total_transcript_bytes - .saturating_sub(existing.transcript_bytes); - } - registry.total_transcript_bytes += transcript_bytes; + }; + owner_state.continuation = Some(state); evict_oldest(registry); } @@ -406,12 +439,8 @@ fn abort_continuation_inner(owner: Option<&ConversationIdentity>, turn_id: Optio .owners .get(owner) .is_some_and(|state| state.current_turn == turn_id) - && let Some(state) = registry.owners.remove(owner) - && let Some(continuation) = state.continuation { - registry.total_transcript_bytes = registry - .total_transcript_bytes - .saturating_sub(continuation.transcript_bytes); + registry.owners.remove(owner); } } @@ -501,13 +530,7 @@ pub(crate) fn clear_continuation_for_owner(owner: Option<&ConversationIdentity>) let Some(registry) = guard.as_mut() else { return; }; - if let Some(state) = registry.owners.remove(owner) - && let Some(continuation) = state.continuation - { - registry.total_transcript_bytes = registry - .total_transcript_bytes - .saturating_sub(continuation.transcript_bytes); - } + registry.owners.remove(owner); } #[deprecated(note = "use the owner-aware test helper for typed conversation ownership")] @@ -632,9 +655,7 @@ fn stable_json(value: &serde_json::Value) -> String { } fn evict_oldest(registry: &mut ContinuationRegistry) { - while registry.owners.len() > MAX_STATES - || registry.total_transcript_bytes > MAX_TOTAL_TRANSCRIPT_BYTES - { + while registry.owners.len() > MAX_STATES { let owner = registry .owners .iter() @@ -643,13 +664,7 @@ fn evict_oldest(registry: &mut ContinuationRegistry) { let Some(owner) = owner else { break; }; - if let Some(state) = registry.owners.remove(&owner) - && let Some(continuation) = state.continuation - { - registry.total_transcript_bytes = registry - .total_transcript_bytes - .saturating_sub(continuation.transcript_bytes); - } + registry.owners.remove(&owner); } } @@ -1196,4 +1211,84 @@ mod tests { &function_call("call_1", "Edit", "{\"a\":1}"), )); } + + #[test] + fn large_encrypted_reasoning_transcripts_keep_continuation() { + let _registry_guard = lock_registry(); + let owner = main_owner("session-large-transcript"); + let large_reasoning = ResponsesInputItem::Reasoning { + id: "rs_1".to_string(), + encrypted_content: "x".repeat(3_000_000), + summary: vec![], + }; + let request = request_with_input(vec![input("one")], None); + record_with_output_items( + &owner, + &request, + "resp_large", + std::slice::from_ref(&large_reasoning), + ); + + assert!(has_continuation_for_owner_for_tests(&owner)); + let next = request_with_input(vec![input("one"), large_reasoning, input("two")], None); + let reservation = continuation_candidate_for_owner(Some(&owner), &next, true); + assert_eq!(reservation.candidate().disabled_reason, None); + assert_eq!( + reservation.candidate().previous_response_id.as_deref(), + Some("resp_large") + ); + assert_eq!(reservation.candidate().input_delta_count, 1); + } + + #[test] + fn resolve_continuation_owner_derives_synthetic_key_when_agent_header_absent() { + let _registry_guard = lock_registry(); + let main = main_owner("session-shared"); + let explicit_agent = agent_owner("session-shared", "explicit-agent"); + + let main_turn1 = request_with_input( + vec![input("main task")], + Some(json!({"instructions": "You are the main coding agent."})), + ); + let main_turn2 = request_with_input( + vec![input("main task"), input("follow up")], + Some(json!({"instructions": "You are the main coding agent."})), + ); + let subagent_diff_first_user = request_with_input( + vec![input("subagent task")], + Some(json!({"instructions": "You are the main coding agent."})), + ); + let subagent_diff_prompt = request_with_input( + vec![input("main task")], + Some(json!({"instructions": "You are a research subagent."})), + ); + + let main_owner_1 = resolve_continuation_owner(Some(&main), &main_turn1).unwrap(); + let main_owner_2 = resolve_continuation_owner(Some(&main), &main_turn2).unwrap(); + let sub_owner_user = + resolve_continuation_owner(Some(&main), &subagent_diff_first_user).unwrap(); + let sub_owner_prompt = + resolve_continuation_owner(Some(&main), &subagent_diff_prompt).unwrap(); + let explicit_resolved = + resolve_continuation_owner(Some(&explicit_agent), &main_turn1).unwrap(); + + assert_eq!(main_owner_1, main_owner_2); + assert_ne!(main_owner_1, sub_owner_user); + assert_ne!(main_owner_1, sub_owner_prompt); + assert_ne!(sub_owner_user, sub_owner_prompt); + assert_eq!(explicit_resolved, explicit_agent); + assert_eq!(resolve_continuation_owner(None, &main_turn1), None); + + start_and_record(&main_owner_1, &main_turn1, "resp_main_1"); + start_and_record(&sub_owner_user, &subagent_diff_first_user, "resp_sub_1"); + + let main_continued = + continuation_candidate_for_owner(Some(&main_owner_2), &main_turn2, true); + assert_eq!( + main_continued.candidate().previous_response_id.as_deref(), + Some("resp_main_1") + ); + assert_eq!(main_continued.candidate().input_delta_count, 1); + assert!(has_continuation_for_owner_for_tests(&sub_owner_user)); + } } diff --git a/src/providers/codex/mod.rs b/src/providers/codex/mod.rs index 4e0f9d12..a4852342 100644 --- a/src/providers/codex/mod.rs +++ b/src/providers/codex/mod.rs @@ -44,7 +44,7 @@ use self::compaction::{ }; use self::continuation::{ ContinuationReservation, abort_continuation_for_owner, continuation_candidate_for_owner, - record_continuation_for_owner, + record_continuation_for_owner, resolve_continuation_owner, }; use self::count_tokens::count_translated_tokens; use self::translate::accumulate::accumulate_response_with_traffic; @@ -227,12 +227,15 @@ impl CodexProvider { && compact_boundary && let Some(session_id) = ctx.session_id.as_deref() { + let compaction_source = apply_compaction_replay(Some(session_id), &translated) + .map(|replay| replay.request) + .unwrap_or_else(|| translated.clone()); let attempt = begin_compaction(session_id, &translated.model); compaction_attempt = Some(attempt); log_compaction_event( "server_compaction_triggered", &ctx, - translated.input.len(), + compaction_source.input.len(), None, ); if let Some(monitor) = ctx.monitor.as_ref() { @@ -240,21 +243,23 @@ impl CodexProvider { } let mut compaction_ctx = ctx.clone(); compaction_ctx.monitor = None; - match request_compaction(self.client.as_ref(), &translated, &compaction_ctx).await { + match request_compaction(self.client.as_ref(), &compaction_source, &compaction_ctx) + .await + { Ok(native_history) => { if store_compaction(session_id, attempt, native_history) { log_compaction_event( "server_compaction_completed", &ctx, - translated.input.len(), + compaction_source.input.len(), None, ); } else { log_compaction_event( "server_compaction_failed", &ctx, - translated.input.len(), - Some("compaction state was superseded or exceeded the in-memory limit"), + compaction_source.input.len(), + Some("compaction state was superseded"), ); } } @@ -263,7 +268,7 @@ impl CodexProvider { log_compaction_event( "server_compaction_failed", &ctx, - translated.input.len(), + compaction_source.input.len(), Some(&error.to_string()), ); return map_codex_error_to_response(&error); @@ -273,7 +278,7 @@ impl CodexProvider { log_compaction_event( "server_compaction_failed", &ctx, - translated.input.len(), + compaction_source.input.len(), Some(&error.to_string()), ); } @@ -288,8 +293,10 @@ impl CodexProvider { // Check continuation let previous_response_id_enabled = config::codex_previous_response_id(); + let resolved_identity = + resolve_continuation_owner(conversation_identity.as_ref(), &translated); let continuation = continuation_candidate_for_owner( - conversation_identity.as_ref(), + resolved_identity.as_ref(), &translated, previous_response_id_enabled, ); diff --git a/src/providers/codex/translate/request.rs b/src/providers/codex/translate/request.rs index fea9a779..2673acc1 100644 --- a/src/providers/codex/translate/request.rs +++ b/src/providers/codex/translate/request.rs @@ -352,9 +352,14 @@ fn is_compact_message_content(content: &Value) -> bool { pub(crate) fn is_compact_messages_request(request: &MessagesRequest) -> bool { is_compact_request(flatten_system_text(request.extra.get("system")).as_deref()) - || request.messages.last().is_some_and(|message| { - message.role == "user" && is_compact_message_content(&message.content) - }) + || request + .messages + .iter() + .rev() + .find(|message| message.role != "system") + .is_some_and(|message| { + message.role == "user" && is_compact_message_content(&message.content) + }) } /// Reasoning-effort cap applied to compaction requests, or None when the @@ -1937,6 +1942,54 @@ mod tests { assert!(is_compact_messages_request(&req)); } + #[test] + fn compact_request_detected_when_hook_system_message_follows_user() { + let req: MessagesRequest = serde_json::from_value(json!({ + "model": "gpt-6-astra", + "messages": [ + {"role": "user", "content": "prior turn"}, + {"role": "assistant", "content": "prior reply"}, + { + "role": "user", + "content": concat!( + "CRITICAL: Respond with TEXT ONLY. Do NOT call any tools.\n\n", + "Your task is to create a detailed summary of the conversation so far, ", + "paying close attention to the user's explicit requests." + ) + }, + { + "role": "system", + "content": "UserPromptSubmit hook success" + } + ], + "system": "You are Claude Code." + })) + .unwrap(); + + assert!(is_compact_messages_request(&req)); + } + + #[test] + fn compact_message_not_detected_when_assistant_message_follows_user() { + let req: MessagesRequest = serde_json::from_value(json!({ + "model": "gpt-6-astra", + "messages": [ + { + "role": "user", + "content": concat!( + "CRITICAL: Respond with TEXT ONLY. Do NOT call any tools.\n", + "Your task is to create a detailed summary of the conversation so far." + ) + }, + {"role": "assistant", "content": "Summary: ..."} + ], + "system": "You are Claude Code." + })) + .unwrap(); + + assert!(!is_compact_messages_request(&req)); + } + #[test] fn compact_message_markers_must_be_in_final_user_message() { let req: MessagesRequest = serde_json::from_value(json!({ diff --git a/tests/codex_agent_continuation.rs b/tests/codex_agent_continuation.rs index 68ab1168..325dca3d 100644 --- a/tests/codex_agent_continuation.rs +++ b/tests/codex_agent_continuation.rs @@ -2115,3 +2115,84 @@ async fn tool_turn_with_noncanonical_arguments_continues_with_delta() { assert_eq!(harness.upstream.snapshot().len(), 2); harness.shutdown().await; } + +#[allow(clippy::await_holding_lock)] +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn subagents_without_agent_header_derive_synthetic_keys_and_interleave_independently() { + let _environment_lock = env_lock(); + let mut harness = TestHarness::start().await; + let case = unique("synthetic-subagent-key"); + let session = tagged(&case, "session"); + // Both main and subagent omit x-claude-code-agent-id (only x-claude-code-session-id). + let shared_headers = IdentityHeaders::main(&session); + + let m1 = tagged(&case, "main-1"); + let s1 = tagged(&case, "sub-1"); + let m2 = tagged(&case, "main-2"); + let s2 = tagged(&case, "sub-2"); + let rm1 = tagged(&case, "reply-main-1"); + let rs1 = tagged(&case, "reply-sub-1"); + let resp_m1 = tagged(&case, "resp-main-1"); + let resp_s1 = tagged(&case, "resp-sub-1"); + + // Start main turn 1 and subagent turn 1 concurrently under the same session header. + let main_req_1 = harness.start_request( + messages_body(false, vec![message("user", &m1)]), + shared_headers.clone(), + ); + let sub_req_1 = harness.start_request( + messages_body(false, vec![message("user", &s1)]), + shared_headers.clone(), + ); + let pending_one = harness.upstream.next_any_request(Some(&session)).await; + let pending_two = harness.upstream.next_any_request(Some(&session)).await; + let (m1_pending, s1_pending) = pending_pair(pending_one, pending_two, &m1, &s1); + assert_full_input(&m1_pending.captured, &[("user", &m1)]); + assert_full_input(&s1_pending.captured, &[("user", &s1)]); + assert_ne!( + m1_pending.captured.socket_ordinal, + s1_pending.captured.socket_ordinal + ); + + // Complete subagent first, then main, to verify completion order cannot supersede either. + let s1_capture = resolve_request(s1_pending, sub_req_1, &resp_s1, &rs1, false).await; + let m1_capture = resolve_request(m1_pending, main_req_1, &resp_m1, &rm1, false).await; + + let main_second = harness + .round_trip( + messages_body( + false, + vec![ + message("user", &m1), + message("assistant", &rm1), + message("user", &m2), + ], + ), + shared_headers.clone(), + &m2, + &tagged(&case, "resp-main-2"), + &tagged(&case, "reply-main-2"), + ) + .await; + let sub_second = harness + .round_trip( + messages_body( + false, + vec![ + message("user", &s1), + message("assistant", &rs1), + message("user", &s2), + ], + ), + shared_headers, + &s2, + &tagged(&case, "resp-sub-2"), + &tagged(&case, "reply-sub-2"), + ) + .await; + + assert_delta_input(&main_second, &resp_m1, m1_capture.socket_ordinal, &m2); + assert_delta_input(&sub_second, &resp_s1, s1_capture.socket_ordinal, &s2); + assert_eq!(harness.upstream.snapshot().len(), 4); + harness.shutdown().await; +} diff --git a/tests/smoke_cutover.rs b/tests/smoke_cutover.rs index a41fe423..5cb7b55a 100644 --- a/tests/smoke_cutover.rs +++ b/tests/smoke_cutover.rs @@ -1772,6 +1772,114 @@ async fn smoke_codex_http_server_compaction_replays_native_history() { clear_all_compactions_for_tests(); } +#[allow(clippy::await_holding_lock)] +#[tokio::test] +async fn smoke_codex_http_server_compaction_chains_across_multiple_compactions() { + let _guard = env_lock(); + let _full_lane_env = EnvGuard::set("CCP_CODEX_FULL_LANE", "0"); + clear_all_compactions_for_tests(); + let config = TempDir::new().unwrap(); + write_auth(config.path(), "codex"); + + let captured = Arc::new(Mutex::new(Vec::new())); + let upstream = spawn_http_upstream({ + let captured = captured.clone(); + move |body: Value| { + let is_compaction = body["input"].as_array().is_some_and(|input| { + input.last().and_then(|item| item["type"].as_str()) + == Some("compaction_trigger") + }); + let request_number = { + let mut requests = captured.lock().unwrap(); + requests.push(body); + requests.len() + }; + if is_compaction { + let encrypted = if request_number == 1 { + "opaque-history-1" + } else { + "opaque-history-2" + }; + format!( + "data: {{\"type\":\"response.output_item.done\",\"output_index\":0,\"item\":{{\"type\":\"compaction\",\"encrypted_content\":\"{encrypted}\"}}}}\n\n\ + data: {{\"type\":\"response.completed\",\"response\":{{\"id\":\"resp_compact_{request_number}\",\"usage\":{{\"input_tokens\":100,\"output_tokens\":1}}}}}}\n\n" + ) + .into_bytes() + } else { + let text = match request_number { + 2 => "first portable summary with enough detail to anchor the compacted conversation", + 4 => "second portable summary with enough detail to anchor the second compaction", + _ => "compacted ok", + }; + format!( + "data: {{\"type\":\"response.output_item.added\",\"output_index\":0,\"item\":{{\"type\":\"message\",\"id\":\"msg_up\"}}}}\n\n\ + data: {{\"type\":\"response.output_text.delta\",\"output_index\":0,\"delta\":\"{text}\"}}\n\n\ + data: {{\"type\":\"response.output_item.done\",\"output_index\":0,\"item\":{{\"type\":\"message\"}}}}\n\n\ + data: {{\"type\":\"response.completed\",\"response\":{{\"id\":\"resp_{request_number}\",\"usage\":{{\"input_tokens\":5,\"output_tokens\":2}}}}}}\n\n" + ) + .into_bytes() + } + } + }) + .await; + + let _config_env = EnvGuard::set("CCP_CONFIG_DIR", config.path()); + let _base_url_env = EnvGuard::set("CCP_CODEX_BASE_URL", &upstream); + let _transport_env = EnvGuard::set("CCP_CODEX_TRANSPORT", "http"); + let _compaction_env = EnvGuard::set("CCP_CODEX_SERVER_COMPACTION", "1"); + + let compact_1 = call_messages_body(json!({ + "model": "gpt-5.6-sol", + "max_tokens": 64, + "system": "You are Claude Code.", + "messages": [ + {"role":"user","content":"turn one"}, + {"role":"user","content":"CRITICAL: Respond with TEXT ONLY. Do NOT call any tools.\n\nYour task is to create a detailed summary of the conversation so far, paying close attention to the user's explicit requests."} + ] + })) + .await; + assert_eq!(compact_1.status(), StatusCode::OK); + + let compact_2 = call_messages_body(json!({ + "model": "gpt-5.6-sol", + "max_tokens": 64, + "system": "You are Claude Code.", + "messages": [ + {"role":"user","content":"first portable summary with enough detail to anchor the compacted conversation"}, + {"role":"user","content":"turn two"}, + {"role":"user","content":"CRITICAL: Respond with TEXT ONLY. Do NOT call any tools.\n\nYour task is to create a detailed summary of the conversation so far, paying close attention to the user's explicit requests."} + ] + })) + .await; + assert_eq!(compact_2.status(), StatusCode::OK); + + let post_compact_2 = call_messages_body(json!({ + "model": "gpt-5.6-sol", + "max_tokens": 64, + "system": "You are Claude Code.", + "messages": [ + {"role":"user","content":"second portable summary with enough detail to anchor the second compaction"}, + {"role":"user","content":"turn three"} + ] + })) + .await; + assert_eq!(post_compact_2.status(), StatusCode::OK); + + let requests = captured.lock().unwrap(); + assert_eq!(requests.len(), 5); + // Request #3 (index 2) is the second compaction_trigger request; it must include opaque-history-1 + let second_compaction_input = requests[2]["input"].as_array().unwrap(); + assert!(second_compaction_input.iter().any(|item| { + item["type"] == "compaction" && item["encrypted_content"] == "opaque-history-1" + })); + // Request #5 (index 4) is the post-second-compaction turn; it must replay opaque-history-2 + let final_replay_input = requests[4]["input"].as_array().unwrap(); + assert!(final_replay_input.iter().any(|item| { + item["type"] == "compaction" && item["encrypted_content"] == "opaque-history-2" + })); + clear_all_compactions_for_tests(); +} + #[allow(clippy::await_holding_lock)] #[tokio::test] async fn smoke_codex_http_compaction_failure_preserves_portable_summary() {