From 1558fffe95cbf098f77744d3dd55264ae8784e74 Mon Sep 17 00:00:00 2001 From: lennney <94768569+lennney@users.noreply.github.com> Date: Thu, 2 Jul 2026 13:46:27 +0800 Subject: [PATCH] perf(launcher): event-driven CDP readiness + eliminate nested retry MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Event-driven stderr pipe (Playwright pattern) replaces 120×1s polling. Uses three-phase fallback: Phase 1: oneshot signal from stderr 'DevTools listening on ws://' Phase 2: exponential backoff TCP probe (100ms→10s) Phase 3: 1s-interval polling as ultimate safety net (30 attempts) Also replaces self.inject() → try_inject() in all three phases to eliminate the hidden 20×500ms retry_injection() loop nested inside every backoff step. Before: Phase 2 worst-case ~101s, Phase 3 ~330s, total ~446s After: Phase 2 worst-case ~21s, Phase 3 ~60s, total ~66s --- crates/codex-plus-core/Cargo.toml | 2 +- crates/codex-plus-core/src/launcher.rs | 106 ++++++++++++++++++++++- crates/codex-plus-core/tests/launcher.rs | 33 ++++++- 3 files changed, 136 insertions(+), 5 deletions(-) diff --git a/crates/codex-plus-core/Cargo.toml b/crates/codex-plus-core/Cargo.toml index 2e5609e2..129c6935 100644 --- a/crates/codex-plus-core/Cargo.toml +++ b/crates/codex-plus-core/Cargo.toml @@ -19,7 +19,7 @@ serde.workspace = true serde_json.workspace = true sha2.workspace = true thiserror.workspace = true -tokio = { workspace = true, features = ["net"] } +tokio = { workspace = true, features = ["net", "io-util"] } tokio-tungstenite.workspace = true toml.workspace = true toml_edit.workspace = true diff --git a/crates/codex-plus-core/src/launcher.rs b/crates/codex-plus-core/src/launcher.rs index 75a851c2..614d6285 100644 --- a/crates/codex-plus-core/src/launcher.rs +++ b/crates/codex-plus-core/src/launcher.rs @@ -8,7 +8,7 @@ use anyhow::Context; use async_trait::async_trait; use futures_util::StreamExt; use serde_json::Value; -use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt}; use tokio::process::{Child, Command}; use tokio::sync::Mutex; @@ -216,6 +216,9 @@ pub struct DefaultLaunchHooks { bridge_watchdog: Mutex>, computer_use_guard_watchdog: Mutex>, computer_use_guard_artifacts: Mutex>, + /// Oneshot receiver that fires when Codex's CDP debug port is ready. + /// Set up by `launch_codex()` (stderr listener), consumed by `ensure_injection()`. + cdp_ready: Mutex>>, } struct HelperRuntime { @@ -727,12 +730,32 @@ impl LaunchHooks for DefaultLaunchHooks { child_command .args(&command[1..]) .stdout(Stdio::null()) - .stderr(Stdio::null()); + .stderr(Stdio::piped()); #[cfg(windows)] child_command.creation_flags(crate::windows_integration::CREATE_NO_WINDOW); - let child = child_command + let mut child = child_command .spawn() .with_context(|| format!("failed to launch Codex executable {executable}"))?; + + // Read stderr to detect when the CDP port is ready. + // Chrome/Electron prints "DevTools listening on ws://..." to stderr. + // This is the same pattern used by Playwright (92k★) for Electron apps. + if let Some(stderr) = child.stderr.take() { + let (tx, rx) = tokio::sync::oneshot::channel(); + *self.cdp_ready.lock().await = Some(rx); + tokio::spawn(async move { + if let Err(error) = wait_for_cdp_ready(stderr).await { + let _ = crate::diagnostic_log::append_diagnostic_log( + "launcher.cdp_stderr_listener_error", + serde_json::json!({"message": error.to_string()}), + ); + } + // Drop sender: if ensure_injection is still waiting, it gets + // Canceled and falls through to the TCP-backoff fallback. + drop(tx); + }); + } + *self.child.lock().await = Some(child); if let Some(inspector_port) = native_menu_inspector_port { start_native_menu_localizer(inspector_port); @@ -869,6 +892,52 @@ impl LaunchHooks for DefaultLaunchHooks { } } + async fn ensure_injection(&self, debug_port: u16, helper_port: u16, _app_dir: &Path) -> bool { + // Phase 1: Event-driven — wait for CDP port readiness signal from stderr. + let cdp_ready = { self.cdp_ready.lock().await.take() }; + + if let Some(rx) = cdp_ready { + // Wait for the stderr signal with a moderate timeout. + // On a healthy system the line arrives within 2-5 seconds. + let signal = tokio::time::timeout(std::time::Duration::from_secs(15), rx).await; + match signal { + Ok(Ok(())) => { + // CDP ready signal received! Try injection once; it + // should succeed if the port is genuinely ready. + if try_inject(debug_port, helper_port).await.is_ok() { + return true; + } + } + _ => { + // Timeout or sender dropped — fall through to backoff. + } + } + } + + // Phase 2: Bounded backoff — TCP connect probing with exponential delay. + // This handles edge cases where the stderr line is not printed + // (Windows packaged app, macOS `.app` bundle, etc.). + let backoff_delays = [100, 200, 400, 800, 1600, 3200, 5000, 10000u64]; + for delay_ms in &backoff_delays { + if try_inject(debug_port, helper_port).await.is_ok() { + return true; + } + tokio::time::sleep(std::time::Duration::from_millis(*delay_ms)).await; + } + + // Phase 3: Original 1s-interval polling as ultimate safety net. + // This matches the original behaviour for systems where injection + // genuinely takes >30 seconds (very slow disks, heavy load). + for _attempt in 1..=30u32 { + if try_inject(debug_port, helper_port).await.is_ok() { + return true; + } + tokio::time::sleep(std::time::Duration::from_secs(1)).await; + } + + false + } + async fn terminate_codex(&self, launch: &CodexLaunch) { match launch { CodexLaunch::Process { @@ -1692,6 +1761,37 @@ pub fn build_packaged_activation_with_native_menu_inspector( }) } +/// Detect when Codex's CDP endpoint is ready by reading its stderr. +/// +/// Chrome/Electron prints "DevTools listening on ws://127.0.0.1:/..." +/// to stderr when the CDP server is ready. This is the same pattern used +/// by Playwright (92k★) for Electron apps. +/// +/// Returns `Ok(())` as soon as the magic line is found, or `Ok(())` on EOF +/// (the caller falls through to TCP-backoff). Errors only on actual I/O +/// failures, which are also non-fatal. +pub async fn wait_for_cdp_ready(mut stderr: impl tokio::io::AsyncRead + Unpin) -> anyhow::Result<()> { + let mut reader = tokio::io::BufReader::new(&mut stderr); + let mut line = String::new(); + let magic = "DevTools listening on ws://"; + + loop { + line.clear(); + let n = reader.read_line(&mut line).await?; + if n == 0 { + // EOF — process closed stderr without printing the magic line + break; + } + if line.contains(magic) { + return Ok(()); + } + } + + // Fallback: even without the stderr line, the port may still be ready. + // Return success so the caller falls through to the TCP-backoff. + Ok(()) +} + async fn retry_injection(debug_port: u16, helper_port: u16) -> anyhow::Result<()> { let mut last_error = None; for _ in 0..20 { diff --git a/crates/codex-plus-core/tests/launcher.rs b/crates/codex-plus-core/tests/launcher.rs index c6a0a4a0..196ae9d7 100644 --- a/crates/codex-plus-core/tests/launcher.rs +++ b/crates/codex-plus-core/tests/launcher.rs @@ -14,7 +14,7 @@ use codex_plus_core::launcher::{ 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, + launch_and_inject_with_hooks, wait_for_cdp_ready, }; #[cfg(windows)] use codex_plus_core::launcher::{WindowsProcessControlStrategy, windows_process_control_strategy}; @@ -22,6 +22,7 @@ use codex_plus_core::ports::{ select_packaged_codex_debug_port_with, select_platform_loopback_port_with, }; use codex_plus_core::settings::{BackendSettings, RelayProfile, RelayProtocol}; +use tokio::io::AsyncWriteExt; use codex_plus_core::status::StatusStore; #[test] @@ -1490,3 +1491,33 @@ impl LaunchHooks for FakeHooks { } } } + +#[tokio::test] +async fn test_wait_for_cdp_ready_detects_magic_line() { + let (mut writer, reader) = tokio::io::duplex(1024); + writer.write_all(b"some garbage\n").await.unwrap(); + writer.write_all(b"DevTools listening on ws://127.0.0.1:9222\n").await.unwrap(); + drop(writer); + let result = wait_for_cdp_ready(reader).await; + assert!(result.is_ok()); +} + +#[tokio::test] +async fn test_wait_for_cdp_ready_eof_is_ok() { + let (writer, reader) = tokio::io::duplex(1024); + drop(writer); + let result = wait_for_cdp_ready(reader).await; + assert!(result.is_ok()); +} + +#[tokio::test] +async fn test_wait_for_cdp_ready_ignores_noise_before_magic() { + let (mut writer, reader) = tokio::io::duplex(1024); + writer.write_all(b"[INFO] Starting process...\n").await.unwrap(); + writer.write_all(b"[INFO] Loading configuration...\n").await.unwrap(); + writer.write_all(b"[WARN] Something deprecated\n").await.unwrap(); + writer.write_all(b"DevTools listening on ws://127.0.0.1:9222\n").await.unwrap(); + drop(writer); + let result = wait_for_cdp_ready(reader).await; + assert!(result.is_ok()); +}