mirror of
https://github.com/openai/codex.git
synced 2026-10-20 17:38:45 +00:00
tongzhou wang
1b6e0d8145
## Why Message roles and content classifications do not identify who supplied each context part. Source attribution needs to survive prompt assembly and history truncation, including messages that combine text from different producers. ## What changed - Add per-part `ContentItemMetadata` for user input, developer instructions, `AGENTS.md`, skills, harness context, command hooks, and tool context with optional callable namespaces. - Carry attribution alongside content and classifications through contextual fragments, extension prompts, hook output, and agent messages. Keep tool-provided context-window hints separate from harness text. - Preserve the source of retained text slices during compaction and attribute generated truncation notices to the harness. Default missing historical attribution to unknown. - Attribute host MCP calls using cached tool namespaces only when catalog revisions remain stable, without starting servers or refreshing catalogs for attribution. - Send the new metadata to Responses Lite and strip it from requests to endpoints that do not support it. ## Testing Extend compaction tests to cover mixed sources, missing or extra metadata entries, and discarded media. Update Responses Lite request assertions to verify endpoint-specific attribution handling. GitOrigin-RevId: b0ebc929b7c60f3ec49d1affbf897312d63edcc8
3517 lines
126 KiB
Rust
3517 lines
126 KiB
Rust
use super::*;
|
|
use crate::agent::types::SpawnAgentOptions;
|
|
use crate::config::RolloutBudgetConfig;
|
|
use crate::config::test_config;
|
|
use crate::init_state_db;
|
|
use crate::installation_id::INSTALLATION_ID_FILENAME;
|
|
use crate::mcp::McpEnvironmentScope;
|
|
use crate::mcp::McpThreadIdentity;
|
|
use crate::rollout::RolloutRecorder;
|
|
use crate::session::session::SessionSettingsUpdate;
|
|
use crate::session::step_context::StepContext;
|
|
use crate::session::tests::build_world_state_from_turn_context;
|
|
use crate::session::tests::make_session_and_context;
|
|
use crate::tasks::InterruptedTurnHistoryMarker;
|
|
use crate::tasks::interrupted_turn_history_marker;
|
|
use crate::windows_sandbox::WindowsSandboxLevelExt;
|
|
use codex_extension_api::SessionIsolation;
|
|
use codex_extension_api::empty_extension_registry;
|
|
use codex_history::InitialHistory;
|
|
use codex_history::ResumedHistory;
|
|
use codex_models_manager::manager::RefreshStrategy;
|
|
use codex_protocol::ResponseItemId;
|
|
use codex_protocol::capabilities::CapabilityRootLocation;
|
|
use codex_protocol::capabilities::SelectedCapabilityRoot;
|
|
use codex_protocol::config_types::WindowsSandboxLevel;
|
|
use codex_protocol::mcp::ClientMcpExtensions;
|
|
use codex_protocol::mcp::MCP_APP_UI_EXTENSION_ID;
|
|
use codex_protocol::mcp::OPENAI_FORM_EXTENSION_ID;
|
|
use codex_protocol::models::ContentItem;
|
|
use codex_protocol::models::ContentItemKind;
|
|
use codex_protocol::models::ContentItemMetadata;
|
|
use codex_protocol::models::InternalChatMessageMetadataPassthrough;
|
|
use codex_protocol::models::ReasoningItemReasoningSummary;
|
|
use codex_protocol::models::ResponseItem;
|
|
use codex_protocol::openai_models::ModelsResponse;
|
|
use codex_protocol::protocol::AgentMessageEvent;
|
|
use codex_protocol::protocol::EnvironmentConfigState;
|
|
use codex_protocol::protocol::InternalSessionSource;
|
|
use codex_protocol::protocol::SessionMeta;
|
|
use codex_protocol::protocol::SessionMetaLine;
|
|
use codex_protocol::protocol::SessionSource;
|
|
use codex_protocol::protocol::SubAgentSource;
|
|
use codex_protocol::protocol::ThreadSource;
|
|
use codex_protocol::protocol::TokenUsage;
|
|
use codex_protocol::protocol::TurnStartedEvent;
|
|
use codex_protocol::protocol::UserMessageEvent;
|
|
use codex_protocol::user_input::UserInput;
|
|
use codex_utils_path_uri::PathUri;
|
|
use core_test_support::PathBufExt;
|
|
use core_test_support::PathExt;
|
|
use core_test_support::responses::mount_models_once;
|
|
use core_test_support::responses::strip_response_item_ids_from_json;
|
|
use pretty_assertions::assert_eq;
|
|
use std::time::Duration;
|
|
use tempfile::tempdir;
|
|
use wiremock::MockServer;
|
|
|
|
const TEST_INSTALLATION_ID: &str = "11111111-1111-4111-8111-111111111111";
|
|
|
|
struct ParentInstructionsProvider(codex_extension_api::Instructions);
|
|
|
|
impl codex_extension_api::UserInstructionsProvider for ParentInstructionsProvider {
|
|
fn load_user_instructions(&self) -> codex_extension_api::LoadInstructionsFuture<'_> {
|
|
Box::pin(async move {
|
|
codex_extension_api::LoadedUserInstructions {
|
|
instructions: Some(self.0.clone()),
|
|
warnings: Vec::new(),
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
impl codex_extension_api::ThreadInstructionsProvider for ParentInstructionsProvider {
|
|
fn load_thread_instructions(&self) -> codex_extension_api::LoadInstructionsFuture<'_> {
|
|
codex_extension_api::UserInstructionsProvider::load_user_instructions(self)
|
|
}
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn live_fork_keeps_instructions_when_source_is_unloaded_during_setup() {
|
|
let temp_dir = tempdir().expect("tempdir");
|
|
let mut config = test_config().await;
|
|
config.codex_home = temp_dir.path().join("codex-home").abs();
|
|
config.cwd = config.codex_home.abs();
|
|
std::fs::create_dir_all(&config.codex_home).expect("create codex home");
|
|
let global = codex_extension_api::Instructions {
|
|
text: "global instructions".to_string(),
|
|
source: None,
|
|
};
|
|
let thread = codex_extension_api::Instructions {
|
|
text: "source thread instructions".to_string(),
|
|
source: None,
|
|
};
|
|
let mut manager = ThreadManager::with_models_provider_and_home_for_tests(
|
|
CodexAuth::from_api_key("dummy"),
|
|
config.model_provider.clone(),
|
|
config.codex_home.to_path_buf(),
|
|
Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()),
|
|
);
|
|
Arc::get_mut(&mut manager.state)
|
|
.expect("unshared manager")
|
|
.user_instructions_provider = Arc::new(ParentInstructionsProvider(global.clone()));
|
|
let source = manager
|
|
.start_thread(StartThreadOptions {
|
|
environments: Some(Vec::new()),
|
|
thread_instructions_provider: Some(Arc::new(ParentInstructionsProvider(
|
|
thread.clone(),
|
|
))),
|
|
..StartThreadOptions::new(config.clone())
|
|
})
|
|
.await
|
|
.expect("start source");
|
|
let history = InitialHistory::Resumed(ResumedHistory {
|
|
history_revision: None,
|
|
conversation_id: source.thread_id,
|
|
history: Arc::new(Vec::new()),
|
|
rollout_path: None,
|
|
});
|
|
|
|
// Queue removal between the first source lookup and subsequent startup
|
|
// lookups. Tokio's fair RwLock makes this ordering deterministic.
|
|
let fork = manager.fork_thread_from_history(
|
|
ForkSnapshot::Interrupted,
|
|
StartThreadOptions {
|
|
environments: Some(Vec::new()),
|
|
..StartThreadOptions::new(config)
|
|
},
|
|
history,
|
|
);
|
|
tokio::pin!(fork);
|
|
{
|
|
let _guard = manager.state.threads.write().await;
|
|
let mut context = std::task::Context::from_waker(std::task::Waker::noop());
|
|
assert!(std::future::Future::poll(fork.as_mut(), &mut context).is_pending());
|
|
}
|
|
let removal = manager.remove_thread_if_matches(&source.thread_id, &source.thread);
|
|
tokio::pin!(removal);
|
|
assert!(futures::poll!(&mut removal).is_pending());
|
|
assert!(futures::poll!(&mut fork).is_pending());
|
|
assert!(removal.await.is_some());
|
|
let fork = fork.await.expect("fork survives source removal");
|
|
let instructions = fork.thread.session.inherited_instructions().await;
|
|
assert_eq!(
|
|
(instructions.user, instructions.thread),
|
|
(Some(global), Some(thread))
|
|
);
|
|
source
|
|
.thread
|
|
.shutdown_and_wait()
|
|
.await
|
|
.expect("shutdown source");
|
|
fork.thread
|
|
.shutdown_and_wait()
|
|
.await
|
|
.expect("shutdown fork");
|
|
}
|
|
|
|
/// A thread opt-out wins over a shared client without disabling its siblings.
|
|
#[tokio::test]
|
|
async fn thread_analytics_opt_out_overrides_shared_client() {
|
|
let server = MockServer::start().await;
|
|
wiremock::Mock::given(wiremock::matchers::method("POST"))
|
|
.and(wiremock::matchers::path("/codex/analytics-events/events"))
|
|
.respond_with(wiremock::ResponseTemplate::new(200))
|
|
.mount(&server)
|
|
.await;
|
|
let temp_dir = tempdir().expect("tempdir");
|
|
let mut config = test_config().await;
|
|
config.chatgpt_base_url = server.uri();
|
|
config.model_provider.base_url = Some(server.uri());
|
|
let auth = CodexAuth::create_dummy_chatgpt_auth_for_testing();
|
|
let shared_client = AnalyticsEventsClient::new(
|
|
AuthManager::from_auth_for_testing(auth.clone()),
|
|
server.uri(),
|
|
/*analytics_enabled*/ Some(true),
|
|
);
|
|
let mut expected_thread_ids = Vec::new();
|
|
let mut opted_out_thread_ids = Vec::new();
|
|
|
|
for (name, client_override, expected_enabled) in [
|
|
(
|
|
"enabled_override",
|
|
Some(shared_client.clone()),
|
|
[false, true, true],
|
|
),
|
|
(
|
|
"disabled_override",
|
|
Some(AnalyticsEventsClient::disabled()),
|
|
[false, false, false],
|
|
),
|
|
("no_override", None, [false, true, true]),
|
|
] {
|
|
config.codex_home = temp_dir.path().join(name).abs();
|
|
config.cwd = config.codex_home.abs();
|
|
std::fs::create_dir_all(&config.codex_home).expect("create codex home");
|
|
let mut manager = ThreadManager::with_models_provider_and_home_for_tests(
|
|
auth.clone(),
|
|
config.model_provider.clone(),
|
|
config.codex_home.to_path_buf(),
|
|
Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()),
|
|
);
|
|
Arc::get_mut(&mut manager.state)
|
|
.expect("unshared thread manager state")
|
|
.analytics_events_client = client_override;
|
|
|
|
for (setting, enabled) in [Some(false), Some(true), None]
|
|
.into_iter()
|
|
.zip(expected_enabled)
|
|
{
|
|
config.analytics_enabled = setting;
|
|
let started = manager
|
|
.start_thread(StartThreadOptions::new(config.clone()))
|
|
.await
|
|
.expect("start analytics test thread");
|
|
let services = &started.thread.session.services;
|
|
assert_eq!(started.thread.analytics_enabled(), enabled);
|
|
assert_eq!(
|
|
services
|
|
.session_extension_data
|
|
.get::<AnalyticsEventsClient>()
|
|
.expect("analytics client in session store")
|
|
.is_enabled(),
|
|
enabled,
|
|
);
|
|
let thread_id = started.thread_id.to_string();
|
|
if enabled {
|
|
expected_thread_ids.push(thread_id.clone());
|
|
} else {
|
|
opted_out_thread_ids.push(thread_id.clone());
|
|
}
|
|
services.analytics_events_client.track_app_used(
|
|
codex_analytics::TrackEventsContext {
|
|
turn_metadata: None,
|
|
model_slug: "test-model".to_string(),
|
|
turn_id: format!("test-turn-{thread_id}"),
|
|
thread_id,
|
|
product_client_id: "codex_work_cca".to_string(),
|
|
},
|
|
codex_analytics::AppInvocation {
|
|
connector_id: Some("test-connector".to_string()),
|
|
app_name: None,
|
|
invocation_type: None,
|
|
},
|
|
/*elicitation_type*/ None,
|
|
);
|
|
services.analytics_events_client.flush().await;
|
|
}
|
|
let shutdown = manager
|
|
.shutdown_all_threads_bounded(Duration::from_secs(10))
|
|
.await;
|
|
assert_eq!(shutdown.completed.len(), 3);
|
|
}
|
|
|
|
let events: Vec<serde_json::Value> = server
|
|
.received_requests()
|
|
.await
|
|
.expect("analytics requests")
|
|
.into_iter()
|
|
.filter(|request| request.url.path() == "/codex/analytics-events/events")
|
|
.flat_map(|request| {
|
|
request.body_json::<serde_json::Value>().expect("JSON body")["events"]
|
|
.as_array()
|
|
.expect("events array")
|
|
.clone()
|
|
})
|
|
.collect();
|
|
assert!(events.iter().all(|event| {
|
|
!opted_out_thread_ids
|
|
.iter()
|
|
.any(|thread_id| event["event_params"]["thread_id"] == thread_id.as_str())
|
|
}));
|
|
let mut actual_thread_ids: Vec<String> = events
|
|
.iter()
|
|
.filter(|event| event["event_type"] == "codex_app_used")
|
|
.map(|event| {
|
|
event["event_params"]["thread_id"]
|
|
.as_str()
|
|
.expect("app usage thread ID")
|
|
.to_string()
|
|
})
|
|
.collect();
|
|
actual_thread_ids.sort();
|
|
expected_thread_ids.sort();
|
|
assert_eq!(actual_thread_ids, expected_thread_ids);
|
|
}
|
|
|
|
/// Controls without a custom allocation policy still produce distinct thread identifiers.
|
|
#[test]
|
|
fn thread_id_generator_defaults_to_standard_ids() {
|
|
let agent_control = LocalAgentControl::default();
|
|
|
|
assert_ne!(
|
|
agent_control.runtime.generate_thread_id(),
|
|
agent_control.runtime.generate_thread_id()
|
|
);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn reserved_thread_id_is_used_without_changing_normal_id_generation() {
|
|
let temp_dir = tempdir().expect("tempdir");
|
|
let mut config = test_config().await;
|
|
config.codex_home = temp_dir.path().join("codex-home").abs();
|
|
config.cwd = config.codex_home.abs();
|
|
std::fs::create_dir_all(&config.codex_home).expect("create codex home");
|
|
|
|
let generated_ids = [
|
|
ThreadId::from_u128(/*value*/ 0x018f_0000_0000_7000_8000_0000_0000_0001),
|
|
ThreadId::from_u128(/*value*/ 0x018f_0000_0000_7000_8000_0000_0000_0002),
|
|
ThreadId::from_u128(/*value*/ 0x018f_0000_0000_7000_8000_0000_0000_0003),
|
|
];
|
|
let next_id = std::sync::atomic::AtomicUsize::new(0);
|
|
let manager = ThreadManager::with_models_provider_and_home_for_tests(
|
|
CodexAuth::from_api_key("dummy"),
|
|
config.model_provider.clone(),
|
|
config.codex_home.to_path_buf(),
|
|
Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()),
|
|
)
|
|
.with_thread_id_generator(move || generated_ids[next_id.fetch_add(1, Ordering::Relaxed)]);
|
|
|
|
let reserved_id = manager.reserve_thread_id();
|
|
let mut reserved_options = StartThreadOptions::new(config.clone());
|
|
reserved_options.reserved_thread_id = Some(reserved_id);
|
|
let reserved = manager
|
|
.start_thread(reserved_options)
|
|
.await
|
|
.expect("start reserved thread");
|
|
let mut resumed_options = StartThreadOptions::new(config.clone());
|
|
resumed_options.initial_history = InitialHistory::Resumed(ResumedHistory {
|
|
history_revision: None,
|
|
conversation_id: reserved.thread_id,
|
|
history: Arc::new(Vec::new()),
|
|
rollout_path: None,
|
|
});
|
|
let resumed_id = manager.reserve_thread_id();
|
|
resumed_options.reserved_thread_id = Some(resumed_id);
|
|
let resume_error = manager
|
|
.start_thread(resumed_options)
|
|
.await
|
|
.err()
|
|
.expect("reject reserved ID for resume");
|
|
let generated = manager
|
|
.start_thread(StartThreadOptions::new(config.clone()))
|
|
.await
|
|
.expect("start generated thread");
|
|
|
|
assert_eq!(reserved.thread_id, generated_ids[0]);
|
|
assert!(matches!(
|
|
resume_error.details(),
|
|
codex_protocol::error::CodexErrorDetails::InvalidRequest(message)
|
|
if message == "reserved thread ID cannot be used when resuming a thread"
|
|
));
|
|
assert_eq!(generated.thread_id, generated_ids[2]);
|
|
}
|
|
|
|
/// One custom ID factory supplies identifiers for roots, actual child agents, and forks.
|
|
#[tokio::test]
|
|
async fn thread_id_generator_applies_to_roots_children_and_forks() {
|
|
let temp_dir = tempdir().expect("tempdir");
|
|
let mut config = test_config().await;
|
|
config.codex_home = temp_dir.path().join("codex-home").abs();
|
|
config.cwd = config.codex_home.abs();
|
|
std::fs::create_dir_all(&config.codex_home).expect("create codex home");
|
|
|
|
let generated_ids = [
|
|
ThreadId::from_u128(/*value*/ 0x018f_0000_0000_7000_8000_0000_0000_0001),
|
|
ThreadId::from_u128(/*value*/ 0x018f_0000_0000_7000_8000_0000_0000_0002),
|
|
ThreadId::from_u128(/*value*/ 0x018f_0000_0000_7000_8000_0000_0000_0003),
|
|
];
|
|
let next_id = std::sync::atomic::AtomicUsize::new(0);
|
|
let manager = ThreadManager::with_models_provider_and_home_for_tests(
|
|
CodexAuth::from_api_key("dummy"),
|
|
config.model_provider.clone(),
|
|
config.codex_home.to_path_buf(),
|
|
Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()),
|
|
)
|
|
.with_thread_id_generator(move || generated_ids[next_id.fetch_add(1, Ordering::Relaxed)]);
|
|
let root = manager
|
|
.start_thread(StartThreadOptions {
|
|
history_mode: Some(ThreadHistoryMode::Legacy),
|
|
..StartThreadOptions::new(config.clone())
|
|
})
|
|
.await
|
|
.expect("start root thread");
|
|
let child = root
|
|
.thread
|
|
.session
|
|
.services
|
|
.local_agent_runtime
|
|
.control(root.thread.session.session_id())
|
|
.spawn_agent_with_metadata(
|
|
config.clone(),
|
|
vec![UserInput::Text {
|
|
text: "child task".to_string(),
|
|
text_elements: Vec::new(),
|
|
}],
|
|
SessionSource::SubAgent(SubAgentSource::ThreadSpawn {
|
|
parent_thread_id: root.thread_id,
|
|
depth: 1,
|
|
agent_path: None,
|
|
agent_nickname: None,
|
|
agent_role: None,
|
|
}),
|
|
SpawnAgentOptions {
|
|
parent_thread_id: Some(root.thread_id),
|
|
..Default::default()
|
|
},
|
|
)
|
|
.await
|
|
.expect("spawn actual child agent");
|
|
let fork = manager
|
|
.spawn_legacy_subagent(root.thread_id, StartThreadOptions::new(config))
|
|
.await
|
|
.expect("fork root thread");
|
|
|
|
assert_eq!(
|
|
[root.thread_id, child.thread_id, fork.thread_id],
|
|
generated_ids
|
|
);
|
|
|
|
let report = manager
|
|
.shutdown_all_threads_bounded(Duration::from_secs(10))
|
|
.await;
|
|
assert_eq!(report.completed.len(), 3);
|
|
}
|
|
|
|
/// Resuming a thread preserves its stored ID instead of invoking the new manager's factory.
|
|
#[tokio::test]
|
|
async fn thread_id_generator_does_not_replace_resumed_thread_id() {
|
|
let temp_dir = tempdir().expect("tempdir");
|
|
let mut config = test_config().await;
|
|
config.codex_home = temp_dir.path().join("codex-home").abs();
|
|
config.cwd = config.codex_home.abs();
|
|
std::fs::create_dir_all(&config.codex_home).expect("create codex home");
|
|
|
|
let original_thread_id =
|
|
ThreadId::from_u128(/*value*/ 0x018f_0000_0000_7000_8000_0000_0000_0001);
|
|
let original_manager = ThreadManager::with_models_provider_and_home_for_tests(
|
|
CodexAuth::from_api_key("dummy"),
|
|
config.model_provider.clone(),
|
|
config.codex_home.to_path_buf(),
|
|
Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()),
|
|
)
|
|
.with_thread_id_generator(move || original_thread_id);
|
|
let original = original_manager
|
|
.start_thread(StartThreadOptions {
|
|
history_mode: Some(ThreadHistoryMode::Legacy),
|
|
..StartThreadOptions::new(config.clone())
|
|
})
|
|
.await
|
|
.expect("start source thread");
|
|
original.thread.ensure_rollout_materialized().await;
|
|
original
|
|
.thread
|
|
.flush_rollout()
|
|
.await
|
|
.expect("flush source rollout");
|
|
let rollout_path = original
|
|
.thread
|
|
.rollout_path()
|
|
.expect("source rollout path should exist");
|
|
assert_eq!(original.thread_id, original_thread_id);
|
|
original
|
|
.thread
|
|
.shutdown_and_wait()
|
|
.await
|
|
.expect("shut down source thread");
|
|
let _ = original_manager.remove_thread(&original_thread_id).await;
|
|
|
|
let resumed_manager = ThreadManager::with_models_provider_and_home_for_tests(
|
|
CodexAuth::from_api_key("dummy"),
|
|
config.model_provider.clone(),
|
|
config.codex_home.to_path_buf(),
|
|
Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()),
|
|
)
|
|
.with_thread_id_generator(|| panic!("resuming must not allocate a new thread ID"));
|
|
let resumed = resumed_manager
|
|
.resume_legacy_thread_from_rollout(
|
|
config,
|
|
rollout_path,
|
|
Arc::clone(&resumed_manager.state.auth_manager),
|
|
/*parent_trace*/ None,
|
|
ClientMcpExtensions::default(),
|
|
)
|
|
.await
|
|
.expect("resume existing source thread");
|
|
|
|
assert_eq!(resumed.thread_id, original_thread_id);
|
|
resumed
|
|
.thread
|
|
.shutdown_and_wait()
|
|
.await
|
|
.expect("shut down resumed thread");
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn child_session_inherits_client_mcp_extensions() {
|
|
let temp_dir = tempdir().expect("tempdir");
|
|
let mut config = test_config().await;
|
|
config.codex_home = temp_dir.path().join("codex-home").abs();
|
|
config.cwd = config.codex_home.abs();
|
|
std::fs::create_dir_all(&config.codex_home).expect("create codex home");
|
|
let manager = ThreadManager::with_models_provider_and_home_for_tests(
|
|
CodexAuth::from_api_key("dummy"),
|
|
config.model_provider.clone(),
|
|
config.codex_home.to_path_buf(),
|
|
Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()),
|
|
);
|
|
let parent = manager
|
|
.start_thread(StartThreadOptions {
|
|
client_mcp_extensions: ClientMcpExtensions::new(HashMap::from([
|
|
(OPENAI_FORM_EXTENSION_ID.to_string(), serde_json::json!({})),
|
|
(
|
|
MCP_APP_UI_EXTENSION_ID.to_string(),
|
|
serde_json::json!({
|
|
"mimeTypes": ["text/html;profile=mcp-app"],
|
|
}),
|
|
),
|
|
])),
|
|
..StartThreadOptions::new(config)
|
|
})
|
|
.await
|
|
.expect("start parent thread");
|
|
|
|
assert_eq!(
|
|
manager
|
|
.state
|
|
.client_mcp_extensions_for_child(Some(parent.thread_id))
|
|
.await,
|
|
ClientMcpExtensions::new(HashMap::from([
|
|
(OPENAI_FORM_EXTENSION_ID.to_string(), serde_json::json!({})),
|
|
(
|
|
MCP_APP_UI_EXTENSION_ID.to_string(),
|
|
serde_json::json!({
|
|
"mimeTypes": ["text/html;profile=mcp-app"],
|
|
}),
|
|
),
|
|
]))
|
|
);
|
|
}
|
|
|
|
struct FakeAgentGraphStore {
|
|
root_thread_id: ThreadId,
|
|
descendant_thread_ids: Vec<ThreadId>,
|
|
}
|
|
|
|
impl codex_agent_graph_store::AgentGraphStore for FakeAgentGraphStore {
|
|
fn upsert_thread_spawn_edge(
|
|
&self,
|
|
_parent_thread_id: ThreadId,
|
|
_child_thread_id: ThreadId,
|
|
_status: codex_agent_graph_store::ThreadSpawnEdgeStatus,
|
|
) -> codex_agent_graph_store::AgentGraphStoreFuture<'_, ()> {
|
|
Box::pin(async { panic!("unexpected graph upsert") })
|
|
}
|
|
|
|
fn set_thread_spawn_edge_status(
|
|
&self,
|
|
_child_thread_id: ThreadId,
|
|
_status: codex_agent_graph_store::ThreadSpawnEdgeStatus,
|
|
) -> codex_agent_graph_store::AgentGraphStoreFuture<'_, ()> {
|
|
Box::pin(async { panic!("unexpected graph status update") })
|
|
}
|
|
|
|
fn list_thread_spawn_children(
|
|
&self,
|
|
_parent_thread_id: ThreadId,
|
|
_status_filter: Option<codex_agent_graph_store::ThreadSpawnEdgeStatus>,
|
|
) -> codex_agent_graph_store::AgentGraphStoreFuture<'_, Vec<ThreadId>> {
|
|
Box::pin(async { panic!("unexpected direct-child listing") })
|
|
}
|
|
|
|
fn list_thread_spawn_descendants(
|
|
&self,
|
|
root_thread_id: ThreadId,
|
|
status_filter: Option<codex_agent_graph_store::ThreadSpawnEdgeStatus>,
|
|
) -> codex_agent_graph_store::AgentGraphStoreFuture<'_, Vec<ThreadId>> {
|
|
assert_eq!(root_thread_id, self.root_thread_id);
|
|
assert_eq!(status_filter, None);
|
|
let descendant_thread_ids = self.descendant_thread_ids.clone();
|
|
Box::pin(async move { Ok(descendant_thread_ids) })
|
|
}
|
|
}
|
|
|
|
fn user_msg(text: &str) -> ResponseItem {
|
|
ResponseItem::Message {
|
|
status: None,
|
|
encrypted_content: None,
|
|
id: None,
|
|
role: "user".to_string(),
|
|
content: vec![ContentItem::OutputText {
|
|
annotations: None,
|
|
logprobs: None,
|
|
text: text.to_string(),
|
|
}],
|
|
phase: None,
|
|
internal_chat_message_metadata_passthrough: None,
|
|
}
|
|
}
|
|
fn assistant_msg(text: &str) -> ResponseItem {
|
|
ResponseItem::Message {
|
|
status: None,
|
|
encrypted_content: None,
|
|
id: None,
|
|
role: "assistant".to_string(),
|
|
content: vec![ContentItem::OutputText {
|
|
annotations: None,
|
|
logprobs: None,
|
|
text: text.to_string(),
|
|
}],
|
|
phase: None,
|
|
internal_chat_message_metadata_passthrough: None,
|
|
}
|
|
}
|
|
|
|
fn contextual_user_interrupted_marker() -> ResponseItem {
|
|
interrupted_turn_history_marker(InterruptedTurnHistoryMarker::ContextualUser)
|
|
.expect("contextual-user interrupted marker should be enabled")
|
|
}
|
|
|
|
fn developer_interrupted_marker() -> ResponseItem {
|
|
interrupted_turn_history_marker(InterruptedTurnHistoryMarker::Developer)
|
|
.expect("developer interrupted marker should be enabled")
|
|
}
|
|
|
|
#[test]
|
|
fn effective_originator_prefers_thread_scoped_sources_before_env_originator() {
|
|
for (metrics_service_name, persisted_originator, inherited_originator, expected_originator) in [
|
|
(
|
|
Some("codex_work_desktop"),
|
|
Some("persisted_originator"),
|
|
Some("inherited_originator"),
|
|
"codex_work_desktop",
|
|
),
|
|
(
|
|
Some("codex_work_web"),
|
|
Some("persisted_originator"),
|
|
Some("inherited_originator"),
|
|
"codex_work_web",
|
|
),
|
|
(
|
|
Some("codex_work_mobile"),
|
|
Some("persisted_originator"),
|
|
Some("inherited_originator"),
|
|
"codex_work_mobile",
|
|
),
|
|
(
|
|
Some("codex_work_cca"),
|
|
Some("persisted_originator"),
|
|
Some("inherited_originator"),
|
|
"codex_work_cca",
|
|
),
|
|
(
|
|
Some("chatgpt_cca"),
|
|
Some("persisted_originator"),
|
|
Some("inherited_originator"),
|
|
"chatgpt_cca",
|
|
),
|
|
(
|
|
Some("chatgpt_cca_extra"),
|
|
Some("persisted_originator"),
|
|
Some("inherited_originator"),
|
|
"persisted_originator",
|
|
),
|
|
(
|
|
None,
|
|
Some("persisted_originator"),
|
|
Some("inherited_originator"),
|
|
"persisted_originator",
|
|
),
|
|
(
|
|
None,
|
|
None,
|
|
Some("inherited_originator"),
|
|
"inherited_originator",
|
|
),
|
|
] {
|
|
assert_eq!(
|
|
effective_originator_value(
|
|
metrics_service_name,
|
|
Some("Codex Desktop".to_string()),
|
|
persisted_originator.map(str::to_string),
|
|
inherited_originator.map(str::to_string),
|
|
"codex_cli_rs".to_string(),
|
|
),
|
|
expected_originator
|
|
);
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn truncates_before_requested_user_message() {
|
|
let items = [
|
|
user_msg("u1"),
|
|
assistant_msg("a1"),
|
|
assistant_msg("a2"),
|
|
user_msg("u2"),
|
|
assistant_msg("a3"),
|
|
ResponseItem::Reasoning {
|
|
id: Some(ResponseItemId::with_suffix("rs", "1")),
|
|
summary: vec![ReasoningItemReasoningSummary::SummaryText {
|
|
text: "s".to_string(),
|
|
}],
|
|
content: None,
|
|
encrypted_content: None,
|
|
internal_chat_message_metadata_passthrough: None,
|
|
},
|
|
ResponseItem::FunctionCall {
|
|
status: None,
|
|
encrypted_content: None,
|
|
id: None,
|
|
call_id: "c1".to_string(),
|
|
name: "tool".to_string(),
|
|
namespace: None,
|
|
arguments: "{}".to_string(),
|
|
encrypted_function_args: None,
|
|
internal_chat_message_metadata_passthrough: None,
|
|
},
|
|
assistant_msg("a4"),
|
|
];
|
|
|
|
let initial: Vec<RolloutItem> = items
|
|
.iter()
|
|
.cloned()
|
|
.map(|item| RolloutItem::ResponseItem(item.into()))
|
|
.collect();
|
|
let truncated = truncate_before_nth_user_message(
|
|
InitialHistory::Forked(initial),
|
|
/*n*/ 1,
|
|
&SnapshotTurnState {
|
|
ends_mid_turn: false,
|
|
active_turn_id: None,
|
|
active_turn_started_at: None,
|
|
active_turn_root_id: None,
|
|
active_turn_start_index: None,
|
|
},
|
|
);
|
|
let got_items = truncated.get_rollout_items();
|
|
let expected_items = vec![
|
|
RolloutItem::ResponseItem(items[0].clone().into()),
|
|
RolloutItem::ResponseItem(items[1].clone().into()),
|
|
RolloutItem::ResponseItem(items[2].clone().into()),
|
|
];
|
|
assert_eq!(
|
|
serde_json::to_value(got_items).unwrap(),
|
|
serde_json::to_value(&expected_items).unwrap()
|
|
);
|
|
|
|
let initial2: Vec<RolloutItem> = items
|
|
.iter()
|
|
.cloned()
|
|
.map(|item| RolloutItem::ResponseItem(item.into()))
|
|
.collect();
|
|
let truncated2 = truncate_before_nth_user_message(
|
|
InitialHistory::Forked(initial2.clone()),
|
|
/*n*/ 2,
|
|
&SnapshotTurnState {
|
|
ends_mid_turn: false,
|
|
active_turn_id: None,
|
|
active_turn_started_at: None,
|
|
active_turn_root_id: None,
|
|
active_turn_start_index: None,
|
|
},
|
|
);
|
|
assert_eq!(
|
|
serde_json::to_value(truncated2.get_rollout_items()).unwrap(),
|
|
serde_json::to_value(initial2).unwrap()
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn out_of_range_truncation_drops_only_unfinished_suffix_mid_turn() {
|
|
let items = vec![
|
|
RolloutItem::ResponseItem(user_msg("u1").into()),
|
|
RolloutItem::ResponseItem(assistant_msg("a1").into()),
|
|
RolloutItem::ResponseItem(user_msg("u2").into()),
|
|
RolloutItem::ResponseItem(assistant_msg("partial").into()),
|
|
];
|
|
|
|
let truncated = truncate_before_nth_user_message(
|
|
InitialHistory::Forked(items.clone()),
|
|
usize::MAX,
|
|
&SnapshotTurnState {
|
|
ends_mid_turn: true,
|
|
active_turn_id: None,
|
|
active_turn_started_at: None,
|
|
active_turn_root_id: None,
|
|
active_turn_start_index: None,
|
|
},
|
|
);
|
|
|
|
assert_eq!(
|
|
serde_json::to_value(truncated.get_rollout_items()).unwrap(),
|
|
serde_json::to_value(items[..2].to_vec()).unwrap()
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn fork_thread_accepts_legacy_usize_snapshot_argument() {
|
|
fn assert_legacy_snapshot_callsite(
|
|
manager: &ThreadManager,
|
|
config: Config,
|
|
path: std::path::PathBuf,
|
|
) {
|
|
let _future =
|
|
manager.fork_legacy_thread(usize::MAX, crate::StartThreadOptions::new(config), path);
|
|
}
|
|
|
|
let _: fn(&ThreadManager, Config, std::path::PathBuf) = assert_legacy_snapshot_callsite;
|
|
}
|
|
|
|
#[test]
|
|
fn out_of_range_truncation_drops_pre_user_active_turn_prefix() {
|
|
let items = vec![
|
|
RolloutItem::ResponseItem(user_msg("u1").into()),
|
|
RolloutItem::ResponseItem(assistant_msg("a1").into()),
|
|
RolloutItem::EventMsg(EventMsg::TurnStarted(TurnStartedEvent {
|
|
turn_attribution: None,
|
|
turn_id: "turn-2".to_string(),
|
|
root_turn_id: None,
|
|
trace_id: None,
|
|
started_at: None,
|
|
model_context_window: None,
|
|
collaboration_mode_kind: Default::default(),
|
|
})),
|
|
RolloutItem::ResponseItem(user_msg("u2").into()),
|
|
RolloutItem::ResponseItem(assistant_msg("partial").into()),
|
|
];
|
|
|
|
let snapshot_state = snapshot_turn_state(&InitialHistory::Forked(items.clone()));
|
|
assert_eq!(
|
|
snapshot_state,
|
|
SnapshotTurnState {
|
|
ends_mid_turn: true,
|
|
active_turn_id: Some("turn-2".to_string()),
|
|
active_turn_started_at: None,
|
|
active_turn_root_id: None,
|
|
active_turn_start_index: Some(2),
|
|
},
|
|
);
|
|
|
|
let truncated = truncate_before_nth_user_message(
|
|
InitialHistory::Forked(items.clone()),
|
|
usize::MAX,
|
|
&snapshot_state,
|
|
);
|
|
|
|
assert_eq!(
|
|
serde_json::to_value(truncated.get_rollout_items()).unwrap(),
|
|
serde_json::to_value(items[..2].to_vec()).unwrap()
|
|
);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn ignores_session_prefix_messages_when_truncating() {
|
|
let (session, turn_context) = make_session_and_context().await;
|
|
let turn_context = Arc::new(turn_context);
|
|
let world_state = build_world_state_from_turn_context(&session, &turn_context).await;
|
|
let step_context = StepContext::for_test(turn_context);
|
|
let updates = session
|
|
.build_initial_context_with_world_state(&step_context, &world_state)
|
|
.await
|
|
.0;
|
|
let mut items = crate::context_manager::updates::merge_world_state_updates(updates);
|
|
items.push(user_msg("feature request"));
|
|
items.push(assistant_msg("ack"));
|
|
items.push(user_msg("second question"));
|
|
items.push(assistant_msg("answer"));
|
|
|
|
let rollout_items: Vec<RolloutItem> = items
|
|
.iter()
|
|
.cloned()
|
|
.map(|item| RolloutItem::ResponseItem(item.into()))
|
|
.collect();
|
|
|
|
let truncated = truncate_before_nth_user_message(
|
|
InitialHistory::Forked(rollout_items),
|
|
/*n*/ 1,
|
|
&SnapshotTurnState {
|
|
ends_mid_turn: false,
|
|
active_turn_id: None,
|
|
active_turn_started_at: None,
|
|
active_turn_root_id: None,
|
|
active_turn_start_index: None,
|
|
},
|
|
);
|
|
let got_items = truncated.get_rollout_items();
|
|
|
|
let expected: Vec<RolloutItem> = vec![
|
|
RolloutItem::ResponseItem(items[0].clone().into()),
|
|
RolloutItem::ResponseItem(items[1].clone().into()),
|
|
RolloutItem::ResponseItem(items[2].clone().into()),
|
|
RolloutItem::ResponseItem(items[3].clone().into()),
|
|
];
|
|
|
|
assert_eq!(
|
|
serde_json::to_value(got_items).unwrap(),
|
|
serde_json::to_value(&expected).unwrap()
|
|
);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn shutdown_all_threads_bounded_submits_shutdown_to_every_thread() {
|
|
let temp_dir = tempdir().expect("tempdir");
|
|
let mut config = test_config().await;
|
|
config.codex_home = temp_dir.path().join("codex-home").abs();
|
|
config.cwd = config.codex_home.abs();
|
|
std::fs::create_dir_all(&config.codex_home).expect("create codex home");
|
|
|
|
let manager = ThreadManager::with_models_provider_and_home_for_tests(
|
|
CodexAuth::from_api_key("dummy"),
|
|
config.model_provider.clone(),
|
|
config.codex_home.to_path_buf(),
|
|
Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()),
|
|
);
|
|
let thread_1 = manager
|
|
.start_thread(StartThreadOptions::new(config.clone()))
|
|
.await
|
|
.expect("start first thread")
|
|
.thread_id;
|
|
let thread_2 = manager
|
|
.start_thread(StartThreadOptions::new(config.clone()))
|
|
.await
|
|
.expect("start second thread")
|
|
.thread_id;
|
|
|
|
let report = manager
|
|
.shutdown_all_threads_bounded(Duration::from_secs(10))
|
|
.await;
|
|
|
|
let mut expected_completed = vec![thread_1, thread_2];
|
|
expected_completed.sort_by_key(std::string::ToString::to_string);
|
|
assert_eq!(report.completed, expected_completed);
|
|
assert!(report.submit_failed.is_empty());
|
|
assert!(report.timed_out.is_empty());
|
|
assert!(manager.list_thread_ids().await.is_empty());
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn code_mode_session_provider_is_shared_across_threads() {
|
|
let temp_dir = tempdir().expect("tempdir");
|
|
let mut config = test_config().await;
|
|
config.codex_home = temp_dir.path().join("codex-home").abs();
|
|
config.cwd = config.codex_home.abs();
|
|
std::fs::create_dir_all(&config.codex_home).expect("create codex home");
|
|
|
|
let provider: Arc<dyn CodeModeSessionProvider> = Arc::new(DisabledCodeModeSessionProvider);
|
|
let manager = ThreadManager::with_models_provider_and_home_for_tests(
|
|
CodexAuth::from_api_key("dummy"),
|
|
config.model_provider.clone(),
|
|
config.codex_home.to_path_buf(),
|
|
Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()),
|
|
)
|
|
.with_code_mode_session_provider(Arc::clone(&provider));
|
|
let first = manager
|
|
.start_thread(StartThreadOptions::new(config.clone()))
|
|
.await
|
|
.expect("start first thread");
|
|
let second = manager
|
|
.start_thread(StartThreadOptions::new(config))
|
|
.await
|
|
.expect("start second thread");
|
|
|
|
let first_provider = first
|
|
.thread
|
|
.session
|
|
.services
|
|
.code_mode_service
|
|
.session_provider();
|
|
let second_provider = second
|
|
.thread
|
|
.session
|
|
.services
|
|
.code_mode_service
|
|
.session_provider();
|
|
assert!(Arc::ptr_eq(&first_provider, &second_provider));
|
|
assert!(Arc::ptr_eq(&first_provider, &provider));
|
|
assert!(Arc::ptr_eq(
|
|
&first_provider,
|
|
&manager.state.code_mode_session_provider
|
|
));
|
|
|
|
let mut completed = vec![first.thread_id, second.thread_id];
|
|
completed.sort_by_key(std::string::ToString::to_string);
|
|
let report = manager
|
|
.shutdown_all_threads_bounded(Duration::from_secs(10))
|
|
.await;
|
|
assert_eq!(
|
|
report,
|
|
ThreadShutdownReport {
|
|
completed,
|
|
submit_failed: Vec::new(),
|
|
timed_out: Vec::new(),
|
|
}
|
|
);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn mcp_invalidation_refreshes_threads_that_are_still_starting() {
|
|
struct BlockingThreadStartup {
|
|
entered: tokio::sync::Notify,
|
|
release: tokio::sync::Notify,
|
|
refreshed: tokio::sync::Notify,
|
|
projections: std::sync::atomic::AtomicUsize,
|
|
}
|
|
|
|
impl codex_extension_api::ThreadLifecycleContributor<Config> for BlockingThreadStartup {
|
|
fn on_thread_start<'a>(
|
|
&'a self,
|
|
_input: codex_extension_api::ThreadStartInput<'a, Config>,
|
|
) -> codex_extension_api::ExtensionFuture<'a, ()> {
|
|
Box::pin(async move {
|
|
self.entered.notify_one();
|
|
self.release.notified().await;
|
|
})
|
|
}
|
|
}
|
|
|
|
impl codex_extension_api::McpServerContributor<Config> for BlockingThreadStartup {
|
|
fn id(&self) -> &'static str {
|
|
"starting_mcp_runtime_refresh_test"
|
|
}
|
|
|
|
fn contribute<'a>(
|
|
&'a self,
|
|
_context: codex_extension_api::McpServerContributionContext<'a, Config>,
|
|
) -> codex_extension_api::ExtensionFuture<'a, Vec<codex_extension_api::McpServerContribution>>
|
|
{
|
|
Box::pin(async move {
|
|
if self.projections.fetch_add(1, Ordering::AcqRel) != 0 {
|
|
self.refreshed.notify_one();
|
|
}
|
|
Vec::new()
|
|
})
|
|
}
|
|
}
|
|
|
|
let temp_dir = tempdir().expect("tempdir");
|
|
let mut config = test_config().await;
|
|
// Resolve the model locally before waiting for the MCP lifecycle hook.
|
|
config.model_catalog = Some(
|
|
codex_models_manager::bundled_models_response().expect("bundled model catalog should load"),
|
|
);
|
|
config.codex_home = temp_dir.path().join("codex-home").abs();
|
|
config.cwd = config.codex_home.abs();
|
|
std::fs::create_dir_all(&config.codex_home).expect("create codex home");
|
|
|
|
let observer = Arc::new(BlockingThreadStartup {
|
|
entered: tokio::sync::Notify::new(),
|
|
release: tokio::sync::Notify::new(),
|
|
refreshed: tokio::sync::Notify::new(),
|
|
projections: std::sync::atomic::AtomicUsize::new(0),
|
|
});
|
|
let mut extensions = codex_extension_api::ExtensionRegistryBuilder::new();
|
|
extensions.thread_lifecycle_contributor(observer.clone());
|
|
extensions.mcp_server_contributor(observer.clone());
|
|
let auth_manager = AuthManager::from_auth_for_testing(CodexAuth::from_api_key("dummy"));
|
|
let manager = Arc::new(ThreadManager::new(
|
|
&config,
|
|
Arc::clone(&auth_manager),
|
|
build_models_manager(&config, auth_manager),
|
|
crate::CodexAppsToolsCache::default(),
|
|
SessionSource::Exec,
|
|
Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()),
|
|
Arc::new(extensions.build()),
|
|
Arc::new(crate::test_support::EmptyUserInstructionsProvider),
|
|
/*analytics_events_client*/ None,
|
|
passthrough_image_store(),
|
|
thread_store_from_config(&config, /*state_db*/ None),
|
|
/*agent_graph_store*/ None,
|
|
TEST_INSTALLATION_ID.to_string(),
|
|
/*attestation_provider*/ None,
|
|
/*external_time_provider*/ None,
|
|
));
|
|
let starting = tokio::spawn({
|
|
let manager = Arc::clone(&manager);
|
|
async move { manager.start_thread(StartThreadOptions::new(config)).await }
|
|
});
|
|
|
|
tokio::time::timeout(Duration::from_secs(5), observer.entered.notified())
|
|
.await
|
|
.expect("thread should enter its startup lifecycle");
|
|
assert!(manager.list_thread_ids().await.is_empty());
|
|
manager.invalidate_mcp_runtimes().await;
|
|
observer.release.notify_one();
|
|
starting
|
|
.await
|
|
.expect("thread startup task should finish")
|
|
.expect("thread should start");
|
|
tokio::time::timeout(Duration::from_secs(5), observer.refreshed.notified())
|
|
.await
|
|
.expect("invalidation during startup should refresh the newly published thread");
|
|
let shutdown = manager
|
|
.shutdown_all_threads_bounded(Duration::from_secs(5))
|
|
.await;
|
|
assert!(shutdown.timed_out.is_empty());
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn start_thread_keeps_internal_threads_hidden_from_normal_lookups() {
|
|
let temp_dir = tempdir().expect("tempdir");
|
|
let mut config = test_config().await;
|
|
config.codex_home = temp_dir.path().join("codex-home").abs();
|
|
config.cwd = config.codex_home.abs();
|
|
std::fs::create_dir_all(&config.codex_home).expect("create codex home");
|
|
|
|
let manager = ThreadManager::with_models_provider_and_home_for_tests(
|
|
CodexAuth::from_api_key("dummy"),
|
|
config.model_provider.clone(),
|
|
config.codex_home.to_path_buf(),
|
|
Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()),
|
|
);
|
|
let thread = manager
|
|
.start_thread(StartThreadOptions {
|
|
session_source: Some(SessionSource::Internal(
|
|
InternalSessionSource::MemoryConsolidation,
|
|
)),
|
|
environments: Some(Vec::new()),
|
|
..StartThreadOptions::new(config)
|
|
})
|
|
.await
|
|
.expect("internal thread should start");
|
|
|
|
assert_eq!(manager.list_thread_ids().await, Vec::new());
|
|
assert!(manager.get_thread(thread.thread_id).await.is_err());
|
|
assert!(
|
|
codex_diagnostics::snapshot()
|
|
.gauges
|
|
.iter()
|
|
.any(|gauge| gauge.name == "core.threads.live" && gauge.value > 0)
|
|
);
|
|
|
|
let report = manager
|
|
.shutdown_all_threads_bounded(Duration::from_secs(10))
|
|
.await;
|
|
assert_eq!(report.completed, vec![thread.thread_id]);
|
|
assert!(report.submit_failed.is_empty());
|
|
assert!(report.timed_out.is_empty());
|
|
assert!(manager.list_thread_ids().await.is_empty());
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn spawn_internal_guardian_session_preserves_windows_sandbox_proxy_settings() {
|
|
let temp_dir = tempdir().expect("tempdir");
|
|
let mut config = test_config().await;
|
|
config.codex_home = temp_dir.path().join("codex-home").abs();
|
|
config.cwd = config.codex_home.abs();
|
|
std::fs::create_dir_all(&config.codex_home).expect("create codex home");
|
|
|
|
let manager = ThreadManager::with_models_provider_and_home_for_tests(
|
|
CodexAuth::from_api_key("dummy"),
|
|
config.model_provider.clone(),
|
|
config.codex_home.to_path_buf(),
|
|
Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()),
|
|
);
|
|
let parent = manager
|
|
.start_thread(StartThreadOptions::new(config.clone()))
|
|
.await
|
|
.expect("start parent thread");
|
|
let reviewer = manager
|
|
.spawn_internal_session(
|
|
parent.thread_id,
|
|
StartThreadOptions {
|
|
session_source: Some(SessionSource::Internal(InternalSessionSource::Guardian)),
|
|
..StartThreadOptions::new(config)
|
|
},
|
|
)
|
|
.await
|
|
.expect("start internal reviewer");
|
|
|
|
assert_eq!(
|
|
(
|
|
parent.thread.session.windows_sandbox_proxy_settings_mode,
|
|
reviewer.thread.session.windows_sandbox_proxy_settings_mode,
|
|
),
|
|
(
|
|
codex_sandboxing::WindowsSandboxProxySettingsMode::Reconcile,
|
|
codex_sandboxing::WindowsSandboxProxySettingsMode::Preserve,
|
|
)
|
|
);
|
|
|
|
manager
|
|
.shutdown_all_threads_bounded(Duration::from_secs(10))
|
|
.await;
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn fork_internal_session_uses_only_the_selected_history() {
|
|
let temp_dir = tempdir().expect("tempdir");
|
|
let mut config = test_config().await;
|
|
config.codex_home = temp_dir.path().join("codex-home").abs();
|
|
config.cwd = config.codex_home.abs();
|
|
std::fs::create_dir_all(&config.codex_home).expect("create codex home");
|
|
let manager = ThreadManager::with_models_provider_and_home_for_tests(
|
|
CodexAuth::from_api_key("dummy"),
|
|
config.model_provider.clone(),
|
|
config.codex_home.to_path_buf(),
|
|
Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()),
|
|
);
|
|
let parent = manager
|
|
.start_thread(StartThreadOptions::new(config.clone()))
|
|
.await
|
|
.expect("start parent");
|
|
let selected = vec![
|
|
user_msg("committed review"),
|
|
assistant_msg("completed assessment"),
|
|
];
|
|
parent
|
|
.thread
|
|
.inject_response_items(vec![user_msg("unrelated parent work")])
|
|
.await
|
|
.expect("inject parent history");
|
|
let reviewer = manager
|
|
.fork_internal_session(
|
|
parent.thread_id,
|
|
StartThreadOptions {
|
|
session_source: Some(SessionSource::Internal(InternalSessionSource::Guardian)),
|
|
..StartThreadOptions::new(config)
|
|
},
|
|
selected
|
|
.iter()
|
|
.cloned()
|
|
.map(|item| RolloutItem::ResponseItem(item.into()))
|
|
.collect(),
|
|
)
|
|
.await
|
|
.expect("fork internal reviewer");
|
|
let history = reviewer.thread.conversation_history_snapshot().await;
|
|
let mut actual = history.items().cloned().collect::<Vec<_>>();
|
|
// Recording history assigns message IDs and provenance; compare the selected content.
|
|
for item in &mut actual {
|
|
if let ResponseItem::Message {
|
|
id,
|
|
internal_chat_message_metadata_passthrough,
|
|
..
|
|
} = item
|
|
{
|
|
*id = None;
|
|
*internal_chat_message_metadata_passthrough = None;
|
|
}
|
|
}
|
|
assert_eq!(actual, selected);
|
|
assert_eq!(
|
|
reviewer.thread.config_snapshot().await.parent_thread_id,
|
|
Some(parent.thread_id)
|
|
);
|
|
assert!(Arc::ptr_eq(
|
|
&reviewer.thread.session.services.auth_manager,
|
|
&parent.thread.session.services.auth_manager,
|
|
));
|
|
assert_eq!(manager.list_thread_ids().await, vec![parent.thread_id]);
|
|
assert!(manager.get_thread(reviewer.thread_id).await.is_err());
|
|
manager
|
|
.shutdown_all_threads_bounded(Duration::from_secs(10))
|
|
.await;
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn spawn_internal_session_preserves_parent_lineage_without_forking_history() {
|
|
struct ParentLifecycleContributor {
|
|
observed_mcp_sources: Arc<std::sync::Mutex<Vec<SessionSource>>>,
|
|
}
|
|
|
|
impl codex_extension_api::ThreadLifecycleContributor<Config> for ParentLifecycleContributor {}
|
|
|
|
impl codex_extension_api::McpServerContributor<Config> for ParentLifecycleContributor {
|
|
fn id(&self) -> &'static str {
|
|
"parent_mcp_contributor"
|
|
}
|
|
|
|
fn contribute<'a>(
|
|
&'a self,
|
|
context: codex_extension_api::McpServerContributionContext<'a, Config>,
|
|
) -> codex_extension_api::ExtensionFuture<'a, Vec<codex_extension_api::McpServerContribution>>
|
|
{
|
|
Box::pin(async move {
|
|
if let Some(session_source) = context.session_source() {
|
|
self.observed_mcp_sources
|
|
.lock()
|
|
.unwrap_or_else(std::sync::PoisonError::into_inner)
|
|
.push(session_source.clone());
|
|
}
|
|
Vec::new()
|
|
})
|
|
}
|
|
}
|
|
|
|
let temp_dir = tempdir().expect("tempdir");
|
|
let mut config = test_config().await;
|
|
config.rollout_budget = Some(RolloutBudgetConfig {
|
|
limit_tokens: 100,
|
|
reminder_at_remaining_tokens: vec![75, 50, 25],
|
|
sampling_token_weight: 1.0,
|
|
prefill_token_weight: 1.0,
|
|
});
|
|
config.codex_home = temp_dir.path().join("codex-home").abs();
|
|
config.cwd = config.codex_home.abs();
|
|
std::fs::create_dir_all(&config.codex_home).expect("create codex home");
|
|
|
|
let mut managed_exec_policy = codex_execpolicy::Policy::empty();
|
|
managed_exec_policy
|
|
.add_prefix_rule(&["rm".to_string()], codex_execpolicy::Decision::Forbidden)
|
|
.expect("add managed execution restriction");
|
|
let mut requirements = config.config_layer_stack.requirements().clone();
|
|
requirements.exec_policy = Some(codex_config::Sourced::new(
|
|
codex_execpolicy::RequirementsExecPolicy::new(managed_exec_policy),
|
|
codex_config::RequirementSource::Unknown,
|
|
));
|
|
requirements.additional_developer_instructions = Some(codex_config::Sourced::new(
|
|
"managed instructions must not shape the reviewer".to_string(),
|
|
codex_config::RequirementSource::Unknown,
|
|
));
|
|
let mut requirements_toml = config.config_layer_stack.requirements_toml().clone();
|
|
requirements_toml.additional_developer_instructions =
|
|
Some("managed instructions must not shape the reviewer".to_string());
|
|
config.config_layer_stack = codex_config::ConfigLayerStack::new(
|
|
config
|
|
.config_layer_stack
|
|
.all_layers_low_to_high()
|
|
.cloned()
|
|
.collect(),
|
|
requirements,
|
|
requirements_toml,
|
|
)
|
|
.expect("managed requirements stack");
|
|
|
|
let parent_instructions = codex_extension_api::Instructions {
|
|
text: "parent user instructions must not be inherited".to_string(),
|
|
source: Some(config.codex_home.join("AGENTS.md")),
|
|
};
|
|
let mut manager = ThreadManager::with_models_provider_and_home_for_tests(
|
|
CodexAuth::from_api_key("dummy"),
|
|
config.model_provider.clone(),
|
|
config.codex_home.to_path_buf(),
|
|
Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()),
|
|
);
|
|
let observed_mcp_sources = Arc::new(std::sync::Mutex::new(Vec::new()));
|
|
let parent_contributor = Arc::new(ParentLifecycleContributor {
|
|
observed_mcp_sources: Arc::clone(&observed_mcp_sources),
|
|
});
|
|
let mut extensions = codex_extension_api::ExtensionRegistryBuilder::new();
|
|
extensions.thread_lifecycle_contributor(parent_contributor.clone());
|
|
extensions.mcp_server_contributor(parent_contributor);
|
|
let manager_state = Arc::get_mut(&mut manager.state).expect("unshared thread manager state");
|
|
manager_state.extensions = Arc::new(extensions.build());
|
|
manager_state.mcp_manager = Arc::new(McpManager::new_with_extensions(
|
|
Arc::clone(&manager_state.plugins_manager),
|
|
Arc::clone(&manager_state.extensions),
|
|
manager_state.mcp_manager.codex_apps_tools_cache(),
|
|
));
|
|
let parent_provider = Arc::new(ParentInstructionsProvider(parent_instructions.clone()));
|
|
manager_state.user_instructions_provider = parent_provider.clone();
|
|
let parent = manager
|
|
.start_thread(StartThreadOptions {
|
|
metrics_service_name: Some("codex_work_desktop".to_string()),
|
|
thread_instructions_provider: Some(parent_provider),
|
|
..StartThreadOptions::new(config.clone())
|
|
})
|
|
.await
|
|
.expect("start parent thread");
|
|
parent
|
|
.thread
|
|
.session
|
|
.set_multi_agent_version_if_unset(MultiAgentVersion::V2);
|
|
let parent_snapshot = parent.thread.session.inherited_instructions().await;
|
|
assert_eq!(
|
|
(parent_snapshot.user, parent_snapshot.thread),
|
|
(Some(parent_instructions.clone()), Some(parent_instructions))
|
|
);
|
|
assert_eq!(
|
|
parent
|
|
.thread
|
|
.session
|
|
.services
|
|
.extensions
|
|
.thread_lifecycle_contributors()
|
|
.len(),
|
|
1
|
|
);
|
|
let mut reviewer_environments = parent
|
|
.thread
|
|
.session
|
|
.services
|
|
.turn_environments
|
|
.selections();
|
|
let reviewer_environment = reviewer_environments
|
|
.first_mut()
|
|
.expect("parent should have an environment");
|
|
reviewer_environment.config =
|
|
EnvironmentConfigState::Ready(codex_protocol::protocol::EnvironmentConfig {
|
|
allow_login_shell: true,
|
|
workspace_roots: reviewer_environment.workspace_roots.clone(),
|
|
permission_profile: config.permissions.permission_profile_state().snapshot(),
|
|
shell_environment_policy: Default::default(),
|
|
windows_sandbox_level: WindowsSandboxLevel::from_config(&config),
|
|
windows_sandbox_type: config.permissions.windows_sandbox_type,
|
|
use_legacy_landlock: config.features.use_legacy_landlock(),
|
|
exec_policy: Some(codex_execpolicy::RequirementsExecPolicy::new(
|
|
codex_execpolicy::Policy::empty(),
|
|
)),
|
|
mcp_policy: None,
|
|
network_policy: None,
|
|
selected_capability_roots: Vec::new(),
|
|
});
|
|
let reviewer = manager
|
|
.spawn_internal_session(
|
|
parent.thread_id,
|
|
StartThreadOptions {
|
|
session_source: Some(SessionSource::Internal(InternalSessionSource::Guardian)),
|
|
initial_history: InitialHistory::Forked(vec![RolloutItem::ResponseItem(
|
|
user_msg("parent history must not be inherited").into(),
|
|
)]),
|
|
environments: Some(
|
|
reviewer_environments
|
|
.into_iter()
|
|
.map(TurnEnvironmentSelection::into_request)
|
|
.collect(),
|
|
),
|
|
..StartThreadOptions::new(config)
|
|
},
|
|
)
|
|
.await
|
|
.expect("start internal reviewer");
|
|
let reviewer_config = reviewer.thread.config_snapshot().await;
|
|
assert!(Arc::ptr_eq(
|
|
&reviewer.thread.session.services.agent_control,
|
|
&parent.thread.session.services.agent_control,
|
|
));
|
|
|
|
assert_eq!(
|
|
reviewer.session_configured.session_id,
|
|
parent.session_configured.session_id
|
|
);
|
|
reviewer
|
|
.thread
|
|
.session
|
|
.services
|
|
.agent_control
|
|
.record_usage(TokenUsage {
|
|
output_tokens: 25,
|
|
..Default::default()
|
|
})
|
|
.await
|
|
.expect("record reviewer usage");
|
|
let reminder = parent
|
|
.thread
|
|
.session
|
|
.services
|
|
.agent_control
|
|
.pending_budget_reminder(parent.thread_id, "window")
|
|
.await
|
|
.expect("parent budget reminder");
|
|
assert_eq!(reminder.remaining_tokens, 75);
|
|
assert_eq!(reviewer_config.parent_thread_id, Some(parent.thread_id));
|
|
assert_eq!(reviewer_config.forked_from_thread_id, None);
|
|
assert_eq!(reviewer_config.originator, "codex_work_desktop");
|
|
assert_eq!(
|
|
reviewer.thread.multi_agent_version(),
|
|
Some(MultiAgentVersion::Disabled)
|
|
);
|
|
assert_eq!(
|
|
reviewer.session_configured.parent_thread_id,
|
|
Some(parent.thread_id)
|
|
);
|
|
assert_eq!(reviewer.session_configured.forked_from_id, None);
|
|
let reviewer_snapshot = reviewer.thread.session.inherited_instructions().await;
|
|
assert_eq!(
|
|
(reviewer_snapshot.user, reviewer_snapshot.thread),
|
|
(None, None)
|
|
);
|
|
assert!(
|
|
reviewer
|
|
.thread
|
|
.session
|
|
.services
|
|
.exec_policy
|
|
.current()
|
|
.rules()
|
|
.contains_key("rm")
|
|
);
|
|
let reviewer_turn = reviewer.thread.session.new_default_turn().await;
|
|
let reviewer_world_state =
|
|
build_world_state_from_turn_context(&reviewer.thread.session, &reviewer_turn).await;
|
|
let reviewer_step = StepContext::for_test(Arc::clone(&reviewer_turn));
|
|
let reviewer_context = reviewer
|
|
.thread
|
|
.session
|
|
.build_initial_context_with_world_state(&reviewer_step, &reviewer_world_state)
|
|
.await
|
|
.0;
|
|
let reviewer_context =
|
|
crate::context_manager::updates::merge_world_state_updates(reviewer_context);
|
|
assert!(
|
|
!serde_json::to_string(&reviewer_context)
|
|
.expect("reviewer context should serialize")
|
|
.contains("managed instructions must not shape the reviewer")
|
|
);
|
|
let reviewer_environment = reviewer
|
|
.thread
|
|
.environment_selections()
|
|
.await
|
|
.into_iter()
|
|
.next()
|
|
.expect("reviewer should retain its selected environment");
|
|
assert!(matches!(
|
|
reviewer_environment.config,
|
|
EnvironmentConfigState::Ready(config) if config.exec_policy.is_some()
|
|
));
|
|
assert!(
|
|
reviewer
|
|
.thread
|
|
.session
|
|
.services
|
|
.extensions
|
|
.thread_lifecycle_contributors()
|
|
.is_empty()
|
|
);
|
|
{
|
|
let observed_mcp_sources = observed_mcp_sources
|
|
.lock()
|
|
.unwrap_or_else(std::sync::PoisonError::into_inner);
|
|
assert!(!observed_mcp_sources.is_empty());
|
|
assert!(
|
|
observed_mcp_sources
|
|
.iter()
|
|
.all(|source| !source.is_internal())
|
|
);
|
|
}
|
|
assert_eq!(manager.list_thread_ids().await, vec![parent.thread_id]);
|
|
assert!(manager.get_thread(reviewer.thread_id).await.is_err());
|
|
assert!(
|
|
reviewer
|
|
.thread
|
|
.session
|
|
.clone_history()
|
|
.await
|
|
.raw_items()
|
|
.next()
|
|
.is_none()
|
|
);
|
|
|
|
// Captured lineage also works for an inline parent with no registry entry.
|
|
manager.remove_thread(&parent.thread_id).await;
|
|
let child = manager
|
|
.start_thread(StartThreadOptions {
|
|
internal_parent: Some(InternalSessionParent {
|
|
thread_id: parent.thread_id,
|
|
auth_manager: Arc::clone(&parent.thread.session.services.auth_manager),
|
|
agent_control: AgentControlInit::Provided {
|
|
control: Arc::clone(&parent.thread.session.services.agent_control),
|
|
runtime: parent.thread.session.services.local_agent_runtime.clone(),
|
|
},
|
|
originator: reviewer_config.originator.clone(),
|
|
inherited_instructions: None,
|
|
}),
|
|
session_source: Some(SessionSource::Internal(
|
|
InternalSessionSource::MemoryConsolidation,
|
|
)),
|
|
..StartThreadOptions::new(parent.thread.session.get_config().await.as_ref().clone())
|
|
})
|
|
.await
|
|
.expect("start child without a parent registry entry");
|
|
let child_config = child.thread.config_snapshot().await;
|
|
assert_eq!(
|
|
(
|
|
child_config.parent_thread_id,
|
|
child_config.originator,
|
|
child.thread.session.session_id(),
|
|
),
|
|
(
|
|
Some(parent.thread_id),
|
|
reviewer_config.originator,
|
|
parent.thread.session.session_id(),
|
|
)
|
|
);
|
|
assert!(Arc::ptr_eq(
|
|
&child.thread.session.services.auth_manager,
|
|
&parent.thread.session.services.auth_manager
|
|
));
|
|
assert!(manager.list_thread_ids().await.is_empty());
|
|
parent
|
|
.thread
|
|
.shutdown_and_wait()
|
|
.await
|
|
.expect("stop parent");
|
|
manager
|
|
.shutdown_all_threads_bounded(Duration::from_secs(10))
|
|
.await;
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn start_thread_seeds_extension_data_for_mcp_and_lifecycle_contributors() {
|
|
struct InitialDataRecorder {
|
|
lifecycle_observed: Arc<std::sync::Mutex<Vec<(String, String)>>>,
|
|
mcp_observed: Arc<std::sync::Mutex<Vec<(String, SessionSource)>>>,
|
|
mcp_loaded: Arc<std::sync::atomic::AtomicUsize>,
|
|
}
|
|
|
|
impl codex_extension_api::ThreadLifecycleContributor<Config> for InitialDataRecorder {
|
|
fn on_thread_start<'a>(
|
|
&'a self,
|
|
input: codex_extension_api::ThreadStartInput<'a, Config>,
|
|
) -> codex_extension_api::ExtensionFuture<'a, ()> {
|
|
Box::pin(async move {
|
|
let selected_root = input
|
|
.thread_store
|
|
.get::<Vec<SelectedCapabilityRoot>>()
|
|
.and_then(|roots| roots.first().cloned())
|
|
.expect("selected root should be available");
|
|
self.lifecycle_observed
|
|
.lock()
|
|
.unwrap_or_else(std::sync::PoisonError::into_inner)
|
|
.push((input.thread_store.level_id().to_string(), selected_root.id));
|
|
input
|
|
.thread_store
|
|
.insert(Vec::<SelectedCapabilityRoot>::new());
|
|
})
|
|
}
|
|
}
|
|
|
|
impl codex_extension_api::McpServerContributor<Config> for InitialDataRecorder {
|
|
fn id(&self) -> &'static str {
|
|
"selected_root_test"
|
|
}
|
|
|
|
fn contribute<'a>(
|
|
&'a self,
|
|
_context: codex_extension_api::McpServerContributionContext<'a, Config>,
|
|
) -> codex_extension_api::ExtensionFuture<'a, Vec<codex_extension_api::McpServerContribution>>
|
|
{
|
|
Box::pin(async { Vec::new() })
|
|
}
|
|
|
|
fn selected_plugins<'a>(
|
|
&'a self,
|
|
context: codex_extension_api::McpServerContributionContext<'a, Config>,
|
|
_plugins_config: &'a codex_config::types::PluginsConfigToml,
|
|
) -> codex_extension_api::ExtensionFuture<'a, Vec<codex_extension_api::SelectedPlugin<'a>>>
|
|
{
|
|
Box::pin(async move {
|
|
assert_eq!(
|
|
context.selected_environments(),
|
|
Some([].as_slice()),
|
|
"thread MCP projection must preserve explicitly empty selections"
|
|
);
|
|
let thread_init = context
|
|
.thread_init()
|
|
.expect("initial MCP resolution should be thread-scoped");
|
|
let selected_root = thread_init
|
|
.get::<Vec<SelectedCapabilityRoot>>()
|
|
.and_then(|roots| roots.first().cloned())
|
|
.expect("selected root should be available");
|
|
self.mcp_observed
|
|
.lock()
|
|
.unwrap_or_else(std::sync::PoisonError::into_inner)
|
|
.push((
|
|
selected_root.id.clone(),
|
|
context
|
|
.session_source()
|
|
.expect("thread-scoped MCP resolution should identify its source")
|
|
.clone(),
|
|
));
|
|
let mut server = codex_mcp::codex_apps_mcp_server_config(
|
|
"https://selected.invalid",
|
|
/*apps_mcp_product_sku*/ None,
|
|
/*originator*/ None,
|
|
);
|
|
let CapabilityRootLocation::Environment { environment_id, .. } =
|
|
&selected_root.location;
|
|
let source_environment_id = environment_id.clone();
|
|
server.environment_id = source_environment_id.clone();
|
|
server.enabled = false;
|
|
let selected_root_id = selected_root.id;
|
|
let plugin_id = format!("plugin-{selected_root_id}");
|
|
let mcp_loaded = Arc::clone(&self.mcp_loaded);
|
|
vec![codex_extension_api::SelectedPlugin {
|
|
selected_root_id: selected_root_id.clone(),
|
|
plugin_id: plugin_id.clone(),
|
|
mcp: Box::pin(async move {
|
|
mcp_loaded.fetch_add(1, Ordering::Relaxed);
|
|
codex_extension_api::SelectedPluginContribution {
|
|
plugin_display_name: plugin_id,
|
|
source_environment_id,
|
|
connector_ids: vec![format!("{selected_root_id}-connector")],
|
|
servers: vec![(selected_root_id, server)],
|
|
}
|
|
}),
|
|
}]
|
|
})
|
|
}
|
|
}
|
|
|
|
let temp_dir = tempdir().expect("tempdir");
|
|
let mut config = test_config().await;
|
|
config.codex_home = temp_dir.path().join("codex-home").abs();
|
|
config.cwd = config.codex_home.abs();
|
|
config
|
|
.features
|
|
.enable(Feature::Apps)
|
|
.expect("test config should allow apps");
|
|
config
|
|
.features
|
|
.enable(Feature::Plugins)
|
|
.expect("enable plugins");
|
|
std::fs::create_dir_all(&config.codex_home).expect("create codex home");
|
|
|
|
let lifecycle_observed = Arc::new(std::sync::Mutex::new(Vec::new()));
|
|
let mcp_observed = Arc::new(std::sync::Mutex::new(Vec::new()));
|
|
let mcp_loaded = Arc::new(std::sync::atomic::AtomicUsize::new(0));
|
|
let recorder = Arc::new(InitialDataRecorder {
|
|
lifecycle_observed: Arc::clone(&lifecycle_observed),
|
|
mcp_observed: Arc::clone(&mcp_observed),
|
|
mcp_loaded: Arc::clone(&mcp_loaded),
|
|
});
|
|
let mut extensions = codex_extension_api::ExtensionRegistryBuilder::new();
|
|
extensions.thread_lifecycle_contributor(recorder.clone());
|
|
extensions.mcp_server_contributor(recorder.clone());
|
|
let auth_manager =
|
|
AuthManager::from_auth_for_testing(CodexAuth::create_dummy_chatgpt_auth_for_testing());
|
|
let manager = ThreadManager::new(
|
|
&config,
|
|
auth_manager.clone(),
|
|
build_models_manager(&config, auth_manager),
|
|
crate::CodexAppsToolsCache::default(),
|
|
SessionSource::Exec,
|
|
Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()),
|
|
Arc::new(extensions.build()),
|
|
Arc::new(crate::test_support::EmptyUserInstructionsProvider),
|
|
/*analytics_events_client*/ None,
|
|
passthrough_image_store(),
|
|
thread_store_from_config(&config, /*state_db*/ None),
|
|
/*agent_graph_store*/ None,
|
|
TEST_INSTALLATION_ID.to_string(),
|
|
/*attestation_provider*/ None,
|
|
/*external_time_provider*/ None,
|
|
);
|
|
let selected_root_init = |id: &str, environment_id: &str| {
|
|
let mut init = codex_extension_api::ExtensionDataInit::new();
|
|
init.insert(vec![SelectedCapabilityRoot {
|
|
id: id.to_string(),
|
|
location: CapabilityRootLocation::Environment {
|
|
environment_id: environment_id.to_string(),
|
|
path: PathUri::parse(&format!("file:///plugins/{id}")).expect("plugin root URI"),
|
|
},
|
|
}]);
|
|
init
|
|
};
|
|
|
|
let first_thread = manager
|
|
.start_thread(StartThreadOptions {
|
|
metrics_service_name: Some("codex_work_desktop".to_string()),
|
|
environments: Some(Vec::new()),
|
|
thread_extension_init: selected_root_init("selected-a", "env-a"),
|
|
..StartThreadOptions::new(config.clone())
|
|
})
|
|
.await
|
|
.expect("start first thread");
|
|
let second_session_source = SessionSource::SubAgent(SubAgentSource::Review);
|
|
let second_thread = manager
|
|
.start_thread(StartThreadOptions {
|
|
environments: Some(Vec::new()),
|
|
session_source: Some(second_session_source.clone()),
|
|
thread_extension_init: selected_root_init("selected-b", "env-b"),
|
|
..StartThreadOptions::new(config.clone())
|
|
})
|
|
.await
|
|
.expect("start second thread");
|
|
let first_session = &first_thread.thread.session;
|
|
let first_originator = first_session.originator().await;
|
|
let first_resolved = first_session
|
|
.services
|
|
.mcp_manager
|
|
.runtime_config_for_step(
|
|
&config,
|
|
&first_session.services.mcp_thread_init,
|
|
&first_session.services.thread_extension_data,
|
|
McpThreadIdentity {
|
|
auth_changed: false,
|
|
session_source: &SessionSource::Exec,
|
|
originator: &first_originator,
|
|
disabled_plugin_ids: &[],
|
|
environments: McpEnvironmentScope::Selected(
|
|
&first_session.services.turn_environments.selections(),
|
|
),
|
|
},
|
|
/*ready_selected_capability_roots*/ &[],
|
|
/*executor_capability_discovery*/ None,
|
|
/*executor_capability_discovery_v2*/ None,
|
|
)
|
|
.await;
|
|
let second_session = &second_thread.thread.session;
|
|
let second_originator = second_session.originator().await;
|
|
let second_resolved = second_session
|
|
.services
|
|
.mcp_manager
|
|
.runtime_config_for_step(
|
|
&config,
|
|
&second_session.services.mcp_thread_init,
|
|
&second_session.services.thread_extension_data,
|
|
McpThreadIdentity {
|
|
auth_changed: false,
|
|
session_source: &second_session_source,
|
|
originator: &second_originator,
|
|
disabled_plugin_ids: &[],
|
|
environments: McpEnvironmentScope::Selected(
|
|
&second_session.services.turn_environments.selections(),
|
|
),
|
|
},
|
|
/*ready_selected_capability_roots*/ &[],
|
|
/*executor_capability_discovery*/ None,
|
|
/*executor_capability_discovery_v2*/ None,
|
|
)
|
|
.await;
|
|
|
|
assert_eq!(
|
|
*lifecycle_observed
|
|
.lock()
|
|
.unwrap_or_else(std::sync::PoisonError::into_inner),
|
|
vec![
|
|
(first_thread.thread_id.to_string(), "selected-a".to_string()),
|
|
(
|
|
second_thread.thread_id.to_string(),
|
|
"selected-b".to_string()
|
|
),
|
|
]
|
|
);
|
|
assert_eq!(
|
|
*mcp_observed
|
|
.lock()
|
|
.unwrap_or_else(std::sync::PoisonError::into_inner),
|
|
vec![
|
|
("selected-a".to_string(), SessionSource::Exec),
|
|
("selected-b".to_string(), second_session_source.clone()),
|
|
("selected-a".to_string(), SessionSource::Exec),
|
|
("selected-b".to_string(), second_session_source),
|
|
]
|
|
);
|
|
let selected_servers = |config: &codex_mcp::McpConfig| {
|
|
codex_mcp::configured_mcp_servers(config)
|
|
.into_iter()
|
|
.filter(|(name, _)| name.starts_with("selected-"))
|
|
.map(|(name, server)| (name, server.environment_id))
|
|
.collect::<std::collections::BTreeMap<_, _>>()
|
|
};
|
|
assert_eq!(
|
|
selected_servers(&first_resolved.config),
|
|
std::collections::BTreeMap::from([("selected-a".to_string(), "env-a".to_string())])
|
|
);
|
|
assert_eq!(
|
|
selected_servers(&second_resolved.config),
|
|
std::collections::BTreeMap::from([("selected-b".to_string(), "env-b".to_string())])
|
|
);
|
|
for (config, name, source_environment_id) in [
|
|
(&first_resolved.config, "selected-a", "env-a"),
|
|
(&second_resolved.config, "selected-b", "env-b"),
|
|
] {
|
|
let server = config
|
|
.mcp_server_catalog
|
|
.server(name)
|
|
.expect("selected plugin server should be registered");
|
|
let plugin_id = format!("plugin-{name}");
|
|
let mut expected = codex_mcp::ResolvedMcpCatalog::builder();
|
|
expected.register(codex_mcp::McpServerRegistration::from_selected_plugin(
|
|
name.to_string(),
|
|
codex_mcp::McpPluginAttribution::new(plugin_id.clone(), plugin_id),
|
|
/*selection_order*/ 0,
|
|
source_environment_id,
|
|
server.config().clone(),
|
|
));
|
|
assert_eq!(Some(server), expected.build().server(name));
|
|
}
|
|
let codex_apps_server = codex_mcp::configured_mcp_servers(&first_resolved.config)
|
|
.remove(codex_mcp::CODEX_APPS_MCP_SERVER_NAME)
|
|
.expect("Codex Apps server should be configured");
|
|
let codex_apps_headers = match codex_apps_server.transport {
|
|
codex_config::McpServerTransportConfig::StreamableHttp { http_headers, .. } => http_headers,
|
|
codex_config::McpServerTransportConfig::Stdio { .. } => {
|
|
panic!("Codex Apps server should use streamable HTTP")
|
|
}
|
|
};
|
|
assert_eq!(
|
|
codex_apps_headers
|
|
.expect("Codex Apps headers should be configured")
|
|
.get("originator"),
|
|
Some(&"codex_work_desktop".to_string())
|
|
);
|
|
for (policy, disabled_plugin_ids, selected_enabled, expected_mcp_loads) in [
|
|
("", vec!["plugin-selected-a".to_string()], false, 1),
|
|
("", vec![], true, 1),
|
|
("[plugins._default]\nenabled = false", vec![], false, 0),
|
|
(
|
|
"[plugins._default]\nenabled = false\n[plugins.plugin-selected-a]\nenabled = true",
|
|
vec![],
|
|
true,
|
|
1,
|
|
),
|
|
] {
|
|
let mut config = config.clone();
|
|
config.config_layer_stack = config
|
|
.config_layer_stack
|
|
.with_user_config(
|
|
&config.codex_home.join("config.toml").abs(),
|
|
toml::from_str(policy).expect("plugin policy"),
|
|
)
|
|
.expect("plugin policy layers");
|
|
config.plugins = toml::from_str::<codex_config::config_toml::ConfigToml>(policy)
|
|
.unwrap()
|
|
.plugins;
|
|
let loaded_before = mcp_loaded.load(Ordering::Relaxed);
|
|
let projection = first_session
|
|
.services
|
|
.mcp_manager
|
|
.runtime_config_for_step(
|
|
&config,
|
|
&first_session.services.mcp_thread_init,
|
|
&first_session.services.thread_extension_data,
|
|
McpThreadIdentity {
|
|
auth_changed: false,
|
|
session_source: &SessionSource::Exec,
|
|
originator: &first_originator,
|
|
disabled_plugin_ids: &disabled_plugin_ids,
|
|
environments: McpEnvironmentScope::Selected(
|
|
&first_session.services.turn_environments.selections(),
|
|
),
|
|
},
|
|
/*ready_selected_capability_roots*/ &[],
|
|
/*executor_capability_discovery*/ None,
|
|
/*executor_capability_discovery_v2*/ None,
|
|
)
|
|
.await;
|
|
assert_eq!(
|
|
projection.selected_plugins.disabled_plugin_roots,
|
|
if selected_enabled {
|
|
vec![]
|
|
} else {
|
|
vec!["selected-a".to_string()]
|
|
}
|
|
);
|
|
assert_eq!(
|
|
selected_servers(&projection.config).contains_key("selected-a"),
|
|
selected_enabled
|
|
);
|
|
assert_eq!(
|
|
projection
|
|
.config
|
|
.connector_snapshot
|
|
.disabled_connector_ids()
|
|
.contains("selected-a-connector"),
|
|
!disabled_plugin_ids.is_empty()
|
|
);
|
|
assert_eq!(
|
|
projection.selected_plugins.plugins.len(),
|
|
usize::from(selected_enabled)
|
|
);
|
|
assert_eq!(
|
|
mcp_loaded.load(Ordering::Relaxed) - loaded_before,
|
|
expected_mcp_loads
|
|
);
|
|
}
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn selected_capability_roots_round_trip_through_fork() {
|
|
let temp_dir = tempdir().expect("tempdir");
|
|
let mut config = test_config().await;
|
|
config.codex_home = temp_dir.path().join("codex-home").abs();
|
|
config.cwd = config.codex_home.abs();
|
|
std::fs::create_dir_all(&config.codex_home).expect("create codex home");
|
|
|
|
let manager = ThreadManager::with_models_provider_and_home_for_tests(
|
|
CodexAuth::from_api_key("dummy"),
|
|
config.model_provider.clone(),
|
|
config.codex_home.to_path_buf(),
|
|
Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()),
|
|
);
|
|
let selected_roots = vec![SelectedCapabilityRoot {
|
|
id: "demo@1".to_string(),
|
|
location: CapabilityRootLocation::Environment {
|
|
environment_id: "build".to_string(),
|
|
path: PathUri::parse("file:///plugins/demo").expect("plugin root URI"),
|
|
},
|
|
}];
|
|
let inherited = manager
|
|
.start_thread(StartThreadOptions {
|
|
initial_history: InitialHistory::Forked(vec![RolloutItem::SessionMeta(
|
|
SessionMetaLine {
|
|
meta: SessionMeta {
|
|
selected_capability_roots: selected_roots.clone(),
|
|
..SessionMeta::default()
|
|
},
|
|
git: None,
|
|
},
|
|
)]),
|
|
environments: Some(Vec::new()),
|
|
..StartThreadOptions::new(config)
|
|
})
|
|
.await
|
|
.expect("start inherited fork");
|
|
inherited.thread.ensure_rollout_materialized().await;
|
|
inherited
|
|
.thread
|
|
.flush_rollout()
|
|
.await
|
|
.expect("flush inherited fork");
|
|
let inherited_history = RolloutRecorder::get_rollout_history(
|
|
&inherited
|
|
.thread
|
|
.rollout_path()
|
|
.expect("inherited fork rollout path"),
|
|
)
|
|
.await
|
|
.expect("read inherited fork rollout");
|
|
|
|
assert_eq!(
|
|
inherited_history.get_selected_capability_roots(),
|
|
selected_roots
|
|
);
|
|
}
|
|
|
|
/// Shared startup honors explicit root selections and only inherits roots when isolation permits.
|
|
#[tokio::test]
|
|
async fn selected_capability_roots_respect_startup_precedence_and_isolation() {
|
|
let temp_dir = tempdir().expect("tempdir");
|
|
let mut config = test_config().await;
|
|
config.codex_home = temp_dir.path().join("codex-home").abs();
|
|
config.cwd = config.codex_home.abs();
|
|
std::fs::create_dir_all(&config.codex_home).expect("create codex home");
|
|
let manager = ThreadManager::with_models_provider_and_home_for_tests(
|
|
CodexAuth::from_api_key("dummy"),
|
|
config.model_provider.clone(),
|
|
config.codex_home.to_path_buf(),
|
|
Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()),
|
|
);
|
|
let root = |id: &str| SelectedCapabilityRoot {
|
|
id: id.to_string(),
|
|
location: CapabilityRootLocation::Environment {
|
|
environment_id: "local".to_string(),
|
|
path: PathUri::from_abs_path(&config.cwd),
|
|
},
|
|
};
|
|
let mut parent_init = ExtensionDataInit::new();
|
|
parent_init.insert(vec![root("parent")]);
|
|
let parent = manager
|
|
.start_thread(StartThreadOptions {
|
|
thread_extension_init: parent_init,
|
|
..StartThreadOptions::new(config.clone())
|
|
})
|
|
.await
|
|
.expect("start parent");
|
|
let parent_environments = parent
|
|
.thread
|
|
.session
|
|
.services
|
|
.turn_environments
|
|
.snapshot()
|
|
.await;
|
|
|
|
for (isolation, explicit_roots, expected_roots) in [
|
|
(SessionIsolation::Inherit, None, vec![root("parent")]),
|
|
(
|
|
SessionIsolation::Inherit,
|
|
Some(vec![root("explicit")]),
|
|
vec![root("explicit")],
|
|
),
|
|
(SessionIsolation::Inherit, Some(Vec::new()), Vec::new()),
|
|
(SessionIsolation::Isolated, None, vec![root("saved")]),
|
|
(
|
|
SessionIsolation::Isolated,
|
|
Some(vec![root("explicit")]),
|
|
vec![root("explicit")],
|
|
),
|
|
] {
|
|
let mut thread_extension_init = ExtensionDataInit::new();
|
|
thread_extension_init.insert(isolation);
|
|
if let Some(roots) = explicit_roots {
|
|
thread_extension_init.insert(roots);
|
|
}
|
|
let child = manager
|
|
.start_thread(StartThreadOptions {
|
|
initial_history: InitialHistory::Forked(vec![RolloutItem::SessionMeta(
|
|
SessionMetaLine {
|
|
meta: SessionMeta {
|
|
selected_capability_roots: vec![root("saved")],
|
|
..SessionMeta::default()
|
|
},
|
|
git: None,
|
|
},
|
|
)]),
|
|
inherited_environments: Some(parent_environments.clone()),
|
|
thread_extension_init,
|
|
..StartThreadOptions::new(config.clone())
|
|
})
|
|
.await
|
|
.expect("start child");
|
|
assert_eq!(
|
|
child.thread.session.services.selected_capability_roots,
|
|
expected_roots
|
|
);
|
|
assert_eq!(
|
|
child
|
|
.thread
|
|
.session
|
|
.services
|
|
.turn_environments
|
|
.snapshot()
|
|
.await
|
|
.selected_capability_roots(),
|
|
expected_roots
|
|
);
|
|
child.thread.ensure_rollout_materialized().await;
|
|
child.thread.flush_rollout().await.expect("flush child");
|
|
let history = RolloutRecorder::get_rollout_history(
|
|
&child.thread.rollout_path().expect("child rollout path"),
|
|
)
|
|
.await
|
|
.expect("read child rollout");
|
|
assert_eq!(history.get_selected_capability_roots(), expected_roots);
|
|
}
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn resume_and_fork_do_not_restore_thread_environments_from_rollout() {
|
|
let temp_dir = tempdir().expect("tempdir");
|
|
let mut config = test_config().await;
|
|
config.codex_home = temp_dir.path().join("codex-home").abs();
|
|
config.cwd = config.codex_home.abs();
|
|
std::fs::create_dir_all(&config.codex_home).expect("create codex home");
|
|
|
|
let auth_manager =
|
|
AuthManager::from_auth_for_testing(CodexAuth::create_dummy_chatgpt_auth_for_testing());
|
|
let manager = ThreadManager::new(
|
|
&config,
|
|
auth_manager.clone(),
|
|
build_models_manager(&config, auth_manager.clone()),
|
|
crate::CodexAppsToolsCache::default(),
|
|
SessionSource::Exec,
|
|
Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()),
|
|
empty_extension_registry(),
|
|
Arc::new(crate::test_support::EmptyUserInstructionsProvider),
|
|
/*analytics_events_client*/ None,
|
|
passthrough_image_store(),
|
|
thread_store_from_config(&config, /*state_db*/ None),
|
|
/*agent_graph_store*/ None,
|
|
TEST_INSTALLATION_ID.to_string(),
|
|
/*attestation_provider*/ None,
|
|
/*external_time_provider*/ None,
|
|
);
|
|
let selected_cwd =
|
|
AbsolutePathBuf::try_from(config.cwd.as_path().join("selected")).expect("absolute path");
|
|
std::fs::create_dir_all(&selected_cwd).expect("create selected cwd");
|
|
let environments = vec![TurnEnvironmentSelection {
|
|
selected_capability_roots: Default::default(),
|
|
environment_id: "local".to_string(),
|
|
cwd: PathUri::from_abs_path(&selected_cwd),
|
|
workspace_roots: Vec::new(),
|
|
config: EnvironmentConfigState::FromThread,
|
|
}];
|
|
let default_cwd = config.cwd.clone();
|
|
let mut source_config = config.clone();
|
|
source_config.cwd = selected_cwd.clone();
|
|
let source = manager
|
|
.start_thread(StartThreadOptions {
|
|
history_mode: Some(ThreadHistoryMode::Legacy),
|
|
environments: Some(
|
|
environments
|
|
.clone()
|
|
.into_iter()
|
|
.map(TurnEnvironmentSelection::into_request)
|
|
.collect(),
|
|
),
|
|
..StartThreadOptions::new(source_config)
|
|
})
|
|
.await
|
|
.expect("start source thread");
|
|
source.thread.ensure_rollout_materialized().await;
|
|
source
|
|
.thread
|
|
.flush_rollout()
|
|
.await
|
|
.expect("flush source rollout");
|
|
let rollout_path = source
|
|
.thread
|
|
.rollout_path()
|
|
.expect("source rollout path should exist");
|
|
source
|
|
.thread
|
|
.shutdown_and_wait()
|
|
.await
|
|
.expect("shutdown source thread before resume");
|
|
let _ = manager.remove_thread(&source.thread_id).await;
|
|
|
|
let resumed = manager
|
|
.resume_legacy_thread_from_rollout(
|
|
config.clone(),
|
|
rollout_path.clone(),
|
|
auth_manager,
|
|
/*parent_trace*/ None,
|
|
ClientMcpExtensions::default(),
|
|
)
|
|
.await
|
|
.expect("resume source thread");
|
|
let (prepared_turn, _) = resumed
|
|
.thread
|
|
.session
|
|
.new_turn_with_sub_id(
|
|
"resume-turn".to_string(),
|
|
SessionSettingsUpdate::default(),
|
|
Default::default(),
|
|
)
|
|
.await
|
|
.expect("build resumed turn context");
|
|
let resumed_turn = prepared_turn;
|
|
assert_eq!(
|
|
resumed_turn
|
|
.initial_environments
|
|
.turn_environments()
|
|
.count(),
|
|
1
|
|
);
|
|
assert_eq!(
|
|
resumed_turn
|
|
.initial_environments
|
|
.primary()
|
|
.expect("primary environment")
|
|
.cwd(),
|
|
&PathUri::from_abs_path(&default_cwd)
|
|
);
|
|
assert_ne!(
|
|
resumed_turn
|
|
.initial_environments
|
|
.primary()
|
|
.expect("primary environment")
|
|
.cwd(),
|
|
&PathUri::from_abs_path(&selected_cwd)
|
|
);
|
|
|
|
let forked = manager
|
|
.fork_legacy_thread(
|
|
ForkSnapshot::Interrupted,
|
|
crate::StartThreadOptions::new(config),
|
|
rollout_path,
|
|
)
|
|
.await
|
|
.expect("fork source thread");
|
|
let (prepared_turn, _) = forked
|
|
.thread
|
|
.session
|
|
.new_turn_with_sub_id(
|
|
"fork-turn".to_string(),
|
|
SessionSettingsUpdate::default(),
|
|
Default::default(),
|
|
)
|
|
.await
|
|
.expect("build forked turn context");
|
|
let forked_turn = prepared_turn;
|
|
assert_eq!(
|
|
forked_turn.initial_environments.turn_environments().count(),
|
|
1
|
|
);
|
|
assert_eq!(
|
|
forked_turn
|
|
.initial_environments
|
|
.primary()
|
|
.expect("primary environment")
|
|
.cwd(),
|
|
&PathUri::from_abs_path(&default_cwd)
|
|
);
|
|
assert_ne!(
|
|
forked_turn
|
|
.initial_environments
|
|
.primary()
|
|
.expect("primary environment")
|
|
.cwd(),
|
|
&PathUri::from_abs_path(&selected_cwd)
|
|
);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn explicit_installation_id_skips_codex_home_file() {
|
|
let temp_dir = tempdir().expect("tempdir");
|
|
let mut config = test_config().await;
|
|
config.codex_home = temp_dir.path().join("codex-home").abs();
|
|
config.cwd = config.codex_home.abs();
|
|
std::fs::create_dir_all(&config.codex_home).expect("create codex home");
|
|
|
|
let auth_manager =
|
|
AuthManager::from_auth_for_testing(CodexAuth::create_dummy_chatgpt_auth_for_testing());
|
|
let installation_id = uuid::Uuid::new_v4().to_string();
|
|
let state_db = init_state_db(&config).await;
|
|
let thread_store = thread_store_from_config(&config, state_db.clone());
|
|
let manager = ThreadManager::new(
|
|
&config,
|
|
auth_manager.clone(),
|
|
build_models_manager(&config, auth_manager),
|
|
crate::CodexAppsToolsCache::default(),
|
|
SessionSource::Exec,
|
|
Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()),
|
|
empty_extension_registry(),
|
|
Arc::new(crate::test_support::EmptyUserInstructionsProvider),
|
|
/*analytics_events_client*/ None,
|
|
passthrough_image_store(),
|
|
thread_store,
|
|
local_agent_graph_store_from_state_db(state_db.as_ref()),
|
|
installation_id.clone(),
|
|
/*attestation_provider*/ None,
|
|
/*external_time_provider*/ None,
|
|
);
|
|
|
|
let thread = manager
|
|
.start_thread(StartThreadOptions::new(config.clone()))
|
|
.await
|
|
.expect("start thread with explicit installation id");
|
|
|
|
assert!(!config.codex_home.join(INSTALLATION_ID_FILENAME).exists());
|
|
assert_eq!(thread.thread.session.installation_id, installation_id);
|
|
|
|
thread
|
|
.thread
|
|
.shutdown_and_wait()
|
|
.await
|
|
.expect("shutdown thread");
|
|
let _ = manager.remove_thread(&thread.thread_id).await;
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn resume_active_thread_from_rollout_returns_running_thread() {
|
|
let temp_dir = tempdir().expect("tempdir");
|
|
let mut config = test_config().await;
|
|
config.codex_home = temp_dir.path().join("codex-home").abs();
|
|
config.cwd = config.codex_home.abs();
|
|
std::fs::create_dir_all(&config.codex_home).expect("create codex home");
|
|
|
|
let auth_manager =
|
|
AuthManager::from_auth_for_testing(CodexAuth::create_dummy_chatgpt_auth_for_testing());
|
|
let manager = ThreadManager::new(
|
|
&config,
|
|
auth_manager.clone(),
|
|
build_models_manager(&config, auth_manager.clone()),
|
|
crate::CodexAppsToolsCache::default(),
|
|
SessionSource::Exec,
|
|
Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()),
|
|
empty_extension_registry(),
|
|
Arc::new(crate::test_support::EmptyUserInstructionsProvider),
|
|
/*analytics_events_client*/ None,
|
|
passthrough_image_store(),
|
|
thread_store_from_config(&config, /*state_db*/ None),
|
|
/*agent_graph_store*/ None,
|
|
TEST_INSTALLATION_ID.to_string(),
|
|
/*attestation_provider*/ None,
|
|
/*external_time_provider*/ None,
|
|
);
|
|
|
|
let source = manager
|
|
.start_thread(StartThreadOptions {
|
|
history_mode: Some(ThreadHistoryMode::Legacy),
|
|
..StartThreadOptions::new(config.clone())
|
|
})
|
|
.await
|
|
.expect("start source thread");
|
|
source.thread.ensure_rollout_materialized().await;
|
|
source
|
|
.thread
|
|
.flush_rollout()
|
|
.await
|
|
.expect("flush source rollout");
|
|
let rollout_path = source
|
|
.thread
|
|
.rollout_path()
|
|
.expect("source rollout path should exist");
|
|
|
|
let resumed = manager
|
|
.resume_legacy_thread_from_rollout(
|
|
config,
|
|
rollout_path,
|
|
auth_manager,
|
|
/*parent_trace*/ None,
|
|
ClientMcpExtensions::default(),
|
|
)
|
|
.await
|
|
.expect("resume active source thread");
|
|
assert_eq!(resumed.thread_id, source.thread_id);
|
|
assert!(Arc::ptr_eq(&resumed.thread, &source.thread));
|
|
|
|
source
|
|
.thread
|
|
.shutdown_and_wait()
|
|
.await
|
|
.expect("shutdown source thread");
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn resume_stopped_thread_from_rollout_spawns_new_thread() {
|
|
let temp_dir = tempdir().expect("tempdir");
|
|
let mut config = test_config().await;
|
|
config.codex_home = temp_dir.path().join("codex-home").abs();
|
|
config.cwd = config.codex_home.abs();
|
|
std::fs::create_dir_all(&config.codex_home).expect("create codex home");
|
|
|
|
let auth_manager =
|
|
AuthManager::from_auth_for_testing(CodexAuth::create_dummy_chatgpt_auth_for_testing());
|
|
let manager = ThreadManager::new(
|
|
&config,
|
|
auth_manager.clone(),
|
|
build_models_manager(&config, auth_manager.clone()),
|
|
crate::CodexAppsToolsCache::default(),
|
|
SessionSource::Exec,
|
|
Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()),
|
|
empty_extension_registry(),
|
|
Arc::new(crate::test_support::EmptyUserInstructionsProvider),
|
|
/*analytics_events_client*/ None,
|
|
passthrough_image_store(),
|
|
thread_store_from_config(&config, /*state_db*/ None),
|
|
/*agent_graph_store*/ None,
|
|
TEST_INSTALLATION_ID.to_string(),
|
|
/*attestation_provider*/ None,
|
|
/*external_time_provider*/ None,
|
|
);
|
|
|
|
let source = manager
|
|
.start_thread(StartThreadOptions {
|
|
history_mode: Some(ThreadHistoryMode::Legacy),
|
|
..StartThreadOptions::new(config.clone())
|
|
})
|
|
.await
|
|
.expect("start source thread");
|
|
source.thread.ensure_rollout_materialized().await;
|
|
source
|
|
.thread
|
|
.flush_rollout()
|
|
.await
|
|
.expect("flush source rollout");
|
|
let rollout_path = source
|
|
.thread
|
|
.rollout_path()
|
|
.expect("source rollout path should exist");
|
|
source
|
|
.thread
|
|
.shutdown_and_wait()
|
|
.await
|
|
.expect("shutdown source thread");
|
|
|
|
let resumed = manager
|
|
.resume_legacy_thread_from_rollout(
|
|
config,
|
|
rollout_path,
|
|
auth_manager,
|
|
/*parent_trace*/ None,
|
|
ClientMcpExtensions::default(),
|
|
)
|
|
.await
|
|
.expect("resume stopped source thread");
|
|
assert_eq!(resumed.thread_id, source.thread_id);
|
|
assert!(!Arc::ptr_eq(&resumed.thread, &source.thread));
|
|
|
|
resumed
|
|
.thread
|
|
.shutdown_and_wait()
|
|
.await
|
|
.expect("shutdown resumed thread");
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn resume_stopped_thread_from_rollout_preserves_thread_source() {
|
|
let temp_dir = tempdir().expect("tempdir");
|
|
let mut config = test_config().await;
|
|
config.codex_home = temp_dir.path().join("codex-home").abs();
|
|
config.cwd = config.codex_home.abs();
|
|
std::fs::create_dir_all(&config.codex_home).expect("create codex home");
|
|
|
|
let auth_manager =
|
|
AuthManager::from_auth_for_testing(CodexAuth::create_dummy_chatgpt_auth_for_testing());
|
|
let state_db = init_state_db(&config).await;
|
|
let thread_store = thread_store_from_config(&config, state_db.clone());
|
|
let manager = ThreadManager::new(
|
|
&config,
|
|
auth_manager.clone(),
|
|
build_models_manager(&config, auth_manager.clone()),
|
|
crate::CodexAppsToolsCache::default(),
|
|
SessionSource::Exec,
|
|
Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()),
|
|
empty_extension_registry(),
|
|
Arc::new(crate::test_support::EmptyUserInstructionsProvider),
|
|
/*analytics_events_client*/ None,
|
|
passthrough_image_store(),
|
|
thread_store,
|
|
local_agent_graph_store_from_state_db(state_db.as_ref()),
|
|
TEST_INSTALLATION_ID.to_string(),
|
|
/*attestation_provider*/ None,
|
|
/*external_time_provider*/ None,
|
|
);
|
|
|
|
let source = manager
|
|
.start_thread(StartThreadOptions {
|
|
history_mode: Some(ThreadHistoryMode::Legacy),
|
|
thread_source: Some(ThreadSource::User),
|
|
environments: Some(Vec::new()),
|
|
..StartThreadOptions::new(config.clone())
|
|
})
|
|
.await
|
|
.expect("start source thread");
|
|
source.thread.ensure_rollout_materialized().await;
|
|
source
|
|
.thread
|
|
.flush_rollout()
|
|
.await
|
|
.expect("flush source rollout");
|
|
let rollout_path = source
|
|
.thread
|
|
.rollout_path()
|
|
.expect("source rollout path should exist");
|
|
source
|
|
.thread
|
|
.shutdown_and_wait()
|
|
.await
|
|
.expect("shutdown source thread before resume");
|
|
let _ = manager.remove_thread(&source.thread_id).await;
|
|
|
|
let resumed = manager
|
|
.resume_legacy_thread_from_rollout(
|
|
config,
|
|
rollout_path,
|
|
auth_manager,
|
|
/*parent_trace*/ None,
|
|
ClientMcpExtensions::default(),
|
|
)
|
|
.await
|
|
.expect("resume source thread");
|
|
|
|
assert_eq!(
|
|
resumed
|
|
.thread
|
|
.config_snapshot()
|
|
.await
|
|
.thread_source
|
|
.as_ref(),
|
|
Some(&ThreadSource::User)
|
|
);
|
|
|
|
resumed
|
|
.thread
|
|
.shutdown_and_wait()
|
|
.await
|
|
.expect("shutdown resumed thread");
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn subtree_listing_uses_injected_graph_store_without_state_db() {
|
|
let temp_dir = tempdir().expect("tempdir");
|
|
let mut config = test_config().await;
|
|
config.codex_home = temp_dir.path().join("codex-home").abs();
|
|
std::fs::create_dir_all(&config.codex_home).expect("create codex home");
|
|
|
|
let root_thread_id = ThreadId::new();
|
|
let descendant_thread_ids = vec![ThreadId::new(), ThreadId::new()];
|
|
let agent_graph_store = Arc::new(FakeAgentGraphStore {
|
|
root_thread_id,
|
|
descendant_thread_ids: descendant_thread_ids.clone(),
|
|
});
|
|
let auth_manager =
|
|
AuthManager::from_auth_for_testing(CodexAuth::create_dummy_chatgpt_auth_for_testing());
|
|
let manager = ThreadManager::new(
|
|
&config,
|
|
auth_manager.clone(),
|
|
build_models_manager(&config, auth_manager),
|
|
crate::CodexAppsToolsCache::default(),
|
|
SessionSource::Exec,
|
|
Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()),
|
|
empty_extension_registry(),
|
|
Arc::new(crate::test_support::EmptyUserInstructionsProvider),
|
|
/*analytics_events_client*/ None,
|
|
passthrough_image_store(),
|
|
thread_store_from_config(&config, /*state_db*/ None),
|
|
Some(agent_graph_store),
|
|
TEST_INSTALLATION_ID.to_string(),
|
|
/*attestation_provider*/ None,
|
|
/*external_time_provider*/ None,
|
|
);
|
|
|
|
let mut expected_thread_ids = vec![root_thread_id];
|
|
expected_thread_ids.extend(descendant_thread_ids);
|
|
assert_eq!(
|
|
manager
|
|
.list_agent_subtree_thread_ids(root_thread_id)
|
|
.await
|
|
.expect("subtree should load from injected graph store"),
|
|
expected_thread_ids
|
|
);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn rollout_path_resume_and_fork_read_history_through_thread_store() {
|
|
let temp_dir = tempdir().expect("tempdir");
|
|
let mut config = test_config().await;
|
|
config.codex_home = temp_dir.path().join("codex-home").abs();
|
|
config.cwd = config.codex_home.abs();
|
|
config.experimental_thread_store = ThreadStoreConfig::InMemory {
|
|
id: format!("thread-manager-{}", uuid::Uuid::new_v4()),
|
|
};
|
|
std::fs::create_dir_all(&config.codex_home).expect("create codex home");
|
|
|
|
let auth_manager =
|
|
AuthManager::from_auth_for_testing(CodexAuth::create_dummy_chatgpt_auth_for_testing());
|
|
let state_db = init_state_db(&config).await;
|
|
let thread_store = thread_store_from_config(&config, state_db.clone());
|
|
let in_memory_store = thread_store
|
|
.as_any()
|
|
.downcast_ref::<InMemoryThreadStore>()
|
|
.expect("configured in-memory store");
|
|
let manager = ThreadManager::new(
|
|
&config,
|
|
auth_manager.clone(),
|
|
build_models_manager(&config, auth_manager.clone()),
|
|
crate::CodexAppsToolsCache::default(),
|
|
SessionSource::Exec,
|
|
Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()),
|
|
empty_extension_registry(),
|
|
Arc::new(crate::test_support::EmptyUserInstructionsProvider),
|
|
/*analytics_events_client*/ None,
|
|
passthrough_image_store(),
|
|
thread_store.clone(),
|
|
local_agent_graph_store_from_state_db(state_db.as_ref()),
|
|
TEST_INSTALLATION_ID.to_string(),
|
|
/*attestation_provider*/ None,
|
|
/*external_time_provider*/ None,
|
|
);
|
|
|
|
let source = manager
|
|
.start_thread(StartThreadOptions::new(config.clone()))
|
|
.await
|
|
.expect("start source thread");
|
|
source
|
|
.thread
|
|
.shutdown_and_wait()
|
|
.await
|
|
.expect("shutdown source thread");
|
|
let _ = manager.remove_thread(&source.thread_id).await;
|
|
|
|
let rollout_path = config
|
|
.codex_home
|
|
.join("rollouts/source.jsonl")
|
|
.to_path_buf();
|
|
let resumed = manager
|
|
.resume_thread_with_history(
|
|
config.clone(),
|
|
InitialHistory::Resumed(ResumedHistory {
|
|
history_revision: None,
|
|
conversation_id: source.thread_id,
|
|
history: Arc::new(vec![RolloutItem::ResponseItem(user_msg("hello").into())]),
|
|
rollout_path: Some(rollout_path.clone()),
|
|
}),
|
|
auth_manager.clone(),
|
|
/*parent_trace*/ None,
|
|
ClientMcpExtensions::default(),
|
|
)
|
|
.await
|
|
.expect("seed rollout path in store");
|
|
resumed
|
|
.thread
|
|
.shutdown_and_wait()
|
|
.await
|
|
.expect("shutdown seeded resumed thread");
|
|
let _ = manager.remove_thread(&resumed.thread_id).await;
|
|
|
|
let resumed_from_path = manager
|
|
.resume_legacy_thread_from_rollout(
|
|
config.clone(),
|
|
rollout_path.clone(),
|
|
auth_manager,
|
|
/*parent_trace*/ None,
|
|
ClientMcpExtensions::default(),
|
|
)
|
|
.await
|
|
.expect("resume from rollout path");
|
|
assert_eq!(resumed_from_path.thread_id, resumed.thread_id);
|
|
|
|
let forked = manager
|
|
.fork_legacy_thread(
|
|
ForkSnapshot::Interrupted,
|
|
crate::StartThreadOptions::new(config),
|
|
rollout_path,
|
|
)
|
|
.await
|
|
.expect("fork from rollout path");
|
|
assert_ne!(forked.thread_id, resumed.thread_id);
|
|
|
|
let calls = in_memory_store.calls().await;
|
|
assert_eq!(calls.read_thread_by_rollout_path, 2);
|
|
|
|
resumed_from_path
|
|
.thread
|
|
.shutdown_and_wait()
|
|
.await
|
|
.expect("shutdown path-resumed thread");
|
|
forked
|
|
.thread
|
|
.shutdown_and_wait()
|
|
.await
|
|
.expect("shutdown forked thread");
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn metadata_update_without_result_reads_only_when_the_caller_needs_the_thread() {
|
|
let temp_dir = tempdir().expect("tempdir");
|
|
let mut config = test_config().await;
|
|
config.codex_home = temp_dir.path().join("codex-home").abs();
|
|
config.cwd = config.codex_home.abs();
|
|
config.experimental_thread_store = ThreadStoreConfig::InMemory {
|
|
id: format!("metadata-update-none-{}", uuid::Uuid::new_v4()),
|
|
};
|
|
std::fs::create_dir_all(&config.codex_home).expect("create codex home");
|
|
|
|
let auth_manager =
|
|
AuthManager::from_auth_for_testing(CodexAuth::create_dummy_chatgpt_auth_for_testing());
|
|
let thread_store = thread_store_from_config(&config, /*state_db*/ None);
|
|
let in_memory_store = thread_store
|
|
.as_any()
|
|
.downcast_ref::<InMemoryThreadStore>()
|
|
.expect("configured in-memory store");
|
|
let manager = ThreadManager::new(
|
|
&config,
|
|
auth_manager.clone(),
|
|
build_models_manager(&config, auth_manager),
|
|
crate::CodexAppsToolsCache::default(),
|
|
SessionSource::Exec,
|
|
Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()),
|
|
empty_extension_registry(),
|
|
Arc::new(crate::test_support::EmptyUserInstructionsProvider),
|
|
/*analytics_events_client*/ None,
|
|
passthrough_image_store(),
|
|
thread_store.clone(),
|
|
/*agent_graph_store*/ None,
|
|
TEST_INSTALLATION_ID.to_string(),
|
|
/*attestation_provider*/ None,
|
|
/*external_time_provider*/ None,
|
|
);
|
|
let started = manager
|
|
.start_thread(StartThreadOptions::new(config))
|
|
.await
|
|
.expect("start thread");
|
|
started
|
|
.thread
|
|
.flush_rollout()
|
|
.await
|
|
.expect("flush initial metadata");
|
|
manager
|
|
.update_thread_metadata(
|
|
started.thread_id,
|
|
ThreadMetadataPatch {
|
|
name: Some(Some("initial name".to_string())),
|
|
..Default::default()
|
|
},
|
|
/*include_archived*/ false,
|
|
)
|
|
.await
|
|
.expect("flush pending live metadata before measuring calls");
|
|
in_memory_store.omit_metadata_update_result_for_testing();
|
|
|
|
let before_loaded_update = in_memory_store.calls().await;
|
|
let loaded = manager
|
|
.update_thread_metadata(
|
|
started.thread_id,
|
|
ThreadMetadataPatch {
|
|
name: Some(Some("loaded name".to_string())),
|
|
..Default::default()
|
|
},
|
|
/*include_archived*/ false,
|
|
)
|
|
.await
|
|
.expect("update loaded thread metadata");
|
|
assert_eq!(loaded.name.as_deref(), Some("loaded name"));
|
|
let after_loaded_update = in_memory_store.calls().await;
|
|
assert_eq!(
|
|
after_loaded_update.update_thread_metadata,
|
|
before_loaded_update.update_thread_metadata + 1
|
|
);
|
|
assert_eq!(
|
|
after_loaded_update.read_thread,
|
|
before_loaded_update.read_thread + 1
|
|
);
|
|
|
|
started
|
|
.thread
|
|
.append_rollout_items(&[RolloutItem::EventMsg(EventMsg::UserMessage(
|
|
UserMessageEvent {
|
|
message: "completion-only metadata".to_string(),
|
|
..Default::default()
|
|
},
|
|
))])
|
|
.await
|
|
.expect("append item with derived metadata");
|
|
let after_completion_only_update = in_memory_store.calls().await;
|
|
assert_eq!(
|
|
after_completion_only_update.update_thread_metadata,
|
|
after_loaded_update.update_thread_metadata + 1
|
|
);
|
|
assert_eq!(
|
|
after_completion_only_update.read_thread,
|
|
after_loaded_update.read_thread
|
|
);
|
|
|
|
started
|
|
.thread
|
|
.shutdown_and_wait()
|
|
.await
|
|
.expect("shutdown loaded thread");
|
|
let _ = manager.remove_thread(&started.thread_id).await;
|
|
let before_cold_update = in_memory_store.calls().await;
|
|
let cold = manager
|
|
.update_thread_metadata(
|
|
started.thread_id,
|
|
ThreadMetadataPatch {
|
|
name: Some(Some("cold name".to_string())),
|
|
..Default::default()
|
|
},
|
|
/*include_archived*/ false,
|
|
)
|
|
.await
|
|
.expect("update cold thread metadata");
|
|
assert_eq!(cold.name.as_deref(), Some("cold name"));
|
|
let after_cold_update = in_memory_store.calls().await;
|
|
assert_eq!(
|
|
after_cold_update.update_thread_metadata,
|
|
before_cold_update.update_thread_metadata + 1
|
|
);
|
|
assert_eq!(
|
|
after_cold_update.read_thread,
|
|
before_cold_update.read_thread + 1
|
|
);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn new_uses_active_provider_for_model_refresh() {
|
|
let server = MockServer::start().await;
|
|
let models_mock = mount_models_once(&server, ModelsResponse { models: vec![] }).await;
|
|
|
|
let temp_dir = tempdir().expect("tempdir");
|
|
let mut config = test_config().await;
|
|
config.codex_home = temp_dir.path().join("codex-home").abs();
|
|
config.cwd = config.codex_home.abs();
|
|
std::fs::create_dir_all(&config.codex_home).expect("create codex home");
|
|
config.model_catalog = None;
|
|
config.model_provider.base_url = Some(server.uri());
|
|
|
|
let auth_manager =
|
|
AuthManager::from_auth_for_testing(CodexAuth::create_dummy_chatgpt_auth_for_testing());
|
|
let manager = ThreadManager::new(
|
|
&config,
|
|
auth_manager.clone(),
|
|
build_models_manager(&config, auth_manager),
|
|
crate::CodexAppsToolsCache::default(),
|
|
SessionSource::Exec,
|
|
Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()),
|
|
empty_extension_registry(),
|
|
Arc::new(crate::test_support::EmptyUserInstructionsProvider),
|
|
/*analytics_events_client*/ None,
|
|
passthrough_image_store(),
|
|
thread_store_from_config(&config, /*state_db*/ None),
|
|
/*agent_graph_store*/ None,
|
|
TEST_INSTALLATION_ID.to_string(),
|
|
/*attestation_provider*/ None,
|
|
/*external_time_provider*/ None,
|
|
);
|
|
|
|
let _ = manager
|
|
.list_models(
|
|
RefreshStrategy::Online,
|
|
crate::test_support::default_http_client_factory(),
|
|
)
|
|
.await;
|
|
assert_eq!(models_mock.requests().len(), 1);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn injected_models_manager_controls_refresh_policy() {
|
|
let server = MockServer::start().await;
|
|
let _ = mount_models_once(&server, ModelsResponse { models: vec![] }).await;
|
|
let _ = mount_models_once(&server, ModelsResponse { models: vec![] }).await;
|
|
|
|
let temp_dir = tempdir().expect("tempdir");
|
|
let mut config = test_config().await;
|
|
config.codex_home = temp_dir.path().join("codex-home").abs();
|
|
config.cwd = config.codex_home.abs();
|
|
std::fs::create_dir_all(&config.codex_home).expect("create codex home");
|
|
config.model_catalog = None;
|
|
config.model_provider.base_url = Some(server.uri());
|
|
|
|
let auth_manager =
|
|
AuthManager::from_auth_for_testing(CodexAuth::create_dummy_chatgpt_auth_for_testing());
|
|
let provider = create_model_provider(
|
|
config.model_provider.clone(),
|
|
Some(Arc::clone(&auth_manager)),
|
|
);
|
|
let models_manager = provider.models_manager_without_cache(config.model_catalog.clone());
|
|
let manager = ThreadManager::new(
|
|
&config,
|
|
auth_manager,
|
|
models_manager,
|
|
crate::CodexAppsToolsCache::default(),
|
|
SessionSource::Custom("test-embedder".to_string()),
|
|
Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()),
|
|
empty_extension_registry(),
|
|
Arc::new(crate::test_support::EmptyUserInstructionsProvider),
|
|
/*analytics_events_client*/ None,
|
|
passthrough_image_store(),
|
|
thread_store_from_config(&config, /*state_db*/ None),
|
|
/*agent_graph_store*/ None,
|
|
TEST_INSTALLATION_ID.to_string(),
|
|
/*attestation_provider*/ None,
|
|
/*external_time_provider*/ None,
|
|
);
|
|
|
|
let http_client_factory = crate::test_support::default_http_client_factory();
|
|
let _ = manager
|
|
.list_models(
|
|
RefreshStrategy::OnlineIfUncached,
|
|
http_client_factory.clone(),
|
|
)
|
|
.await;
|
|
let _ = manager
|
|
.list_models(RefreshStrategy::OnlineIfUncached, http_client_factory)
|
|
.await;
|
|
|
|
assert_eq!(
|
|
server.received_requests().await.unwrap_or_default().len(),
|
|
2
|
|
);
|
|
assert!(!config.codex_home.join("models_cache.json").exists());
|
|
}
|
|
|
|
#[test]
|
|
fn interrupted_fork_snapshot_appends_interrupt_boundary() {
|
|
let committed_history =
|
|
InitialHistory::Forked(vec![RolloutItem::ResponseItem(user_msg("hello").into())]);
|
|
|
|
assert_eq!(
|
|
serde_json::to_value(
|
|
append_interrupted_boundary(
|
|
committed_history,
|
|
/*turn_id*/ None,
|
|
/*root_turn_id*/ None,
|
|
/*started_at*/ None,
|
|
InterruptedTurnHistoryMarker::ContextualUser,
|
|
)
|
|
.get_rollout_items()
|
|
)
|
|
.expect("serialize interrupted fork history"),
|
|
serde_json::to_value(vec![
|
|
RolloutItem::ResponseItem(user_msg("hello").into()),
|
|
RolloutItem::ResponseItem(contextual_user_interrupted_marker().into()),
|
|
RolloutItem::EventMsg(EventMsg::TurnAborted(TurnAbortedEvent {
|
|
root_turn_id: None,
|
|
turn_id: None,
|
|
started_at: None,
|
|
reason: TurnAbortReason::Interrupted,
|
|
error: None,
|
|
completed_at: None,
|
|
duration_ms: None,
|
|
})),
|
|
])
|
|
.expect("serialize expected interrupted fork history"),
|
|
);
|
|
assert_eq!(
|
|
serde_json::to_value(
|
|
append_interrupted_boundary(
|
|
InitialHistory::New,
|
|
/*turn_id*/ None,
|
|
/*root_turn_id*/ None,
|
|
/*started_at*/ None,
|
|
InterruptedTurnHistoryMarker::ContextualUser,
|
|
)
|
|
.get_rollout_items()
|
|
)
|
|
.expect("serialize interrupted empty fork history"),
|
|
serde_json::to_value(vec![
|
|
RolloutItem::ResponseItem(contextual_user_interrupted_marker().into()),
|
|
RolloutItem::EventMsg(EventMsg::TurnAborted(TurnAbortedEvent {
|
|
root_turn_id: None,
|
|
turn_id: None,
|
|
started_at: None,
|
|
reason: TurnAbortReason::Interrupted,
|
|
error: None,
|
|
completed_at: None,
|
|
duration_ms: None,
|
|
})),
|
|
])
|
|
.expect("serialize expected interrupted empty history"),
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn disabled_interrupted_fork_snapshot_appends_only_interrupt_event() {
|
|
let committed_history =
|
|
InitialHistory::Forked(vec![RolloutItem::ResponseItem(user_msg("hello").into())]);
|
|
|
|
assert_eq!(
|
|
serde_json::to_value(
|
|
append_interrupted_boundary(
|
|
committed_history,
|
|
/*turn_id*/ None,
|
|
/*root_turn_id*/ None,
|
|
/*started_at*/ None,
|
|
InterruptedTurnHistoryMarker::Disabled,
|
|
)
|
|
.get_rollout_items()
|
|
)
|
|
.expect("serialize disabled interrupted fork history"),
|
|
serde_json::to_value(vec![
|
|
RolloutItem::ResponseItem(user_msg("hello").into()),
|
|
RolloutItem::EventMsg(EventMsg::TurnAborted(TurnAbortedEvent {
|
|
root_turn_id: None,
|
|
turn_id: None,
|
|
started_at: None,
|
|
reason: TurnAbortReason::Interrupted,
|
|
error: None,
|
|
completed_at: None,
|
|
duration_ms: None,
|
|
})),
|
|
])
|
|
.expect("serialize expected disabled interrupted fork history"),
|
|
);
|
|
assert_eq!(
|
|
serde_json::to_value(
|
|
append_interrupted_boundary(
|
|
InitialHistory::New,
|
|
/*turn_id*/ None,
|
|
/*root_turn_id*/ None,
|
|
/*started_at*/ None,
|
|
InterruptedTurnHistoryMarker::Disabled,
|
|
)
|
|
.get_rollout_items()
|
|
)
|
|
.expect("serialize disabled interrupted empty fork history"),
|
|
serde_json::to_value(vec![RolloutItem::EventMsg(EventMsg::TurnAborted(
|
|
TurnAbortedEvent {
|
|
root_turn_id: None,
|
|
turn_id: None,
|
|
started_at: None,
|
|
reason: TurnAbortReason::Interrupted,
|
|
error: None,
|
|
completed_at: None,
|
|
duration_ms: None,
|
|
},
|
|
))])
|
|
.expect("serialize expected disabled interrupted empty fork history"),
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn interrupted_snapshot_is_not_mid_turn() {
|
|
let interrupted_history = InitialHistory::Forked(vec![
|
|
RolloutItem::ResponseItem(user_msg("hello").into()),
|
|
RolloutItem::ResponseItem(assistant_msg("partial").into()),
|
|
RolloutItem::ResponseItem(contextual_user_interrupted_marker().into()),
|
|
RolloutItem::EventMsg(EventMsg::TurnAborted(TurnAbortedEvent {
|
|
root_turn_id: None,
|
|
turn_id: Some("turn-1".to_string()),
|
|
started_at: None,
|
|
reason: TurnAbortReason::Interrupted,
|
|
error: None,
|
|
completed_at: None,
|
|
duration_ms: None,
|
|
})),
|
|
]);
|
|
|
|
assert_eq!(
|
|
snapshot_turn_state(&interrupted_history),
|
|
SnapshotTurnState {
|
|
ends_mid_turn: false,
|
|
active_turn_id: None,
|
|
active_turn_started_at: None,
|
|
active_turn_root_id: None,
|
|
active_turn_start_index: None,
|
|
},
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn multi_agent_v2_interrupted_marker_uses_developer_input_message() {
|
|
assert_eq!(
|
|
developer_interrupted_marker(),
|
|
ResponseItem::Message {
|
|
status: None,
|
|
encrypted_content: None,
|
|
id: None,
|
|
role: "developer".to_string(),
|
|
content: vec![ContentItem::InputText {
|
|
text: format!(
|
|
"<turn_aborted>\n{}\n</turn_aborted>",
|
|
crate::context::TurnAborted::INTERRUPTED_DEVELOPER_GUIDANCE
|
|
),
|
|
}],
|
|
phase: None,
|
|
internal_chat_message_metadata_passthrough: Some(
|
|
InternalChatMessageMetadataPassthrough {
|
|
content_item_kinds: Some(vec![ContentItemKind(
|
|
"generic.turn_aborted".to_string()
|
|
)]),
|
|
content_item_metadata: Some(vec![ContentItemMetadata::harness()]),
|
|
..Default::default()
|
|
}
|
|
),
|
|
}
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn completed_legacy_event_history_is_not_mid_turn() {
|
|
let completed_history = InitialHistory::Forked(vec![
|
|
RolloutItem::EventMsg(EventMsg::UserMessage(UserMessageEvent {
|
|
client_id: None,
|
|
message: "hello".to_string(),
|
|
images: None,
|
|
text_elements: Vec::new(),
|
|
local_images: Vec::new(),
|
|
..Default::default()
|
|
})),
|
|
RolloutItem::EventMsg(EventMsg::AgentMessage(AgentMessageEvent {
|
|
message: "done".to_string(),
|
|
phase: None,
|
|
memory_citation: None,
|
|
delivery: None,
|
|
questions: None,
|
|
})),
|
|
]);
|
|
|
|
assert_eq!(
|
|
snapshot_turn_state(&completed_history),
|
|
SnapshotTurnState {
|
|
ends_mid_turn: false,
|
|
active_turn_id: None,
|
|
active_turn_started_at: None,
|
|
active_turn_root_id: None,
|
|
active_turn_start_index: None,
|
|
},
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn mixed_response_and_legacy_user_event_history_is_mid_turn() {
|
|
let mixed_history = InitialHistory::Forked(vec![
|
|
RolloutItem::ResponseItem(user_msg("hello").into()),
|
|
RolloutItem::EventMsg(EventMsg::UserMessage(UserMessageEvent {
|
|
client_id: None,
|
|
message: "hello".to_string(),
|
|
images: None,
|
|
text_elements: Vec::new(),
|
|
local_images: Vec::new(),
|
|
..Default::default()
|
|
})),
|
|
]);
|
|
|
|
assert_eq!(
|
|
snapshot_turn_state(&mixed_history),
|
|
SnapshotTurnState {
|
|
ends_mid_turn: true,
|
|
active_turn_id: None,
|
|
active_turn_started_at: None,
|
|
active_turn_root_id: None,
|
|
active_turn_start_index: None,
|
|
},
|
|
);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn interrupted_fork_snapshot_does_not_synthesize_turn_id_for_legacy_history() {
|
|
let temp_dir = tempdir().expect("tempdir");
|
|
let mut config = test_config().await;
|
|
config.codex_home = temp_dir.path().join("codex-home").abs();
|
|
config.cwd = config.codex_home.abs();
|
|
std::fs::create_dir_all(&config.codex_home).expect("create codex home");
|
|
|
|
let auth_manager =
|
|
AuthManager::from_auth_for_testing(CodexAuth::create_dummy_chatgpt_auth_for_testing());
|
|
let state_db = init_state_db(&config).await;
|
|
let manager = ThreadManager::new(
|
|
&config,
|
|
auth_manager.clone(),
|
|
build_models_manager(&config, auth_manager.clone()),
|
|
crate::CodexAppsToolsCache::default(),
|
|
SessionSource::Exec,
|
|
Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()),
|
|
empty_extension_registry(),
|
|
Arc::new(crate::test_support::EmptyUserInstructionsProvider),
|
|
/*analytics_events_client*/ None,
|
|
passthrough_image_store(),
|
|
thread_store_from_config(&config, state_db.clone()),
|
|
local_agent_graph_store_from_state_db(state_db.as_ref()),
|
|
TEST_INSTALLATION_ID.to_string(),
|
|
/*attestation_provider*/ None,
|
|
/*external_time_provider*/ None,
|
|
);
|
|
|
|
let source = manager
|
|
.start_thread(StartThreadOptions {
|
|
history_mode: Some(ThreadHistoryMode::Legacy),
|
|
initial_history: InitialHistory::Forked(vec![
|
|
RolloutItem::ResponseItem(user_msg("hello").into()),
|
|
RolloutItem::ResponseItem(assistant_msg("partial").into()),
|
|
]),
|
|
..StartThreadOptions::new(config.clone())
|
|
})
|
|
.await
|
|
.expect("create source thread from completed history");
|
|
let source_path = source
|
|
.thread
|
|
.rollout_path()
|
|
.expect("source rollout path should exist");
|
|
let source_history = RolloutRecorder::get_rollout_history(&source_path)
|
|
.await
|
|
.expect("read source rollout history");
|
|
let source_snapshot_state = snapshot_turn_state(&source_history);
|
|
assert!(source_snapshot_state.ends_mid_turn);
|
|
let expected_turn_id = source_snapshot_state.active_turn_id.clone();
|
|
assert_eq!(expected_turn_id, None);
|
|
|
|
let forked = manager
|
|
.fork_legacy_thread(
|
|
ForkSnapshot::Interrupted,
|
|
crate::StartThreadOptions::new(config.clone()),
|
|
source_path,
|
|
)
|
|
.await
|
|
.expect("fork interrupted snapshot");
|
|
let forked_path = forked
|
|
.thread
|
|
.rollout_path()
|
|
.expect("forked rollout path should exist");
|
|
let history = RolloutRecorder::get_rollout_history(&forked_path)
|
|
.await
|
|
.expect("read forked rollout history");
|
|
assert!(!snapshot_turn_state(&history).ends_mid_turn);
|
|
let rollout_items: Vec<_> = history
|
|
.get_rollout_items()
|
|
.iter()
|
|
.filter(|item| !matches!(item, RolloutItem::SessionMeta(_)))
|
|
.collect();
|
|
let interrupted_marker_json = serde_json::to_value(RolloutItem::ResponseItem(
|
|
contextual_user_interrupted_marker().into(),
|
|
))
|
|
.expect("serialize interrupted marker");
|
|
let interrupted_abort_json = serde_json::to_value(RolloutItem::EventMsg(
|
|
EventMsg::TurnAborted(TurnAbortedEvent {
|
|
root_turn_id: None,
|
|
turn_id: expected_turn_id,
|
|
started_at: None,
|
|
reason: TurnAbortReason::Interrupted,
|
|
error: None,
|
|
completed_at: None,
|
|
duration_ms: None,
|
|
}),
|
|
))
|
|
.expect("serialize interrupted abort event");
|
|
assert_eq!(
|
|
rollout_items
|
|
.iter()
|
|
.filter(|item| {
|
|
strip_response_item_ids_from_json(
|
|
serde_json::to_value(item).expect("serialize rollout item"),
|
|
) == interrupted_marker_json
|
|
})
|
|
.count(),
|
|
1,
|
|
);
|
|
assert_eq!(
|
|
rollout_items
|
|
.iter()
|
|
.filter(|item| {
|
|
serde_json::to_value(item).expect("serialize rollout item")
|
|
== interrupted_abort_json
|
|
})
|
|
.count(),
|
|
1,
|
|
);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn interrupted_fork_snapshot_preserves_explicit_turn_id() {
|
|
let temp_dir = tempdir().expect("tempdir");
|
|
let mut config = test_config().await;
|
|
config.codex_home = temp_dir.path().join("codex-home").abs();
|
|
config.cwd = config.codex_home.abs();
|
|
std::fs::create_dir_all(&config.codex_home).expect("create codex home");
|
|
|
|
let auth_manager =
|
|
AuthManager::from_auth_for_testing(CodexAuth::create_dummy_chatgpt_auth_for_testing());
|
|
let state_db = init_state_db(&config).await;
|
|
let manager = ThreadManager::new(
|
|
&config,
|
|
auth_manager.clone(),
|
|
build_models_manager(&config, auth_manager.clone()),
|
|
crate::CodexAppsToolsCache::default(),
|
|
SessionSource::Exec,
|
|
Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()),
|
|
empty_extension_registry(),
|
|
Arc::new(crate::test_support::EmptyUserInstructionsProvider),
|
|
/*analytics_events_client*/ None,
|
|
passthrough_image_store(),
|
|
thread_store_from_config(&config, state_db.clone()),
|
|
local_agent_graph_store_from_state_db(state_db.as_ref()),
|
|
TEST_INSTALLATION_ID.to_string(),
|
|
/*attestation_provider*/ None,
|
|
/*external_time_provider*/ None,
|
|
);
|
|
|
|
let source = manager
|
|
.start_thread(StartThreadOptions {
|
|
history_mode: Some(ThreadHistoryMode::Legacy),
|
|
initial_history: InitialHistory::Forked(vec![
|
|
RolloutItem::EventMsg(EventMsg::TurnStarted(TurnStartedEvent {
|
|
turn_attribution: None,
|
|
turn_id: "turn-explicit".to_string(),
|
|
root_turn_id: None,
|
|
trace_id: None,
|
|
started_at: None,
|
|
model_context_window: None,
|
|
collaboration_mode_kind: Default::default(),
|
|
})),
|
|
RolloutItem::ResponseItem(user_msg("hello").into()),
|
|
RolloutItem::ResponseItem(assistant_msg("partial").into()),
|
|
]),
|
|
..StartThreadOptions::new(config.clone())
|
|
})
|
|
.await
|
|
.expect("create source thread from explicit partial history");
|
|
let source_path = source
|
|
.thread
|
|
.rollout_path()
|
|
.expect("source rollout path should exist");
|
|
let source_history = RolloutRecorder::get_rollout_history(&source_path)
|
|
.await
|
|
.expect("read source rollout history");
|
|
let source_snapshot_state = snapshot_turn_state(&source_history);
|
|
assert_eq!(
|
|
source_snapshot_state,
|
|
SnapshotTurnState {
|
|
ends_mid_turn: true,
|
|
active_turn_id: Some("turn-explicit".to_string()),
|
|
active_turn_started_at: None,
|
|
active_turn_root_id: None,
|
|
active_turn_start_index: Some(1),
|
|
},
|
|
);
|
|
|
|
let forked = manager
|
|
.fork_legacy_thread(
|
|
ForkSnapshot::Interrupted,
|
|
crate::StartThreadOptions::new(config.clone()),
|
|
source_path,
|
|
)
|
|
.await
|
|
.expect("fork interrupted snapshot");
|
|
let forked_path = forked
|
|
.thread
|
|
.rollout_path()
|
|
.expect("forked rollout path should exist");
|
|
let history = RolloutRecorder::get_rollout_history(&forked_path)
|
|
.await
|
|
.expect("read forked rollout history");
|
|
let rollout_items: Vec<_> = history
|
|
.get_rollout_items()
|
|
.iter()
|
|
.filter(|item| !matches!(item, RolloutItem::SessionMeta(_)))
|
|
.collect();
|
|
|
|
assert!(rollout_items.iter().any(|item| {
|
|
matches!(
|
|
item,
|
|
RolloutItem::EventMsg(EventMsg::TurnAborted(TurnAbortedEvent {
|
|
root_turn_id: None,
|
|
turn_id: Some(turn_id),
|
|
started_at: None,
|
|
reason: TurnAbortReason::Interrupted,
|
|
error: None,
|
|
completed_at: None,
|
|
duration_ms: None,
|
|
})) if turn_id == "turn-explicit"
|
|
)
|
|
}));
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn interrupted_fork_snapshot_uses_persisted_mid_turn_history_without_live_source() {
|
|
let temp_dir = tempdir().expect("tempdir");
|
|
let mut config = test_config().await;
|
|
config.codex_home = temp_dir.path().join("codex-home").abs();
|
|
config.cwd = config.codex_home.abs();
|
|
std::fs::create_dir_all(&config.codex_home).expect("create codex home");
|
|
|
|
let auth_manager =
|
|
AuthManager::from_auth_for_testing(CodexAuth::create_dummy_chatgpt_auth_for_testing());
|
|
let state_db = init_state_db(&config).await;
|
|
let manager = ThreadManager::new(
|
|
&config,
|
|
auth_manager.clone(),
|
|
build_models_manager(&config, auth_manager.clone()),
|
|
crate::CodexAppsToolsCache::default(),
|
|
SessionSource::Exec,
|
|
Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()),
|
|
empty_extension_registry(),
|
|
Arc::new(crate::test_support::EmptyUserInstructionsProvider),
|
|
/*analytics_events_client*/ None,
|
|
passthrough_image_store(),
|
|
thread_store_from_config(&config, state_db.clone()),
|
|
local_agent_graph_store_from_state_db(state_db.as_ref()),
|
|
TEST_INSTALLATION_ID.to_string(),
|
|
/*attestation_provider*/ None,
|
|
/*external_time_provider*/ None,
|
|
);
|
|
|
|
let source = manager
|
|
.start_thread(StartThreadOptions {
|
|
history_mode: Some(ThreadHistoryMode::Legacy),
|
|
initial_history: InitialHistory::Forked(vec![
|
|
RolloutItem::ResponseItem(user_msg("hello").into()),
|
|
RolloutItem::ResponseItem(assistant_msg("partial").into()),
|
|
]),
|
|
..StartThreadOptions::new(config.clone())
|
|
})
|
|
.await
|
|
.expect("create source thread from partial history");
|
|
let source_path = source
|
|
.thread
|
|
.rollout_path()
|
|
.expect("source rollout path should exist");
|
|
let source_history = RolloutRecorder::get_rollout_history(&source_path)
|
|
.await
|
|
.expect("read source rollout history");
|
|
assert!(snapshot_turn_state(&source_history).ends_mid_turn);
|
|
manager.remove_thread(&source.thread_id).await;
|
|
|
|
let forked = manager
|
|
.fork_legacy_thread(
|
|
ForkSnapshot::Interrupted,
|
|
crate::StartThreadOptions::new(config.clone()),
|
|
source_path,
|
|
)
|
|
.await
|
|
.expect("fork interrupted snapshot");
|
|
let forked_path = forked
|
|
.thread
|
|
.rollout_path()
|
|
.expect("forked rollout path should exist");
|
|
let history = RolloutRecorder::get_rollout_history(&forked_path)
|
|
.await
|
|
.expect("read forked rollout history");
|
|
assert!(!snapshot_turn_state(&history).ends_mid_turn);
|
|
|
|
let forked_rollout_items: Vec<_> = history
|
|
.get_rollout_items()
|
|
.iter()
|
|
.filter(|item| !matches!(item, RolloutItem::SessionMeta(_)))
|
|
.collect();
|
|
let interrupted_marker_json = serde_json::to_value(RolloutItem::ResponseItem(
|
|
contextual_user_interrupted_marker().into(),
|
|
))
|
|
.expect("serialize interrupted marker");
|
|
assert_eq!(
|
|
forked_rollout_items
|
|
.iter()
|
|
.filter(|item| {
|
|
strip_response_item_ids_from_json(
|
|
serde_json::to_value(item).expect("serialize forked rollout item"),
|
|
) == interrupted_marker_json
|
|
})
|
|
.count(),
|
|
1,
|
|
);
|
|
|
|
manager.remove_thread(&forked.thread_id).await;
|
|
let reforked = manager
|
|
.fork_legacy_thread(
|
|
ForkSnapshot::Interrupted,
|
|
crate::StartThreadOptions::new(config.clone()),
|
|
forked_path,
|
|
)
|
|
.await
|
|
.expect("re-fork interrupted snapshot");
|
|
let reforked_path = reforked
|
|
.thread
|
|
.rollout_path()
|
|
.expect("re-forked rollout path should exist");
|
|
let reforked_history = RolloutRecorder::get_rollout_history(&reforked_path)
|
|
.await
|
|
.expect("read re-forked rollout history");
|
|
let reforked_rollout_items: Vec<_> = reforked_history
|
|
.get_rollout_items()
|
|
.iter()
|
|
.filter(|item| !matches!(item, RolloutItem::SessionMeta(_)))
|
|
.collect();
|
|
|
|
assert_eq!(
|
|
reforked_rollout_items
|
|
.iter()
|
|
.filter(|item| {
|
|
strip_response_item_ids_from_json(
|
|
serde_json::to_value(item).expect("serialize re-forked rollout item"),
|
|
) == interrupted_marker_json
|
|
})
|
|
.count(),
|
|
1,
|
|
);
|
|
assert_eq!(
|
|
reforked_rollout_items
|
|
.iter()
|
|
.filter(|item| {
|
|
matches!(
|
|
item,
|
|
RolloutItem::EventMsg(EventMsg::TurnAborted(TurnAbortedEvent {
|
|
reason: TurnAbortReason::Interrupted,
|
|
..
|
|
}))
|
|
)
|
|
})
|
|
.count(),
|
|
1,
|
|
);
|
|
}
|