mirror of
https://github.com/pchuan98/codex.git
synced 2026-07-01 00:31:56 +08:00
330ae6a516
## Summary Resuming a persisted thread currently deep-clones its complete rollout history several times. `InitialHistory` is retained for the app-server response, copied into thread persistence, and copied again by read-only accessors. These copies scale with the complete rollout rather than the bounded model context and add measurable latency for large sessions. This change stores resumed rollout history in `Arc<Vec<RolloutItem>>`. Rollout loading wraps the parsed vector once, while app-server response construction, session initialization, and thread persistence share it through inexpensive `Arc` clones. Read-only history access now returns a borrowed slice, and fork paths use `Arc::unwrap_or_clone` where they genuinely need mutable ownership. Rollout reconstruction also consumes its temporary context instead of cloning the reconstructed model history. The serialized representation remains unchanged. In an artificial 123 MB rollout benchmark, sharing resumed history reduced cold resume latency by roughly 9–10%. The affected crates compile with their test targets, all 80 thread-store tests pass, and the Bazel dependency lock remains valid.
140 lines
5.0 KiB
Rust
140 lines
5.0 KiB
Rust
#![allow(clippy::unwrap_used)]
|
|
|
|
use std::sync::Arc;
|
|
|
|
use codex_core::NewThread;
|
|
use codex_login::CodexAuth;
|
|
use codex_protocol::ThreadId;
|
|
use codex_protocol::config_types::ModeKind;
|
|
use codex_protocol::config_types::ReasoningSummary;
|
|
use codex_protocol::protocol::EventMsg;
|
|
use codex_protocol::protocol::InitialHistory;
|
|
use codex_protocol::protocol::ResumedHistory;
|
|
use codex_protocol::protocol::RolloutItem;
|
|
use codex_protocol::protocol::TurnCompleteEvent;
|
|
use codex_protocol::protocol::TurnContextItem;
|
|
use codex_protocol::protocol::TurnStartedEvent;
|
|
use codex_protocol::protocol::UserMessageEvent;
|
|
use codex_protocol::protocol::WarningEvent;
|
|
use core::time::Duration;
|
|
use core_test_support::load_default_config_for_test;
|
|
use core_test_support::wait_for_event;
|
|
use tempfile::TempDir;
|
|
|
|
fn resume_history(
|
|
config: &codex_core::config::Config,
|
|
previous_model: &str,
|
|
rollout_path: &std::path::Path,
|
|
) -> InitialHistory {
|
|
let turn_id = "resume-warning-seed-turn".to_string();
|
|
let turn_ctx = TurnContextItem {
|
|
turn_id: Some(turn_id.clone()),
|
|
cwd: config.cwd.clone(),
|
|
workspace_roots: None,
|
|
current_date: None,
|
|
timezone: None,
|
|
approval_policy: config.permissions.approval_policy.value(),
|
|
sandbox_policy: config.legacy_sandbox_policy(),
|
|
permission_profile: None,
|
|
network: None,
|
|
file_system_sandbox_policy: None,
|
|
model: previous_model.to_string(),
|
|
comp_hash: None,
|
|
personality: None,
|
|
collaboration_mode: None,
|
|
multi_agent_version: None,
|
|
multi_agent_mode: None,
|
|
realtime_active: None,
|
|
effort: config.model_reasoning_effort.clone(),
|
|
summary: config
|
|
.model_reasoning_summary
|
|
.unwrap_or(ReasoningSummary::Auto),
|
|
};
|
|
|
|
InitialHistory::Resumed(ResumedHistory {
|
|
conversation_id: ThreadId::default(),
|
|
history: Arc::new(vec![
|
|
RolloutItem::EventMsg(EventMsg::TurnStarted(TurnStartedEvent {
|
|
turn_id: turn_id.clone(),
|
|
trace_id: None,
|
|
started_at: None,
|
|
model_context_window: None,
|
|
collaboration_mode_kind: ModeKind::Default,
|
|
})),
|
|
RolloutItem::EventMsg(EventMsg::UserMessage(UserMessageEvent {
|
|
client_id: None,
|
|
message: "seed".to_string(),
|
|
images: None,
|
|
local_images: vec![],
|
|
text_elements: vec![],
|
|
..Default::default()
|
|
})),
|
|
RolloutItem::TurnContext(turn_ctx),
|
|
RolloutItem::EventMsg(EventMsg::TurnComplete(TurnCompleteEvent {
|
|
turn_id,
|
|
last_agent_message: None,
|
|
completed_at: None,
|
|
duration_ms: None,
|
|
time_to_first_token_ms: None,
|
|
})),
|
|
]),
|
|
rollout_path: Some(rollout_path.to_path_buf()),
|
|
})
|
|
}
|
|
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn emits_warning_when_resumed_model_differs() {
|
|
// Arrange a config with a current model and a prior rollout recorded under a different model.
|
|
let home = TempDir::new().expect("tempdir");
|
|
let mut config = load_default_config_for_test(&home).await;
|
|
config.model = Some("current-model".to_string());
|
|
// Ensure cwd is absolute (the helper sets it to the temp dir already).
|
|
assert!(config.cwd.is_absolute());
|
|
|
|
let rollout_path = home.path().join("rollout.jsonl");
|
|
std::fs::write(&rollout_path, "").expect("create rollout placeholder");
|
|
|
|
let initial_history = resume_history(&config, "previous-model", &rollout_path);
|
|
|
|
let thread_manager = codex_core::test_support::thread_manager_with_models_provider(
|
|
CodexAuth::from_api_key("test"),
|
|
config.model_provider.clone(),
|
|
);
|
|
let auth_manager =
|
|
codex_core::test_support::auth_manager_from_auth(CodexAuth::from_api_key("test"));
|
|
|
|
// Act: resume the conversation.
|
|
let NewThread {
|
|
thread: conversation,
|
|
..
|
|
} = thread_manager
|
|
.resume_thread_with_history(
|
|
config.clone(),
|
|
initial_history,
|
|
auth_manager,
|
|
/*parent_trace*/ None,
|
|
/*supports_openai_form_elicitation*/ false,
|
|
)
|
|
.await
|
|
.expect("resume conversation");
|
|
|
|
// Assert: a Warning event is emitted describing the model mismatch.
|
|
let warning = wait_for_event(&conversation, |ev| {
|
|
matches!(
|
|
ev,
|
|
EventMsg::Warning(WarningEvent { message })
|
|
if message.contains("previous-model") && message.contains("current-model")
|
|
)
|
|
})
|
|
.await;
|
|
let EventMsg::Warning(WarningEvent { message }) = warning else {
|
|
panic!("expected warning event");
|
|
};
|
|
assert!(message.contains("previous-model"));
|
|
assert!(message.contains("current-model"));
|
|
|
|
// Drain the TurnComplete/Shutdown window to avoid leaking tasks between tests.
|
|
// The warning is emitted during initialization, so a short sleep is sufficient.
|
|
tokio::time::sleep(Duration::from_millis(50)).await;
|
|
}
|