mod thread_list_cwd_filter_tests { use super::super::normalize_thread_list_cwd_filters; use codex_app_server_protocol::ThreadListCwdFilter; use codex_utils_absolute_path::AbsolutePathBuf; use pretty_assertions::assert_eq; use std::path::PathBuf; #[test] fn normalize_thread_list_cwd_filter_preserves_absolute_paths() { let cwd = if cfg!(windows) { String::from(r"C:\srv\repo-b") } else { String::from("/srv/repo-b") }; assert_eq!( normalize_thread_list_cwd_filters(Some(ThreadListCwdFilter::One(cwd.clone()))) .expect("cwd filter should parse"), Some(vec![PathBuf::from(cwd)]) ); } #[test] fn normalize_thread_list_cwd_filter_resolves_relative_paths_against_server_cwd() -> std::io::Result<()> { let expected = AbsolutePathBuf::relative_to_current_dir("repo-b")?.to_path_buf(); assert_eq!( normalize_thread_list_cwd_filters(Some(ThreadListCwdFilter::Many(vec![String::from( "repo-b" ),]))) .expect("cwd filter should parse"), Some(vec![expected]) ); Ok(()) }}mod background_terminal_pagination_tests { use super::super::paginate_background_terminals; use codex_app_server_protocol::ThreadBackgroundTerminal; use codex_utils_absolute_path::AbsolutePathBuf; use pretty_assertions::assert_eq; fn terminal(process_id: &str) -> ThreadBackgroundTerminal { let cwd = if cfg!(windows) { r"C:\tmp" } else { "/tmp" }; ThreadBackgroundTerminal { item_id: format!("item-{process_id}"), process_id: process_id.to_string(), command: format!("command-{process_id}"), cwd: AbsolutePathBuf::from_absolute_path(cwd).expect("absolute cwd"), os_pid: None, cpu_percent: None, rss_kb: None, } } #[test] fn paginates_with_process_id_cursor() { let terminals = vec![ terminal("1"), terminal("2"), terminal("3"), terminal("4"), terminal("5"), ]; let (data, next_cursor) = paginate_background_terminals(&terminals, /*cursor*/ None, Some(2)) .expect("valid page"); assert_eq!(data, vec![terminal("1"), terminal("2")]); assert_eq!(next_cursor, Some("2".to_string())); let first_cursor = next_cursor; let terminals_without_anchor = vec![terminal("1"), terminal("3"), terminal("4")]; let (data, next_cursor) = paginate_background_terminals(&terminals_without_anchor, first_cursor.clone(), Some(2)) .expect("valid page"); assert_eq!(data, vec![terminal("3"), terminal("4")]); assert_eq!(next_cursor, None); let (data, next_cursor) = paginate_background_terminals(&terminals, first_cursor, Some(2)).expect("valid page"); assert_eq!(data, vec![terminal("3"), terminal("4")]); assert_eq!(next_cursor, Some("4".to_string())); assert!( paginate_background_terminals(&terminals, Some("missing".to_string()), Some(1)) .is_err() ); }}mod thread_processor_behavior_tests { async fn forked_from_id_from_rollout(path: &Path) -> Option<String> { codex_core::read_session_meta_line(path) .await .ok() .and_then(|meta_line| meta_line.meta.forked_from_id) .map(|thread_id| thread_id.to_string()) } use super::super::*; use crate::outgoing_message::OutgoingEnvelope; use crate::outgoing_message::OutgoingMessage; use anyhow::Result; use chrono::DateTime; use chrono::Utc; use codex_app_server_protocol::ServerRequestPayload; use codex_app_server_protocol::ThreadItem; use codex_app_server_protocol::ToolRequestUserInputParams; use codex_config::CloudConfigBundleLoader; use codex_config::LoaderOverrides; use codex_config::SessionThreadConfig; use codex_config::StaticThreadConfigLoader; use codex_config::ThreadConfigSource; use codex_model_provider_info::ModelProviderInfo; use codex_model_provider_info::WireApi; use codex_protocol::ThreadId; use codex_protocol::config_types::CollaborationMode; use codex_protocol::config_types::ModeKind; use codex_protocol::config_types::Settings; use codex_protocol::models::BUILT_IN_PERMISSION_PROFILE_DANGER_FULL_ACCESS; use codex_protocol::models::BUILT_IN_PERMISSION_PROFILE_READ_ONLY; use codex_protocol::models::BUILT_IN_PERMISSION_PROFILE_WORKSPACE; use codex_protocol::models::PermissionProfile; use codex_protocol::openai_models::ReasoningEffort; use codex_protocol::permissions::FileSystemAccessMode; use codex_protocol::permissions::FileSystemPath; use codex_protocol::permissions::FileSystemSandboxEntry; use codex_protocol::permissions::NetworkSandboxPolicy; use codex_protocol::protocol::AskForApproval; use codex_protocol::protocol::SessionSource; use codex_protocol::protocol::SubAgentSource; use codex_protocol::protocol::TurnEnvironmentSelections; use codex_state::ThreadMetadataBuilder; use codex_thread_store::StoredThread; use codex_utils_absolute_path::test_support::PathBufExt; use codex_utils_absolute_path::test_support::test_path_buf; use pretty_assertions::assert_eq; use serde_json::Value; use serde_json::json; use std::collections::BTreeMap; use std::path::PathBuf; use std::sync::Arc; use tempfile::TempDir; fn dynamic_tool( namespace: Option<&str>, name: impl Into<String>, input_schema: Value, defer_loading: bool, ) -> DynamicToolSpec { let function = DynamicToolFunctionSpec { name: name.into(), description: "test".to_string(), input_schema, defer_loading, }; match namespace { Some(namespace) => { DynamicToolSpec::Namespace(codex_app_server_protocol::DynamicToolNamespaceSpec { name: namespace.to_string(), description: "test namespace".to_string(), tools: vec![DynamicToolNamespaceTool::Function(function)], }) } None => DynamicToolSpec::Function(function), } } #[test] fn validate_dynamic_tools_rejects_unsupported_input_schema() { let tools = vec![dynamic_tool( /*namespace*/ None, "my_tool", json!({"type": "null"}), /*defer_loading*/ false, )]; let err = validate_dynamic_tools(&tools).expect_err("invalid schema"); assert!(err.contains("my_tool"), "unexpected error: {err}"); } #[test] fn validate_dynamic_tools_accepts_sanitizable_input_schema() { let tools = vec![dynamic_tool( /*namespace*/ None, "my_tool", // Missing `type` is common; core sanitizes these to a supported schema. json!({"properties": {}}), /*defer_loading*/ false, )]; validate_dynamic_tools(&tools).expect("valid schema"); } #[test] fn validate_dynamic_tools_accepts_nullable_field_schema() { let tools = vec![dynamic_tool( /*namespace*/ None, "my_tool", json!({ "type": "object", "properties": { "query": {"type": ["string", "null"]} }, "required": ["query"], "additionalProperties": false }), /*defer_loading*/ false, )]; validate_dynamic_tools(&tools).expect("valid schema"); } #[test] fn validate_dynamic_tools_accepts_same_name_in_different_namespaces() { let tools = vec![ dynamic_tool( Some("codex_app"), "my_tool", json!({ "type": "object", "properties": {}, "additionalProperties": false }), /*defer_loading*/ true, ), dynamic_tool( Some("other_app"), "my_tool", json!({ "type": "object", "properties": {}, "additionalProperties": false }), /*defer_loading*/ true, ), ]; validate_dynamic_tools(&tools).expect("valid schema"); } #[test] fn validate_dynamic_tools_accepts_responses_compatible_identifiers() { let tools = vec![dynamic_tool( Some("Codex-App_2"), "lookup-ticket_2", json!({ "type": "object", "properties": {}, "additionalProperties": false }), /*defer_loading*/ true, )]; validate_dynamic_tools(&tools).expect("valid schema"); } #[test] fn validate_dynamic_tools_rejects_duplicate_name_in_same_namespace() { let function = || DynamicToolFunctionSpec { name: "my_tool".to_string(), description: "test".to_string(), input_schema: json!({ "type": "object", "properties": {}, "additionalProperties": false }), defer_loading: true, }; let tools = vec![DynamicToolSpec::Namespace( codex_app_server_protocol::DynamicToolNamespaceSpec { name: "codex_app".to_string(), description: "test namespace".to_string(), tools: vec![ DynamicToolNamespaceTool::Function(function()), DynamicToolNamespaceTool::Function(function()), ], }, )]; let err = validate_dynamic_tools(&tools).expect_err("duplicate name"); assert!(err.contains("codex_app"), "unexpected error: {err}"); assert!(err.contains("my_tool"), "unexpected error: {err}"); } #[test] fn thread_turns_list_merges_in_progress_active_turn_before_agent_status_running() { let persisted_items = vec![RolloutItem::EventMsg(EventMsg::UserMessage( codex_protocol::protocol::UserMessageEvent { client_id: None, message: "persisted".to_string(), images: None, local_images: Vec::new(), text_elements: Vec::new(), ..Default::default() }, ))]; let active_turn = Turn { id: "live-turn".to_string(), items: vec![ThreadItem::UserMessage { id: "live-user-message".to_string(), client_id: None, content: vec![V2UserInput::Text { text: "live".to_string(), text_elements: Vec::new(), }], }], items_view: TurnItemsView::Full, error: None, status: TurnStatus::InProgress, started_at: None, completed_at: None, duration_ms: None, }; let turns = reconstruct_thread_turns_for_turns_list( &persisted_items, ThreadStatus::Idle, /*has_live_running_thread*/ false, Some(active_turn.clone()), ); assert_eq!(turns.last(), Some(&active_turn)); } #[test] fn validate_dynamic_tools_rejects_empty_namespace() { let tools = vec![dynamic_tool( Some(""), "my_tool", json!({ "type": "object", "properties": {}, "additionalProperties": false }), /*defer_loading*/ false, )]; let err = validate_dynamic_tools(&tools).expect_err("empty namespace"); assert!(err.contains("namespace"), "unexpected error: {err}"); } #[test] fn validate_dynamic_tools_rejects_reserved_namespace() { let tools = vec![dynamic_tool( Some("mcp__server__"), "my_tool", json!({ "type": "object", "properties": {}, "additionalProperties": false }), /*defer_loading*/ false, )]; let err = validate_dynamic_tools(&tools).expect_err("reserved namespace"); assert!(err.contains("reserved"), "unexpected error: {err}"); } #[test] fn validate_dynamic_tools_rejects_name_not_supported_by_responses() { let tools = vec![dynamic_tool( /*namespace*/ None, "lookup.ticket", json!({ "type": "object", "properties": {}, "additionalProperties": false }), /*defer_loading*/ false, )]; let err = validate_dynamic_tools(&tools).expect_err("invalid name"); assert!(err.contains("lookup.ticket"), "unexpected error: {err}"); assert!( err.contains("Responses API") && err.contains("^[a-zA-Z0-9_-]+$"), "unexpected error: {err}" ); } #[test] fn validate_dynamic_tools_rejects_namespace_not_supported_by_responses() { let tools = vec![dynamic_tool( Some("codex.app"), "lookup_ticket", json!({ "type": "object", "properties": {}, "additionalProperties": false }), /*defer_loading*/ true, )]; let err = validate_dynamic_tools(&tools).expect_err("invalid namespace"); assert!(err.contains("codex.app"), "unexpected error: {err}"); assert!( err.contains("Responses API") && err.contains("^[a-zA-Z0-9_-]+$"), "unexpected error: {err}" ); } #[test] fn validate_dynamic_tools_rejects_name_longer_than_responses_limit() { let long_name = "a".repeat(129); let tools = vec![dynamic_tool( /*namespace*/ None, long_name.clone(), json!({ "type": "object", "properties": {}, "additionalProperties": false }), /*defer_loading*/ false, )]; let err = validate_dynamic_tools(&tools).expect_err("name too long"); assert!(err.contains("at most 128"), "unexpected error: {err}"); assert!(err.contains(&long_name), "unexpected error: {err}"); } #[test] fn validate_dynamic_tools_rejects_namespace_fields_over_limits() { let long_namespace = "a".repeat(65); let mut tools = vec![dynamic_tool( Some(&long_namespace), "lookup_ticket", json!({ "type": "object", "properties": {}, "additionalProperties": false }), /*defer_loading*/ true, )]; let err = validate_dynamic_tools(&tools).expect_err("namespace too long"); assert!(err.contains("at most 64"), "unexpected error: {err}"); assert!(err.contains(&long_namespace), "unexpected error: {err}"); let DynamicToolSpec::Namespace(namespace) = &mut tools[0] else { unreachable!("expected namespace") }; namespace.name = "tickets".to_string(); namespace.description = "a".repeat(1025); let err = validate_dynamic_tools(&tools).expect_err("namespace description too long"); assert!(err.contains("at most 1024"), "unexpected error: {err}"); } #[test] fn validate_dynamic_tools_rejects_reserved_responses_namespace() { let tools = vec![dynamic_tool( Some("functions"), "lookup_ticket", json!({ "type": "object", "properties": {}, "additionalProperties": false }), /*defer_loading*/ true, )]; let err = validate_dynamic_tools(&tools).expect_err("reserved Responses namespace"); assert!(err.contains("functions"), "unexpected error: {err}"); assert!(err.contains("Responses API"), "unexpected error: {err}"); } #[test] fn summary_from_stored_thread_preserves_millisecond_precision() { let created_at = DateTime::parse_from_rfc3339("2025-01-02T03:04:05.678Z").expect("valid timestamp"); let updated_at = DateTime::parse_from_rfc3339("2025-01-02T03:04:06.789Z").expect("valid timestamp"); let thread_id = ThreadId::from_string("00000000-0000-0000-0000-000000000123").expect("valid thread"); let stored_thread = StoredThread { thread_id, extra_config: None, rollout_path: Some(PathBuf::from("/tmp/thread.jsonl")), forked_from_id: None, parent_thread_id: None, preview: "preview".to_string(), name: None, model_provider: "openai".to_string(), model: None, reasoning_effort: None, created_at: created_at.with_timezone(&Utc), updated_at: updated_at.with_timezone(&Utc), archived_at: None, cwd: PathBuf::from("/tmp"), cli_version: "0.0.0".to_string(), source: SessionSource::Cli, thread_source: Some(codex_protocol::protocol::ThreadSource::User), agent_nickname: None, agent_role: None, agent_path: None, git_info: None, approval_mode: AskForApproval::OnRequest, permission_profile: PermissionProfile::read_only(), token_usage: None, first_user_message: Some("first user message".to_string()), history: None, }; let summary = summary_from_stored_thread(stored_thread, "fallback"); assert_eq!( summary.timestamp.as_deref(), Some("2025-01-02T03:04:05.678Z") ); assert_eq!( summary.updated_at.as_deref(), Some("2025-01-02T03:04:06.789Z") ); } #[test] fn requested_permissions_trust_project_uses_permission_profile_intent() { let cwd = test_path_buf("/tmp/project").abs(); let full_access_profile = codex_protocol::models::PermissionProfile::Disabled; let workspace_write_profile = codex_protocol::models::PermissionProfile::workspace_write(); let read_only_profile = codex_protocol::models::PermissionProfile::read_only(); let split_write_profile = codex_protocol::models::PermissionProfile::from_runtime_permissions( &FileSystemSandboxPolicy::restricted(vec![ FileSystemSandboxEntry { path: FileSystemPath::Path { path: cwd.clone() }, access: FileSystemAccessMode::Write, }, FileSystemSandboxEntry { path: FileSystemPath::GlobPattern { pattern: "/tmp/project/**/*.env".to_string(), }, access: FileSystemAccessMode::Deny, }, ]), NetworkSandboxPolicy::Restricted, ); assert!(requested_permissions_trust_project( &ConfigOverrides { permission_profile: Some(full_access_profile), ..Default::default() }, cwd.as_path() )); assert!(requested_permissions_trust_project( &ConfigOverrides { permission_profile: Some(workspace_write_profile), ..Default::default() }, cwd.as_path() )); assert!(requested_permissions_trust_project( &ConfigOverrides { permission_profile: Some(split_write_profile), ..Default::default() }, cwd.as_path() )); assert!(requested_permissions_trust_project( &ConfigOverrides { default_permissions: Some(BUILT_IN_PERMISSION_PROFILE_WORKSPACE.to_string()), ..Default::default() }, cwd.as_path() )); assert!(requested_permissions_trust_project( &ConfigOverrides { default_permissions: Some( BUILT_IN_PERMISSION_PROFILE_DANGER_FULL_ACCESS.to_string() ), ..Default::default() }, cwd.as_path() )); assert!(!requested_permissions_trust_project( &ConfigOverrides { permission_profile: Some(read_only_profile), ..Default::default() }, cwd.as_path() )); assert!(!requested_permissions_trust_project( &ConfigOverrides { default_permissions: Some(BUILT_IN_PERMISSION_PROFILE_READ_ONLY.to_string()), ..Default::default() }, cwd.as_path() )); } #[test] fn config_load_error_marks_cloud_config_bundle_failures_for_relogin() { let err = std::io::Error::other(CloudConfigBundleLoadError::new( CloudConfigBundleLoadErrorCode::Auth, Some(401), "Your authentication session could not be refreshed automatically. Please log out and sign in again.", )); let error = config_load_error(&err); assert_eq!( error.data, Some(json!({ "reason": "cloudConfigBundle", "errorCode": "Auth", "action": "relogin", "statusCode": 401, "detail": "Your authentication session could not be refreshed automatically. Please log out and sign in again.", })) ); assert!( error.message.contains("failed to load configuration"), "unexpected error message: {}", error.message ); } #[test] fn config_load_error_leaves_non_cloud_config_bundle_failures_unmarked() { let err = std::io::Error::other("required MCP servers failed to initialize"); let error = config_load_error(&err); assert_eq!(error.data, None); assert!( error.message.contains("failed to load configuration"), "unexpected error message: {}", error.message ); } #[test] fn config_load_error_marks_non_auth_cloud_config_bundle_failures_without_relogin() { let err = std::io::Error::other(CloudConfigBundleLoadError::new( CloudConfigBundleLoadErrorCode::RequestFailed, /*status_code*/ None, "Failed to load cloud config bundle (workspace-managed policies).", )); let error = config_load_error(&err); assert_eq!( error.data, Some(json!({ "reason": "cloudConfigBundle", "errorCode": "RequestFailed", "detail": "Failed to load cloud config bundle (workspace-managed policies).", })) ); } #[test] fn config_load_error_marks_invalid_cloud_config_bundle_failures_without_relogin() { let err = std::io::Error::other(CloudConfigBundleLoadError::new( CloudConfigBundleLoadErrorCode::InvalidBundle, /*status_code*/ None, "invalid cloud config bundle: invalid cloud config fragment Base policy (cfg_123)", )); let error = config_load_error(&err); assert_eq!( error.data, Some(json!({ "reason": "cloudConfigBundle", "errorCode": "InvalidBundle", "detail": "invalid cloud config bundle: invalid cloud config fragment Base policy (cfg_123)", })) ); } #[tokio::test] async fn derive_config_from_params_uses_session_thread_config_model_provider() -> Result<()> { let temp_dir = TempDir::new()?; let session_provider = ModelProviderInfo { name: "session".to_string(), base_url: Some("http://127.0.0.1:8061/api/codex".to_string()), env_key: None, env_key_instructions: None, experimental_bearer_token: None, auth: None, aws: None, wire_api: WireApi::Responses, query_params: None, http_headers: None, env_http_headers: None, request_max_retries: None, stream_max_retries: None, stream_idle_timeout_ms: None, websocket_connect_timeout_ms: None, requires_openai_auth: false, supports_websockets: true, }; let config_manager = ConfigManager::new( temp_dir.path().to_path_buf(), Vec::new(), LoaderOverrides::default(), /*strict_config*/ false, CloudConfigBundleLoader::default(), Arg0DispatchPaths::default(), Arc::new(StaticThreadConfigLoader::new(vec![ ThreadConfigSource::Session(SessionThreadConfig { model_provider: Some("session".to_string()), model_providers: HashMap::from([( "session".to_string(), session_provider.clone(), )]), features: BTreeMap::from([("plugins".to_string(), false)]), }), ])), ); let config = config_manager .load_with_overrides( Some(HashMap::from([ ("model_provider".to_string(), json!("request")), ("features.plugins".to_string(), json!(true)), ("bypass_hook_trust".to_string(), json!(true)), ( "model_providers.session".to_string(), json!({ "name": "request", "base_url": "http://127.0.0.1:9999/api/codex", "wire_api": "responses", }), ), ])), ConfigOverrides::default(), ) .await?; assert_eq!(config.model_provider_id, "session"); assert_eq!(config.model_provider, session_provider); assert!(!config.features.enabled(Feature::Plugins)); assert!(config.bypass_hook_trust); Ok(()) } #[test] fn collect_resume_override_mismatches_includes_service_tier() { let cwd = test_path_buf("/tmp").abs(); let request = ThreadResumeParams { thread_id: "thread-1".to_string(), history: None, path: None, model: None, model_provider: None, service_tier: Some(Some("priority".to_string())), cwd: None, runtime_workspace_roots: None, approval_policy: None, approvals_reviewer: None, sandbox: None, permissions: None, config: None, base_instructions: None, developer_instructions: None, personality: None, exclude_turns: false, initial_turns_page: None, }; let config_snapshot = ThreadConfigSnapshot { model: "gpt-5".to_string(), model_provider_id: "openai".to_string(), service_tier: Some("flex".to_string()), approval_policy: codex_protocol::protocol::AskForApproval::OnRequest, approvals_reviewer: codex_protocol::config_types::ApprovalsReviewer::User, permission_profile: codex_protocol::models::PermissionProfile::Disabled, active_permission_profile: None, environments: TurnEnvironmentSelections::new(cwd, Vec::new()), workspace_roots: Vec::new(), profile_workspace_roots: Vec::new(), ephemeral: false, reasoning_effort: None, reasoning_summary: None, personality: None, collaboration_mode: CollaborationMode { mode: ModeKind::Default, settings: Settings { model: "gpt-5".to_string(), reasoning_effort: None, developer_instructions: None, }, }, session_source: SessionSource::Cli, forked_from_thread_id: None, parent_thread_id: None, thread_source: None, }; assert_eq!( collect_resume_override_mismatches(&request, &config_snapshot), vec!["service_tier requested=Some(\"priority\") active=Some(\"flex\")".to_string()] ); } fn test_thread_metadata( model: Option<&str>, reasoning_effort: Option<ReasoningEffort>, ) -> Result<ThreadMetadata> { let thread_id = ThreadId::from_string("3f941c35-29b3-493b-b0a4-e25800d9aeb0")?; let mut builder = ThreadMetadataBuilder::new( thread_id, PathBuf::from("/tmp/rollout.jsonl"), Utc::now(), codex_protocol::protocol::SessionSource::default(), ); builder.model_provider = Some("mock_provider".to_string()); let mut metadata = builder.build("mock_provider"); metadata.model = model.map(ToString::to_string); metadata.reasoning_effort = reasoning_effort; Ok(metadata) } #[test] fn summary_from_thread_metadata_formats_protocol_timestamps_as_seconds() -> Result<()> { let mut metadata = test_thread_metadata(/*model*/ None, /*reasoning_effort*/ None)?; metadata.created_at = DateTime::parse_from_rfc3339("2025-09-05T16:53:11.123Z")?.with_timezone(&Utc); metadata.updated_at = DateTime::parse_from_rfc3339("2025-09-05T16:53:12.456Z")?.with_timezone(&Utc); let summary = summary_from_thread_metadata(&metadata); assert_eq!(summary.timestamp, Some("2025-09-05T16:53:11Z".to_string())); assert_eq!(summary.updated_at, Some("2025-09-05T16:53:12Z".to_string())); Ok(()) } #[test] fn merge_persisted_resume_metadata_prefers_persisted_model_and_reasoning_effort() -> Result<()> { let mut request_overrides = None; let mut typesafe_overrides = ConfigOverrides::default(); let persisted_metadata = test_thread_metadata(Some("gpt-5.1-codex-max"), Some(ReasoningEffort::High))?; merge_persisted_resume_metadata( &mut request_overrides, &mut typesafe_overrides, &persisted_metadata, ); assert_eq!( typesafe_overrides.model, Some("gpt-5.1-codex-max".to_string()) ); assert_eq!( typesafe_overrides.model_provider, Some("mock_provider".to_string()) ); assert_eq!( request_overrides, Some(HashMap::from([( "model_reasoning_effort".to_string(), serde_json::Value::String("high".to_string()), )])) ); Ok(()) } #[test] fn merge_persisted_resume_metadata_preserves_explicit_overrides() -> Result<()> { let mut request_overrides = Some(HashMap::from([( "model_reasoning_effort".to_string(), serde_json::Value::String("low".to_string()), )])); let mut typesafe_overrides = ConfigOverrides { model: Some("gpt-5.2-codex".to_string()), ..Default::default() }; let persisted_metadata = test_thread_metadata(Some("gpt-5.1-codex-max"), Some(ReasoningEffort::High))?; merge_persisted_resume_metadata( &mut request_overrides, &mut typesafe_overrides, &persisted_metadata, ); assert_eq!(typesafe_overrides.model, Some("gpt-5.2-codex".to_string())); assert_eq!(typesafe_overrides.model_provider, None); assert_eq!( request_overrides, Some(HashMap::from([( "model_reasoning_effort".to_string(), serde_json::Value::String("low".to_string()), )])) ); Ok(()) } #[test] fn merge_persisted_resume_metadata_skips_persisted_values_when_model_overridden() -> Result<()> { let mut request_overrides = Some(HashMap::from([( "model".to_string(), serde_json::Value::String("gpt-5.2-codex".to_string()), )])); let mut typesafe_overrides = ConfigOverrides::default(); let persisted_metadata = test_thread_metadata(Some("gpt-5.1-codex-max"), Some(ReasoningEffort::High))?; merge_persisted_resume_metadata( &mut request_overrides, &mut typesafe_overrides, &persisted_metadata, ); assert_eq!(typesafe_overrides.model, None); assert_eq!(typesafe_overrides.model_provider, None); assert_eq!( request_overrides, Some(HashMap::from([( "model".to_string(), serde_json::Value::String("gpt-5.2-codex".to_string()), )])) ); Ok(()) } #[test] fn merge_persisted_resume_metadata_skips_persisted_values_when_provider_overridden() -> Result<()> { let mut request_overrides = None; let mut typesafe_overrides = ConfigOverrides { model_provider: Some("oss".to_string()), ..Default::default() }; let persisted_metadata = test_thread_metadata(Some("gpt-5.1-codex-max"), Some(ReasoningEffort::High))?; merge_persisted_resume_metadata( &mut request_overrides, &mut typesafe_overrides, &persisted_metadata, ); assert_eq!(typesafe_overrides.model, None); assert_eq!(typesafe_overrides.model_provider, Some("oss".to_string())); assert_eq!(request_overrides, None); Ok(()) } #[test] fn merge_persisted_resume_metadata_skips_persisted_values_when_reasoning_effort_overridden() -> Result<()> { let mut request_overrides = Some(HashMap::from([( "model_reasoning_effort".to_string(), serde_json::Value::String("low".to_string()), )])); let mut typesafe_overrides = ConfigOverrides::default(); let persisted_metadata = test_thread_metadata(Some("gpt-5.1-codex-max"), Some(ReasoningEffort::High))?; merge_persisted_resume_metadata( &mut request_overrides, &mut typesafe_overrides, &persisted_metadata, ); assert_eq!(typesafe_overrides.model, None); assert_eq!(typesafe_overrides.model_provider, None); assert_eq!( request_overrides, Some(HashMap::from([( "model_reasoning_effort".to_string(), serde_json::Value::String("low".to_string()), )])) ); Ok(()) } #[test] fn merge_persisted_resume_metadata_skips_missing_values() -> Result<()> { let mut request_overrides = None; let mut typesafe_overrides = ConfigOverrides::default(); let persisted_metadata = test_thread_metadata(/*model*/ None, /*reasoning_effort*/ None)?; merge_persisted_resume_metadata( &mut request_overrides, &mut typesafe_overrides, &persisted_metadata, ); assert_eq!(typesafe_overrides.model, None); assert_eq!( typesafe_overrides.model_provider, Some("mock_provider".to_string()) ); assert_eq!(request_overrides, None); Ok(()) } #[tokio::test] async fn read_summary_from_rollout_returns_empty_preview_when_no_user_message() -> Result<()> { use codex_protocol::protocol::RolloutItem; use codex_protocol::protocol::RolloutLine; use codex_protocol::protocol::SessionMetaLine; use std::fs; use std::fs::FileTimes; let temp_dir = TempDir::new()?; let path = temp_dir.path().join("rollout.jsonl"); let conversation_id = ThreadId::from_string("bfd12a78-5900-467b-9bc5-d3d35df08191")?; let timestamp = "2025-09-05T16:53:11.850Z".to_string(); let session_meta = SessionMeta { id: conversation_id, timestamp: timestamp.clone(), model_provider: None, ..SessionMeta::default() }; let line = RolloutLine { timestamp: timestamp.clone(), item: RolloutItem::SessionMeta(SessionMetaLine { meta: session_meta.clone(), git: None, }), }; fs::write(&path, format!("{}\n", serde_json::to_string(&line)?))?; let parsed = chrono::DateTime::parse_from_rfc3339(×tamp)?.with_timezone(&Utc); let times = FileTimes::new().set_modified(parsed.into()); std::fs::OpenOptions::new() .append(true) .open(&path)? .set_times(times)?; let summary = read_summary_from_rollout(path.as_path(), "fallback").await?; let expected = ConversationSummary { conversation_id, timestamp: Some(timestamp.clone()), updated_at: Some(timestamp), path: path.clone(), preview: String::new(), model_provider: "fallback".to_string(), cwd: PathBuf::new(), cli_version: String::new(), source: SessionSource::VSCode, git_info: None, }; assert_eq!(summary, expected); Ok(()) } #[tokio::test] async fn read_summary_from_rollout_preserves_agent_nickname() -> Result<()> { use codex_protocol::protocol::RolloutItem; use codex_protocol::protocol::RolloutLine; use codex_protocol::protocol::SessionMetaLine; use std::fs; let temp_dir = TempDir::new()?; let path = temp_dir.path().join("rollout.jsonl"); let conversation_id = ThreadId::from_string("bfd12a78-5900-467b-9bc5-d3d35df08191")?; let parent_thread_id = ThreadId::from_string("ad7f0408-99b8-4f6e-a46f-bd0eec433370")?; let timestamp = "2025-09-05T16:53:11.850Z".to_string(); let session_meta = SessionMeta { id: conversation_id, timestamp: timestamp.clone(), source: SessionSource::SubAgent(SubAgentSource::ThreadSpawn { parent_thread_id, depth: 1, agent_path: None, agent_nickname: None, agent_role: None, }), thread_source: Some(codex_protocol::protocol::ThreadSource::Subagent), agent_nickname: Some("atlas".to_string()), agent_role: Some("explorer".to_string()), model_provider: Some("test-provider".to_string()), ..SessionMeta::default() }; let line = RolloutLine { timestamp, item: RolloutItem::SessionMeta(SessionMetaLine { meta: session_meta, git: None, }), }; fs::write(&path, format!("{}\n", serde_json::to_string(&line)?))?; let summary = read_summary_from_rollout(path.as_path(), "fallback").await?; let fallback_cwd = AbsolutePathBuf::from_absolute_path("/")?; let thread = summary_to_thread(summary, &fallback_cwd); assert_eq!(thread.agent_nickname, Some("atlas".to_string())); assert_eq!(thread.agent_role, Some("explorer".to_string())); assert_eq!(thread.thread_source, None); Ok(()) } #[tokio::test] async fn read_summary_from_rollout_preserves_forked_from_id() -> Result<()> { use codex_protocol::protocol::RolloutItem; use codex_protocol::protocol::RolloutLine; use codex_protocol::protocol::SessionMetaLine; use std::fs; let temp_dir = TempDir::new()?; let path = temp_dir.path().join("rollout.jsonl"); let conversation_id = ThreadId::from_string("bfd12a78-5900-467b-9bc5-d3d35df08191")?; let forked_from_id = ThreadId::from_string("ad7f0408-99b8-4f6e-a46f-bd0eec433370")?; let timestamp = "2025-09-05T16:53:11.850Z".to_string(); let session_meta = SessionMeta { id: conversation_id, forked_from_id: Some(forked_from_id), timestamp: timestamp.clone(), model_provider: Some("test-provider".to_string()), ..SessionMeta::default() }; let line = RolloutLine { timestamp, item: RolloutItem::SessionMeta(SessionMetaLine { meta: session_meta, git: None, }), }; fs::write(&path, format!("{}\n", serde_json::to_string(&line)?))?; assert_eq!( forked_from_id_from_rollout(path.as_path()).await, Some(forked_from_id.to_string()) ); Ok(()) } #[tokio::test] async fn aborting_pending_request_clears_pending_state() -> Result<()> { let thread_id = ThreadId::from_string("bfd12a78-5900-467b-9bc5-d3d35df08191")?; let connection_id = ConnectionId(7); let (outgoing_tx, mut outgoing_rx) = tokio::sync::mpsc::channel(8); let outgoing = Arc::new(OutgoingMessageSender::new( outgoing_tx, codex_analytics::AnalyticsEventsClient::disabled(), )); let thread_outgoing = ThreadScopedOutgoingMessageSender::new( outgoing.clone(), vec![connection_id], thread_id, ); let (request_id, client_request_rx) = thread_outgoing .send_request(ServerRequestPayload::ToolRequestUserInput( ToolRequestUserInputParams { thread_id: thread_id.to_string(), turn_id: "turn-1".to_string(), item_id: "call-1".to_string(), questions: vec![], auto_resolution_ms: None, }, )) .await; thread_outgoing.abort_pending_server_requests().await; let request_message = outgoing_rx.recv().await.expect("request should be sent"); let OutgoingEnvelope::ToConnection { connection_id: request_connection_id, message: OutgoingMessage::Request(ServerRequest::ToolRequestUserInput { request_id: sent_request_id, .. }), .. } = request_message else { panic!("expected tool request to be sent to the subscribed connection"); }; assert_eq!(request_connection_id, connection_id); assert_eq!(sent_request_id, request_id); let response = client_request_rx .await .expect("callback should be resolved"); let error = response.expect_err("request should be aborted during cleanup"); assert_eq!( error.message, "client request resolved because the turn state was changed" ); assert_eq!(error.data, Some(json!({ "reason": "turnTransition" }))); assert!( outgoing .pending_requests_for_thread(thread_id) .await .is_empty() ); assert!(outgoing_rx.try_recv().is_err()); Ok(()) } #[test] fn summary_from_state_db_metadata_preserves_agent_nickname() -> Result<()> { let conversation_id = ThreadId::from_string("bfd12a78-5900-467b-9bc5-d3d35df08191")?; let source = serde_json::to_string(&SessionSource::SubAgent(SubAgentSource::ThreadSpawn { parent_thread_id: ThreadId::from_string("ad7f0408-99b8-4f6e-a46f-bd0eec433370")?, depth: 1, agent_path: None, agent_nickname: None, agent_role: None, }))?; let summary = summary_from_state_db_metadata( conversation_id, PathBuf::from("/tmp/rollout.jsonl"), Some("hi".to_string()), /*preview*/ None, "2025-09-05T16:53:11Z".to_string(), "2025-09-05T16:53:12Z".to_string(), "test-provider".to_string(), PathBuf::from("/"), "0.0.0".to_string(), source, Some(codex_protocol::protocol::ThreadSource::Subagent), Some("atlas".to_string()), Some("explorer".to_string()), /*git_sha*/ None, /*git_branch*/ None, /*git_origin_url*/ None, ); let fallback_cwd = AbsolutePathBuf::from_absolute_path("/")?; let thread = summary_to_thread(summary, &fallback_cwd); assert_eq!(thread.agent_nickname, Some("atlas".to_string())); assert_eq!(thread.agent_role, Some("explorer".to_string())); Ok(()) } #[tokio::test] async fn removing_thread_state_clears_listener_and_active_turn_history() -> Result<()> { let manager = ThreadStateManager::new(); let thread_id = ThreadId::from_string("ad7f0408-99b8-4f6e-a46f-bd0eec433370")?; let connection = ConnectionId(1); let (cancel_tx, cancel_rx) = oneshot::channel(); manager .connection_initialized(connection, ConnectionCapabilities::default()) .await; manager .try_ensure_connection_subscribed( thread_id, connection, /*experimental_raw_events*/ false, ) .await .expect("connection should be live"); { let state = manager.thread_state(thread_id).await; let mut state = state.lock().await; state.cancel_tx = Some(cancel_tx); state.track_current_turn_event( "turn-1", &EventMsg::TurnStarted(codex_protocol::protocol::TurnStartedEvent { turn_id: "turn-1".to_string(), trace_id: None, started_at: None, model_context_window: None, collaboration_mode_kind: Default::default(), }), ); } manager.remove_thread_state(thread_id).await; assert_eq!(cancel_rx.await, Ok(())); let state = manager.thread_state(thread_id).await; let subscribed_connection_ids = manager.subscribed_connection_ids(thread_id).await; assert!(subscribed_connection_ids.is_empty()); let state = state.lock().await; assert!(state.cancel_tx.is_none()); assert!(state.active_turn_snapshot().is_none()); Ok(()) } #[tokio::test] async fn removing_auto_attached_connection_preserves_listener_for_other_connections() -> Result<()> { let manager = ThreadStateManager::new(); let thread_id = ThreadId::from_string("ad7f0408-99b8-4f6e-a46f-bd0eec433370")?; let connection_a = ConnectionId(1); let connection_b = ConnectionId(2); let (cancel_tx, mut cancel_rx) = oneshot::channel(); manager .connection_initialized(connection_a, ConnectionCapabilities::default()) .await; manager .connection_initialized(connection_b, ConnectionCapabilities::default()) .await; manager .try_ensure_connection_subscribed( thread_id, connection_a, /*experimental_raw_events*/ false, ) .await .expect("connection_a should be live"); manager .try_ensure_connection_subscribed( thread_id, connection_b, /*experimental_raw_events*/ false, ) .await .expect("connection_b should be live"); { let state = manager.thread_state(thread_id).await; state.lock().await.cancel_tx = Some(cancel_tx); } let threads_to_unload = manager.remove_connection(connection_a).await; assert_eq!(threads_to_unload, Vec::<ThreadId>::new()); assert!( tokio::time::timeout(Duration::from_millis(20), &mut cancel_rx) .await .is_err() ); assert_eq!( manager.subscribed_connection_ids(thread_id).await, vec![connection_b] ); Ok(()) } #[tokio::test] async fn adding_connection_to_thread_updates_has_connections_watcher() -> Result<()> { let manager = ThreadStateManager::new(); let thread_id = ThreadId::from_string("ad7f0408-99b8-4f6e-a46f-bd0eec433370")?; let connection_a = ConnectionId(1); let connection_b = ConnectionId(2); manager .connection_initialized(connection_a, ConnectionCapabilities::default()) .await; manager .connection_initialized(connection_b, ConnectionCapabilities::default()) .await; manager .try_ensure_connection_subscribed( thread_id, connection_a, /*experimental_raw_events*/ false, ) .await .expect("connection_a should be live"); let mut has_connections = manager .subscribe_to_has_connections(thread_id) .await .expect("thread should have a has-connections watcher"); assert!(*has_connections.borrow()); assert!( manager .unsubscribe_connection_from_thread(thread_id, connection_a) .await ); tokio::time::timeout(Duration::from_secs(1), has_connections.changed()) .await .expect("timed out waiting for no-subscriber update") .expect("has-connections watcher should remain open"); assert!(!*has_connections.borrow()); assert!( manager .try_add_connection_to_thread(thread_id, connection_b) .await ); tokio::time::timeout(Duration::from_secs(1), has_connections.changed()) .await .expect("timed out waiting for subscriber update") .expect("has-connections watcher should remain open"); assert!(*has_connections.borrow()); Ok(()) } #[tokio::test] async fn closed_connection_cannot_be_reintroduced_by_auto_subscribe() -> Result<()> { let manager = ThreadStateManager::new(); let thread_id = ThreadId::from_string("ad7f0408-99b8-4f6e-a46f-bd0eec433370")?; let connection = ConnectionId(1); manager .connection_initialized(connection, ConnectionCapabilities::default()) .await; let threads_to_unload = manager.remove_connection(connection).await; assert_eq!(threads_to_unload, Vec::<ThreadId>::new()); assert!( manager .try_ensure_connection_subscribed( thread_id, connection, /*experimental_raw_events*/ false ) .await .is_none() ); assert!(!manager.has_subscribers(thread_id).await); Ok(()) } #[tokio::test] async fn first_attestation_capable_connection_for_thread_only_uses_thread_subscribers() -> Result<()> { let manager = ThreadStateManager::new(); let thread_id = ThreadId::from_string("dfbd9a95-2f44-470a-8bd8-1cfc04efc243")?; let other_thread_id = ThreadId::from_string("6c9a74e4-5e59-479e-90bf-5c5798bb50aa")?; let unrelated_supported_connection = ConnectionId(1); let earlier_supported_connection = ConnectionId(2); let later_supported_connection = ConnectionId(3); let unsupported_connection = ConnectionId(4); manager .connection_initialized( unrelated_supported_connection, ConnectionCapabilities { request_attestation: true, }, ) .await; manager .connection_initialized( earlier_supported_connection, ConnectionCapabilities { request_attestation: true, }, ) .await; manager .connection_initialized( later_supported_connection, ConnectionCapabilities { request_attestation: true, }, ) .await; manager .connection_initialized(unsupported_connection, ConnectionCapabilities::default()) .await; assert!( manager .try_add_connection_to_thread(other_thread_id, unrelated_supported_connection) .await ); assert!( manager .try_add_connection_to_thread(thread_id, later_supported_connection) .await ); assert!( manager .try_add_connection_to_thread(thread_id, earlier_supported_connection) .await ); assert!( manager .try_add_connection_to_thread(thread_id, unsupported_connection) .await ); assert_eq!( manager .first_attestation_capable_connection_for_thread(thread_id) .await, Some(earlier_supported_connection) ); assert_eq!( manager .first_attestation_capable_connection_for_thread(other_thread_id) .await, Some(unrelated_supported_connection) ); Ok(()) }}