From b112d79196830ff0114476fd488a6357e509ca0e Mon Sep 17 00:00:00 2001 From: Ghibli1024 Date: Thu, 13 Aug 2026 00:33:29 +0800 Subject: [PATCH] fix(runtime): stabilize CDP injection and bridge lifecycle --- apps/codex-plus-launcher/src/main.rs | 15 +- crates/codex-plus-core/src/bridge.rs | 266 ++++++++++++++---- crates/codex-plus-core/src/cdp.rs | 45 ++- crates/codex-plus-core/src/launcher.rs | 69 ++++- crates/codex-plus-core/tests/cdp_bridge.rs | 304 ++++++++++++++++++++- crates/codex-plus-core/tests/launcher.rs | 37 ++- 6 files changed, 650 insertions(+), 86 deletions(-) diff --git a/apps/codex-plus-launcher/src/main.rs b/apps/codex-plus-launcher/src/main.rs index a6947b83e..093e902b6 100644 --- a/apps/codex-plus-launcher/src/main.rs +++ b/apps/codex-plus-launcher/src/main.rs @@ -188,16 +188,13 @@ async fn activate_existing_codex_app(options: &LaunchOptions) -> anyhow::Result< hooks.start_helper(helper_port).await?; } let process_ids = codex_plus_core::watcher::find_codex_processes(); - let mut activated = false; #[cfg(windows)] - { - for process_id in &process_ids { - if codex_plus_core::windows_activate_process_window(*process_id) { - activated = true; - break; - } - } - } + let activated = process_ids + .iter() + .copied() + .any(codex_plus_core::windows_activate_process_window); + #[cfg(not(windows))] + let activated = false; let injection_ready = if settings.enhancements_enabled { hooks .ensure_injection(options.debug_port, helper_port, &app_dir) diff --git a/crates/codex-plus-core/src/bridge.rs b/crates/codex-plus-core/src/bridge.rs index 29991523c..6a1417e20 100644 --- a/crates/codex-plus-core/src/bridge.rs +++ b/crates/codex-plus-core/src/bridge.rs @@ -9,6 +9,7 @@ use std::time::Duration; use anyhow::{Context, bail}; use base64::Engine; +use futures_util::stream::FuturesUnordered; use futures_util::{SinkExt, StreamExt}; use serde_json::{Value, json}; use tokio_tungstenite::connect_async; @@ -26,6 +27,69 @@ pub type BridgeHandler = Arc< static NEXT_MESSAGE_ID: AtomicU64 = AtomicU64::new(100); +/// Bridge 会话按注入目标分代。 +/// +/// 同一目标再次安装 Bridge 时，旧会话会在下一次消息循环中退出并关闭 socket， +/// 避免多份 CDP 会话同时应答同一个页面请求。不同目标互不影响。 +static NEXT_BRIDGE_GENERATION: AtomicU64 = AtomicU64::new(1); +static CURRENT_BRIDGE_GENERATIONS: std::sync::LazyLock>> = + std::sync::LazyLock::new(|| std::sync::Mutex::new(HashMap::new())); + +#[derive(Clone)] +struct BridgeGeneration { + target: String, + id: u64, +} + +type PendingBridgeCall = Pin + Send>>; + +struct CompletedBridgeCall { + request_id: String, + generation: Option, + response: Result, +} + +fn next_bridge_generation(target: &str) -> BridgeGeneration { + BridgeGeneration { + target: target.to_string(), + id: NEXT_BRIDGE_GENERATION.fetch_add(1, Ordering::SeqCst), + } +} + +fn publish_bridge_generation(generation: &BridgeGeneration) -> bool { + let mut generations = CURRENT_BRIDGE_GENERATIONS + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + if generations + .get(&generation.target) + .is_some_and(|current| *current > generation.id) + { + return false; + } + generations.insert(generation.target.clone(), generation.id); + true +} + +fn bridge_generation_is_current(generation: &BridgeGeneration) -> bool { + CURRENT_BRIDGE_GENERATIONS + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .get(&generation.target) + .is_some_and(|current| *current == generation.id) +} + +fn release_bridge_generation(generation: &BridgeGeneration) { + let mut generations = CURRENT_BRIDGE_GENERATIONS + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + if generations + .get(&generation.target) + .is_some_and(|current| *current == generation.id) + { + generations.remove(&generation.target); + } +} + pub fn build_bridge_script(binding_name: &str) -> String { format!( r#" @@ -192,6 +256,8 @@ pub async fn install_bridge( ) -> anyhow::Result<()> { let socket = connect_cdp_websocket(websocket_url).await?; let mut session = CdpSession::new(socket).with_handler(handler); + let generation = next_bridge_generation(websocket_url); + session = session.with_generation(generation.clone()); session.send_command(1, "Runtime.enable", json!({})).await?; session @@ -236,17 +302,51 @@ pub async fn install_bridge( .await?; } - session.drain_binding_queue().await?; + if !publish_bridge_generation(&generation) { + let _ = crate::diagnostic_log::append_diagnostic_log( + "bridge.generation_superseded_before_publish", + json!({ "generation": generation.id }), + ); + session.close().await; + return Ok(()); + } + let _ = crate::diagnostic_log::append_diagnostic_log( + "bridge.generation_published", + json!({ "generation": generation.id }), + ); + + let mut pending_calls = FuturesUnordered::new(); + session.enqueue_binding_calls(&mut pending_calls); tokio::spawn(async move { loop { - if session.drain_binding_queue().await.is_err() { + if !bridge_generation_is_current(&generation) { + let _ = crate::diagnostic_log::append_diagnostic_log( + "bridge.generation_superseded", + json!({ "generation": generation.id }), + ); break; } - match session.next_message().await { - Ok(Some(_)) => {} - Ok(None) | Err(_) => break, + + session.enqueue_binding_calls(&mut pending_calls); + tokio::select! { + completed = pending_calls.next(), if !pending_calls.is_empty() => { + let Some(completed) = completed else { + continue; + }; + if session.finish_binding_call(completed).await.is_err() { + break; + } + } + message = session.next_message() => { + match message { + Ok(Some(_)) => {} + Ok(None) | Err(_) => break, + } + } } } + session.close().await; + release_bridge_generation(&generation); }); Ok(()) @@ -308,6 +408,7 @@ struct CdpSession { responses: HashMap, binding_calls: VecDeque, handler: Option, + generation: Option, } impl CdpSession @@ -324,6 +425,7 @@ where responses: HashMap::new(), binding_calls: VecDeque::new(), handler: None, + generation: None, } } @@ -332,6 +434,22 @@ where self } + fn with_generation(mut self, generation: BridgeGeneration) -> Self { + self.generation = Some(generation); + self + } + + fn is_current(&self) -> bool { + self.generation + .as_ref() + .is_none_or(bridge_generation_is_current) + } + + async fn close(&mut self) { + let _ = self.socket.send(Message::Close(None)).await; + let _ = self.socket.close().await; + } + async fn send_command( &mut self, message_id: u64, @@ -421,73 +539,117 @@ where Ok(Some(value)) } - async fn drain_binding_queue(&mut self) -> anyhow::Result<()> { + fn enqueue_binding_calls(&mut self, pending_calls: &mut FuturesUnordered) { while let Some(message) = self.binding_calls.pop_front() { - self.route_binding_call(message).await?; + self.enqueue_binding_call(message, pending_calls); } - Ok(()) } - fn route_binding_call( + fn enqueue_binding_call( &mut self, message: Value, - ) -> Pin> + Send + '_>> { - Box::pin(async move { - let Some(handler) = self.handler.clone() else { - return Ok(()); - }; - - let Some(payload_text) = message - .get("params") - .and_then(|params| params.get("payload")) - .and_then(Value::as_str) - else { - return Ok(()); - }; + pending_calls: &mut FuturesUnordered, + ) { + let Some(handler) = self.handler.clone() else { + return; + }; - let parsed: Value = match serde_json::from_str(payload_text) { - Ok(parsed) => parsed, - Err(error) => { - if let Some(request_id) = extract_string_field(payload_text, "id") { - self.reject_bridge_request( - &request_id, - &format!("failed to parse bridge payload: {error}"), - ) - .await?; - } - return Ok(()); - } - }; - self.route_parsed_binding_call(&handler, parsed).await - }) - } + let Some(payload_text) = message + .get("params") + .and_then(|params| params.get("payload")) + .and_then(Value::as_str) + else { + return; + }; - async fn route_parsed_binding_call( - &mut self, - handler: &BridgeHandler, - parsed: Value, - ) -> anyhow::Result<()> { - let Some(request_id) = parsed.get("id").and_then(Value::as_str) else { - return Ok(()); + let parsed: Value = match serde_json::from_str(payload_text) { + Ok(parsed) => parsed, + Err(error) => { + let Some(request_id) = extract_string_field(payload_text, "id") else { + return; + }; + self.enqueue_completed_binding_call( + request_id, + Err(format!("failed to parse bridge payload: {error}")), + pending_calls, + ); + return; + } + }; + let Some(request_id) = parsed.get("id").and_then(Value::as_str).map(str::to_string) else { + return; }; + if !self.is_current() { + let _ = crate::diagnostic_log::append_diagnostic_log( + "bridge.stale_request_dropped", + json!({ + "request_id": request_id, + "generation": self.generation.as_ref().map(|generation| generation.id) + }), + ); + return; + } let path = parsed .get("path") .and_then(Value::as_str) .unwrap_or_default() .to_string(); let payload = parsed.get("payload").cloned().unwrap_or_else(|| json!({})); + let generation = self.generation.clone(); + + pending_calls.push(Box::pin(async move { + CompletedBridgeCall { + request_id, + generation, + response: handler(path, payload) + .await + .map_err(|error| error.to_string()), + } + })); + } - match handler(path, payload).await { + fn enqueue_completed_binding_call( + &self, + request_id: String, + response: Result, + pending_calls: &mut FuturesUnordered, + ) { + let generation = self.generation.clone(); + pending_calls.push(Box::pin(async move { + CompletedBridgeCall { + request_id, + generation, + response, + } + })); + } + + async fn finish_binding_call(&mut self, completed: CompletedBridgeCall) -> anyhow::Result<()> { + if completed + .generation + .as_ref() + .is_some_and(|generation| !bridge_generation_is_current(generation)) + { + let _ = crate::diagnostic_log::append_diagnostic_log( + "bridge.stale_response_dropped", + json!({ + "request_id": completed.request_id, + "generation": completed.generation.as_ref().map(|generation| generation.id) + }), + ); + return Ok(()); + } + + match completed.response { Ok(result) => { - self.resolve_bridge_request(request_id, &result).await?; + self.resolve_bridge_request(&completed.request_id, &result) + .await } - Err(error) => { - self.reject_bridge_request(request_id, &error.to_string()) - .await?; + Err(message) => { + self.reject_bridge_request(&completed.request_id, &message) + .await } } - - Ok(()) } async fn resolve_bridge_request( diff --git a/crates/codex-plus-core/src/cdp.rs b/crates/codex-plus-core/src/cdp.rs index 9a813e979..047d7d942 100644 --- a/crates/codex-plus-core/src/cdp.rs +++ b/crates/codex-plus-core/src/cdp.rs @@ -253,16 +253,19 @@ pub fn pick_page_target(targets: &[CdpTarget]) -> anyhow::Result { } pub fn pick_injectable_codex_page_target(targets: &[CdpTarget]) -> anyhow::Result { - // Only inject into Codex's own app:// page (or the supported ChatGPT - // desktop page). Embedded browser pages can have titles or URLs containing - // "Codex" (for example a GitHub PR), but they must never become the target. - if let Some(target) = targets.iter().find(|target| { - is_injectable_page_target(target) - && is_primary_codex_page_target(target) - && (is_codex_app_page_target(target) - || is_chatgpt_desktop_page(&target.title, &target.url)) - }) { - return Ok(target.clone()); + let priorities: [fn(&CdpTarget) -> bool; 4] = [ + is_exact_codex_app_main_target, + is_primary_codex_app_target, + is_chatgpt_desktop_page_target, + is_supported_codex_page_target, + ]; + for matches_priority in priorities { + if let Some(target) = targets + .iter() + .find(|target| is_injectable_page_target(target) && matches_priority(target)) + { + return Ok(target.clone()); + } } bail!("No injectable Codex page target found") } @@ -298,6 +301,23 @@ pub fn is_primary_codex_page_target(target: &CdpTarget) -> bool { && !is_quick_chat_page_target(target) } +fn is_exact_codex_app_main_target(target: &CdpTarget) -> bool { + target.url.trim().eq_ignore_ascii_case("app://-/index.html") +} + +fn is_primary_codex_app_target(target: &CdpTarget) -> bool { + is_codex_app_page_target(target) && is_primary_codex_page_target(target) +} + +fn is_chatgpt_desktop_page_target(target: &CdpTarget) -> bool { + is_primary_codex_page_target(target) && is_chatgpt_desktop_page(&target.title, &target.url) +} + +fn is_supported_codex_page_target(target: &CdpTarget) -> bool { + is_primary_codex_page_target(target) + && (is_codex_app_page_target(target) || is_chatgpt_desktop_page(&target.title, &target.url)) +} + pub fn is_avatar_overlay_page_target(target: &CdpTarget) -> bool { initial_route(target).is_some_and(|route| route.eq_ignore_ascii_case("/avatar-overlay")) } @@ -316,10 +336,7 @@ fn initial_route(target: &CdpTarget) -> Option { return None; } let url = reqwest::Url::parse(target.url.trim()).ok()?; - if !url.scheme().eq_ignore_ascii_case("app") - || url.host_str() != Some("-") - || !url.path().eq_ignore_ascii_case("/index.html") - { + if !is_codex_app_page_target(target) { return None; } url.query_pairs() diff --git a/crates/codex-plus-core/src/launcher.rs b/crates/codex-plus-core/src/launcher.rs index 102404fc0..0493e3d8a 100644 --- a/crates/codex-plus-core/src/launcher.rs +++ b/crates/codex-plus-core/src/launcher.rs @@ -25,6 +25,8 @@ const POST_LAUNCH_COMPUTER_USE_GUARD_SECONDS: &[u64] = &[0, 5, 15, 30, 60, 120, const POST_LAUNCH_COMPUTER_USE_GUARD_STABLE_ATTEMPTS: usize = 3; static PET_OVERLAY_SYNC_FAILED: AtomicBool = AtomicBool::new(false); static PET_CURSOR_DRIVER_FAILED: AtomicBool = AtomicBool::new(false); +const MACOS_DEBUG_TAKEOVER_WAIT_MS: u64 = 5_000; +const MACOS_DEBUG_TAKEOVER_INTERVAL_MS: u64 = 100; /// Asynchronous callback used by the bridge watchdog to restore a launcher-specific bridge. /// @@ -59,6 +61,13 @@ pub enum MacosCleanupPolicy { SkipQuitBecauseAlreadyRunning, } +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum MacosDebugLaunchAction { + LaunchNew, + ReuseRunningDebugApp, + RestartRunningApp, +} + #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum WindowsProcessControlStrategy { NativeWindowsApi, @@ -766,10 +775,26 @@ impl LaunchHooks for DefaultLaunchHooks { } if app_dir.extension().and_then(|value| value.to_str()) == Some("app") { - let cleanup_policy = if is_macos_app_running(app_dir).await { - MacosCleanupPolicy::SkipQuitBecauseAlreadyRunning - } else { - MacosCleanupPolicy::QuitIfNotPreviouslyRunning + let launch_action = select_macos_debug_launch_action( + is_macos_app_running(app_dir).await, + crate::cdp::endpoint_available(debug_port), + ); + let cleanup_policy = match launch_action { + MacosDebugLaunchAction::LaunchNew => MacosCleanupPolicy::QuitIfNotPreviouslyRunning, + MacosDebugLaunchAction::ReuseRunningDebugApp => { + MacosCleanupPolicy::SkipQuitBecauseAlreadyRunning + } + MacosDebugLaunchAction::RestartRunningApp => { + let _ = crate::diagnostic_log::append_diagnostic_log( + "launcher.macos_existing_app_without_cdp_restart_requested", + serde_json::json!({ + "app_dir": app_dir, + "debug_port": debug_port + }), + ); + quit_macos_app_and_wait(app_dir).await?; + MacosCleanupPolicy::QuitIfNotPreviouslyRunning + } }; let command = if let Some(inspector_port) = native_menu_inspector_port { build_macos_open_command_with_native_menu_inspector( @@ -2780,6 +2805,17 @@ pub fn build_macos_cleanup_command( ]) } +pub fn select_macos_debug_launch_action( + app_running: bool, + codex_cdp_available: bool, +) -> MacosDebugLaunchAction { + match (app_running, codex_cdp_available) { + (false, _) => MacosDebugLaunchAction::LaunchNew, + (true, true) => MacosDebugLaunchAction::ReuseRunningDebugApp, + (true, false) => MacosDebugLaunchAction::RestartRunningApp, + } +} + async fn run_macos_cleanup_command( app_dir: &Path, policy: MacosCleanupPolicy, @@ -2800,6 +2836,31 @@ async fn run_macos_cleanup_command( Ok(()) } +async fn quit_macos_app_and_wait(app_dir: &Path) -> anyhow::Result<()> { + run_macos_cleanup_command(app_dir, MacosCleanupPolicy::QuitIfNotPreviouslyRunning).await?; + let deadline = tokio::time::Instant::now() + + std::time::Duration::from_millis(MACOS_DEBUG_TAKEOVER_WAIT_MS); + while is_macos_app_running(app_dir).await { + if tokio::time::Instant::now() >= deadline { + anyhow::bail!( + "macOS app did not exit before debug relaunch: {}", + app_dir.display() + ); + } + tokio::time::sleep(std::time::Duration::from_millis( + MACOS_DEBUG_TAKEOVER_INTERVAL_MS, + )) + .await; + } + let _ = crate::diagnostic_log::append_diagnostic_log( + "launcher.macos_existing_app_without_cdp_stopped", + serde_json::json!({ + "app_dir": app_dir + }), + ); + Ok(()) +} + fn macos_app_dir_from_open_command(command: &[String]) -> Option { let app_index = command.iter().position(|part| part == "-a")?; command.get(app_index + 1).map(PathBuf::from) diff --git a/crates/codex-plus-core/tests/cdp_bridge.rs b/crates/codex-plus-core/tests/cdp_bridge.rs index 03025f736..ce5500e40 100644 --- a/crates/codex-plus-core/tests/cdp_bridge.rs +++ b/crates/codex-plus-core/tests/cdp_bridge.rs @@ -18,7 +18,7 @@ use std::sync::Arc; use std::sync::atomic::{AtomicBool, Ordering}; use std::time::Duration; use tokio::net::TcpListener; -use tokio::sync::oneshot; +use tokio::sync::{Notify, oneshot}; use tokio_tungstenite::accept_async; use tokio_tungstenite::tungstenite::Message; @@ -2764,6 +2764,31 @@ fn pick_injectable_codex_page_target_accepts_chatgpt_desktop_error_page() { assert_eq!(picked.id, "chatgpt-error"); } +#[test] +fn pick_injectable_codex_page_target_prefers_app_main_over_incidental_codex_page() { + let targets = vec![ + target( + "help", + "page", + "Using Codex with your ChatGPT plan", + "https://help.openai.com/en/articles/using-codex", + Some("ws://help"), + ), + target( + "main", + "page", + "Codex", + "app://-/index.html", + Some("ws://main"), + ), + ]; + + let picked = pick_injectable_codex_page_target(&targets) + .expect("the exact app main renderer should win regardless of target order"); + + assert_eq!(picked.id, "main"); +} + #[test] fn avatar_overlay_target_detection_is_narrow() { let overlay = target( @@ -3438,8 +3463,285 @@ async fn install_bridge_does_not_wait_for_resolve_runtime_evaluate_ack() { .expect("server task should finish without panicking"); } +#[tokio::test] +async fn install_bridge_keeps_status_responsive_while_generate_is_pending() { + let generate_started = Arc::new(Notify::new()); + let release_generate = Arc::new(Notify::new()); + let server_generate_started = Arc::clone(&generate_started); + let server_release_generate = Arc::clone(&release_generate); + let (url, request_rx) = spawn_cdp_server(move |mut socket| async move { + acknowledge_bridge_install(&mut socket).await; + + send_json( + &mut socket, + json!({ + "method": "Runtime.bindingCalled", + "params": { + "payload": serde_json::to_string(&json!({ + "id": "generate", + "path": "/stepwise/generate", + "payload": {}, + })).unwrap(), + }, + }), + ) + .await; + tokio::time::timeout( + Duration::from_millis(500), + server_generate_started.notified(), + ) + .await + .expect("generate handler should start before the status probe"); + + send_json( + &mut socket, + json!({ + "method": "Runtime.bindingCalled", + "params": { + "payload": serde_json::to_string(&json!({ + "id": "status", + "path": "/backend/status", + "payload": {}, + })).unwrap(), + }, + }), + ) + .await; + + let status_resolve = + tokio::time::timeout(Duration::from_millis(500), recv_json(&mut socket)) + .await + .expect("status should resolve while generate remains pending"); + assert_eq!(status_resolve["method"], "Runtime.evaluate"); + assert_expression_contains_request(&status_resolve, "status"); + + server_release_generate.notify_one(); + let generate_resolve = recv_json(&mut socket).await; + assert_eq!(generate_resolve["method"], "Runtime.evaluate"); + assert_expression_contains_request(&generate_resolve, "generate"); + close_socket(&mut socket).await; + }) + .await; + + let handler_generate_started = Arc::clone(&generate_started); + let handler_release_generate = Arc::clone(&release_generate); + let handler = Arc::new(move |path: String, _payload: serde_json::Value| { + let generate_started = Arc::clone(&handler_generate_started); + let release_generate = Arc::clone(&handler_release_generate); + Box::pin(async move { + if path == "/stepwise/generate" { + generate_started.notify_one(); + release_generate.notified().await; + } + Ok(json!({ "status": "ok", "path": path })) + }) as Pin> + Send>> + }); + + tokio::time::timeout( + Duration::from_secs(2), + bridge::install_bridge(&url, BRIDGE_BINDING_NAME, handler, &[]), + ) + .await + .expect("bridge install should return while generate is pending") + .expect("bridge install should start the concurrent message pump"); + request_rx + .await + .expect("server task should finish without panicking"); +} + +#[tokio::test] +async fn superseded_bridge_session_stops_answering_binding_calls() { + let (url, stale_rx, fresh_rx) = spawn_two_session_cdp_server().await; + + bridge::install_bridge(&url, BRIDGE_BINDING_NAME, noop_handler(), &[]) + .await + .expect("first bridge install should succeed"); + bridge::install_bridge(&url, BRIDGE_BINDING_NAME, noop_handler(), &[]) + .await + .expect("second bridge install should succeed"); + + let stale_resolved = stale_rx + .await + .expect("stale server task should finish without panicking"); + assert!( + !stale_resolved, + "superseded session must not resolve bridge requests" + ); + fresh_rx + .await + .expect("fresh server task should finish without panicking"); +} + +#[tokio::test] +async fn failed_bridge_reinstall_keeps_existing_session_current() { + let (url, active_rx, failed_rx) = spawn_failed_reinstall_cdp_server().await; + + bridge::install_bridge(&url, BRIDGE_BINDING_NAME, noop_handler(), &[]) + .await + .expect("first bridge install should succeed"); + let error = bridge::install_bridge(&url, BRIDGE_BINDING_NAME, noop_handler(), &[]) + .await + .expect_err("second bridge install should fail"); + assert!(error.to_string().contains("Runtime.addBinding")); + + assert!( + active_rx + .await + .expect("active server task should finish without panicking"), + "failed reinstall must not supersede the existing bridge session" + ); + failed_rx + .await + .expect("failed reinstall server task should finish without panicking"); +} + type TestSocket = tokio_tungstenite::WebSocketStream; +async fn spawn_two_session_cdp_server() -> (String, oneshot::Receiver, oneshot::Receiver<()>) +{ + let listener = TcpListener::bind("127.0.0.1:0") + .await + .expect("test listener should bind"); + let address = listener.local_addr().expect("listener should have address"); + let (stale_tx, stale_rx) = oneshot::channel(); + let (fresh_tx, fresh_rx) = oneshot::channel(); + + tokio::spawn(async move { + let (stale_stream, _) = listener + .accept() + .await + .expect("stale client should connect"); + let mut stale = accept_async(stale_stream) + .await + .expect("stale websocket should upgrade"); + acknowledge_bridge_install(&mut stale).await; + + let (fresh_stream, _) = listener + .accept() + .await + .expect("fresh client should connect"); + let mut fresh = accept_async(fresh_stream) + .await + .expect("fresh websocket should upgrade"); + acknowledge_bridge_install(&mut fresh).await; + + send_json( + &mut stale, + json!({ + "method": "Runtime.bindingCalled", + "params": { + "payload": serde_json::to_string(&json!({ + "id": "stale", + "path": "/backend/status", + "payload": {}, + })).unwrap(), + }, + }), + ) + .await; + + let stale_resolved = + tokio::time::timeout(Duration::from_millis(500), recv_text_message(&mut stale)) + .await + .is_ok_and(|message| message.is_some_and(|text| text.contains("Runtime.evaluate"))); + let _ = stale_tx.send(stale_resolved); + close_socket(&mut fresh).await; + let _ = fresh_tx.send(()); + }); + + (websocket_url(address), stale_rx, fresh_rx) +} + +async fn spawn_failed_reinstall_cdp_server() +-> (String, oneshot::Receiver, oneshot::Receiver<()>) { + let listener = TcpListener::bind("127.0.0.1:0") + .await + .expect("test listener should bind"); + let address = listener.local_addr().expect("listener should have address"); + let (active_tx, active_rx) = oneshot::channel(); + let (failed_tx, failed_rx) = oneshot::channel(); + + tokio::spawn(async move { + let (active_stream, _) = listener + .accept() + .await + .expect("active client should connect"); + let mut active = accept_async(active_stream) + .await + .expect("active websocket should upgrade"); + acknowledge_bridge_install(&mut active).await; + + let (failed_stream, _) = listener + .accept() + .await + .expect("failed client should connect"); + let mut failed = accept_async(failed_stream) + .await + .expect("failed websocket should upgrade"); + for expected_id in 1..=2 { + let command = recv_json(&mut failed).await; + assert_eq!(command["id"], expected_id); + send_json(&mut failed, json!({ "id": expected_id, "result": {} })).await; + } + let add_binding = recv_json(&mut failed).await; + assert_eq!(add_binding["id"], 3); + assert_eq!(add_binding["method"], "Runtime.addBinding"); + send_json( + &mut failed, + json!({ + "id": 3, + "error": { "code": -32000, "message": "binding install failed" } + }), + ) + .await; + close_socket(&mut failed).await; + let _ = failed_tx.send(()); + + send_json( + &mut active, + json!({ + "method": "Runtime.bindingCalled", + "params": { + "payload": serde_json::to_string(&json!({ + "id": "active", + "path": "/backend/status", + "payload": {}, + })).unwrap(), + }, + }), + ) + .await; + let active_resolved = + tokio::time::timeout(Duration::from_millis(500), recv_json(&mut active)) + .await + .is_ok_and(|message| { + message["method"] == "Runtime.evaluate" + && message["params"]["expression"] + .as_str() + .is_some_and(|expression| expression.contains("active")) + }); + let _ = active_tx.send(active_resolved); + close_socket(&mut active).await; + }); + + (websocket_url(address), active_rx, failed_rx) +} + +async fn acknowledge_bridge_install(socket: &mut TestSocket) { + for expected_id in 1..=5 { + let command = recv_json(socket).await; + assert_eq!(command["id"], expected_id); + send_json(socket, json!({ "id": expected_id, "result": {} })).await; + } +} + +async fn recv_text_message(socket: &mut TestSocket) -> Option { + match socket.next().await { + Some(Ok(Message::Text(text))) => Some(text.to_string()), + _ => None, + } +} + async fn spawn_cdp_server(handler: F) -> (String, oneshot::Receiver<()>) where F: FnOnce(TestSocket) -> Fut + Send + 'static, diff --git a/crates/codex-plus-core/tests/launcher.rs b/crates/codex-plus-core/tests/launcher.rs index 2995c55a7..b8a9b375d 100644 --- a/crates/codex-plus-core/tests/launcher.rs +++ b/crates/codex-plus-core/tests/launcher.rs @@ -8,12 +8,13 @@ use codex_plus_core::app_paths::{ }; use codex_plus_core::launcher::{ CodexLaunch, DefaultLaunchHooks, LaunchHooks, LaunchOptions, MacosCleanupPolicy, - browser_identity_changed, build_codex_arguments, build_codex_arguments_for_settings, - build_codex_arguments_with_native_menu_inspector, build_codex_command, - build_codex_command_with_native_menu_inspector, build_macos_cleanup_command, - build_macos_open_command, build_macos_open_command_with_native_menu_inspector, - build_packaged_activation, build_packaged_activation_with_native_menu_inspector, - launch_and_inject_with_hooks, + MacosDebugLaunchAction, browser_identity_changed, build_codex_arguments, + build_codex_arguments_for_settings, build_codex_arguments_with_native_menu_inspector, + build_codex_command, build_codex_command_with_native_menu_inspector, + build_macos_cleanup_command, build_macos_open_command, + build_macos_open_command_with_native_menu_inspector, build_packaged_activation, + build_packaged_activation_with_native_menu_inspector, launch_and_inject_with_hooks, + select_macos_debug_launch_action, }; #[cfg(windows)] use codex_plus_core::launcher::{WindowsProcessControlStrategy, windows_process_control_strategy}; @@ -1745,6 +1746,30 @@ fn launcher_macos_cleanup_is_skipped_when_app_was_already_running() { assert_eq!(command, None); } +#[test] +fn launcher_macos_debug_launch_starts_when_app_is_not_running() { + assert_eq!( + select_macos_debug_launch_action(false, false), + MacosDebugLaunchAction::LaunchNew + ); +} + +#[test] +fn launcher_macos_debug_launch_reuses_existing_codex_cdp_instance() { + assert_eq!( + select_macos_debug_launch_action(true, true), + MacosDebugLaunchAction::ReuseRunningDebugApp + ); +} + +#[test] +fn launcher_macos_debug_launch_restarts_existing_non_cdp_instance() { + assert_eq!( + select_macos_debug_launch_action(true, false), + MacosDebugLaunchAction::RestartRunningApp + ); +} + #[tokio::test] async fn default_launch_hooks_provider_sync_enabled_returns_explicit_error() { let error = DefaultLaunchHooks::default()