diff --git a/apps/codex-plus-launcher/src/main.rs b/apps/codex-plus-launcher/src/main.rs index 8d6eb6afa..55a0a353a 100644 --- a/apps/codex-plus-launcher/src/main.rs +++ b/apps/codex-plus-launcher/src/main.rs @@ -350,6 +350,10 @@ pub struct LaunchReceipt { /// 与 start 相同，但只从已就绪的本进程会话返回实际端口，不读取旧状态文件猜测。 pub fn start_with_receipt(options: LaunchOptions) -> Result { start(options)?; + ready_session_receipt() +} + +fn ready_session_receipt() -> Result { let probe = session_probe_slot() .lock() .unwrap_or_else(|error| error.into_inner()); @@ -372,6 +376,14 @@ pub fn resume_if_running(options: LaunchOptions) -> Result { Ok(true) } +/// 仅 attach 已有桌面，并返回本进程已就绪会话的实际端口。 +pub fn resume_with_receipt(options: LaunchOptions) -> Result> { + if !resume_if_running(options)? { + return Ok(None); + } + ready_session_receipt().map(Some) +} + /// 释放当前 session 的 helper、watchdog 和浏览器兼容状态，保留官方 Codex。 pub fn stop() -> Result<()> { send_runtime_command(RuntimeCommand::Stop) @@ -761,10 +773,38 @@ async fn activate_and_write_running_status( options: &LaunchOptions, started_at_ms: u64, ) -> Result<()> { + let latest = options.status_store.load_latest()?; activate_existing_codex_app(options).await?; - options.status_store.save_latest(&LaunchStatus { - status: "running".to_string(), - message: "Existing Codex instance activated".to_string(), + options.status_store.save_latest(&activated_launch_status( + options, + started_at_ms, + latest.as_ref(), + )) +} + +fn activated_launch_status( + options: &LaunchOptions, + started_at_ms: u64, + latest: Option<&LaunchStatus>, +) -> LaunchStatus { + // 激活窗口没有重新验证增强，不能把同代会话的降级状态升级为成功。 + let degraded = latest.is_some_and(|latest| { + latest.started_at_ms == started_at_ms + && latest.debug_port == Some(options.debug_port) + && latest.status == "running_degraded" + }); + LaunchStatus { + status: if degraded { + "running_degraded" + } else { + "running" + } + .to_string(), + message: if degraded { + latest.unwrap().message.clone() + } else { + "Existing Codex instance activated".to_string() + }, started_at_ms, debug_port: Some(options.debug_port), helper_port: Some(options.helper_port), @@ -774,7 +814,7 @@ async fn activate_and_write_running_status( .map(|path| path.to_string_lossy().to_string()), aumid: None, ..LaunchStatus::default() - }) + } } async fn wait_for_session_exit_or_stop( @@ -988,7 +1028,13 @@ async fn activate_existing_codex_app(options: &LaunchOptions) -> anyhow::Result< json!({"message": error.to_string(), "phase": "activate_existing"}), ); } - let process_ids = codex_plus_core::watcher::find_codex_processes(); + // 仅激活本次启动返回的目标，不按全局同名进程选择其他窗口。 + let process_ids: Vec = launch_result + .as_ref() + .ok() + .and_then(|launch| launch.process_id()) + .into_iter() + .collect(); #[cfg(windows)] let activated = process_ids .iter() @@ -1303,6 +1349,9 @@ impl LaunchHooks for LauncherHooks { &self, settings: &codex_plus_core::settings::BackendSettings, ) -> anyhow::Result<()> { + if self.attach_only { + return Ok(()); + } self.core .ensure_active_protocol_proxy_config(settings) .await @@ -2137,11 +2186,10 @@ mod tests { + start; let body = &source[start..end]; let multi_guard = body.find("if multi_instance_mode()").unwrap(); - let activate = body.find("activate_and_write_running_status(&options").unwrap(); - assert!( - multi_guard < activate, - "多开判断必须排在激活已有实例之前" - ); + let activate = body + .find("activate_and_write_running_status(&options") + .unwrap(); + assert!(multi_guard < activate, "多开判断必须排在激活已有实例之前"); assert!(body[multi_guard..activate].contains("anyhow::bail!")); } @@ -2775,6 +2823,25 @@ mod tests { hooks.apply_native_fast_unlock_on_launch(0).await.unwrap(); } + #[test] + fn activating_same_generation_keeps_degraded_status() { + let options = LaunchOptions::default(); + let previous = LaunchStatus { + status: "running_degraded".into(), + message: "injection unavailable".into(), + started_at_ms: 7, + debug_port: Some(options.debug_port), + ..LaunchStatus::default() + }; + let activated = activated_launch_status(&options, 7, Some(&previous)); + assert_eq!(activated.status, "running_degraded"); + assert_eq!(activated.message, previous.message); + assert_eq!( + activated_launch_status(&options, 8, Some(&previous)).status, + "running" + ); + } + #[tokio::test] async fn watchdog_reuses_bridge_context_with_data_service() { let test_dir = std::env::temp_dir().join(format!( diff --git a/apps/codex-plus-manager/src-tauri/src/commands.rs b/apps/codex-plus-manager/src-tauri/src/commands.rs index 9d04c5632..afb61568d 100644 --- a/apps/codex-plus-manager/src-tauri/src/commands.rs +++ b/apps/codex-plus-manager/src-tauri/src/commands.rs @@ -20,6 +20,8 @@ use tauri::Emitter; use crate::install::{self, InstallActionResult}; +pub(crate) mod multi_instance_resume; + // 扫描快照仅驻留后端内存，不接受前端指定磁盘路径，不开放给扩展路由。 static AGENT_CACHE_PLAN: OnceLock>> = OnceLock::new(); @@ -678,6 +680,10 @@ pub struct CodexMultiWorkerOptions { /// 显式路径为准（同时仍设置 `CODEX_PLUS_STATE_DIR` 覆盖子进程）。 #[serde(default)] pub state_dir: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub resume_identity: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub resume_token: Option, } /// 一个已经由「Codex 多开」创建过的实例。记录文件放在实例自己的隔离目录， @@ -749,6 +755,8 @@ pub struct CodexMultiInstancesPayload { #[serde(rename_all = "camelCase")] pub struct CodexMultiInstanceRequest { pub instance_id: String, + #[serde(default)] + pub generation: Option, } #[derive(Debug, Clone, serde::Deserialize)] @@ -1001,6 +1009,12 @@ pub async fn load_overview() -> CommandResult { #[tauri::command] pub async fn launch_codex_plus(request: LaunchRequest) -> CommandResult { match tauri::async_runtime::spawn_blocking(move || { + if let Err(error) = prepare_main_launch_options(Default::default()) { + return failed( + &format!("显式启动目录无效，未执行启动：{error:#}"), + json!({}), + ); + } let _guard = match try_acquire_launch_guard() { Ok(guard) => guard, Err(message) => return failed(message, json!({})), @@ -1099,6 +1113,17 @@ fn launch_codex_multi_instance_blocking(request: CodexMultiLaunchRequest) -> Com json!({"instanceId": instance_id}), ); } + if previous.as_ref().is_some_and(|instance| { + instance.identity.as_ref().is_some_and(|identity| { + multi_manifest_identity_matches(instance, identity) + && !codex_plus_core::multi_worker::codex_instance_process_exited(identity) + }) && !multi_instance_is_running(instance) + }) { + return failed( + "原 Codex 进程尚未确认退出，但连接暂不可用。请先结束或重启该实例，未创建重复窗口。", + json!({"instanceId": instance_id}), + ); + } if previous.as_ref().is_some_and(|previous| { !multi_instance_is_running(previous) && previous.pid.is_some_and(|pid| { @@ -1331,6 +1356,8 @@ fn launch_codex_multi_instance_blocking(request: CodexMultiLaunchRequest) -> Com debug_port, helper_port, state_dir: worker_state_dir.to_string_lossy().to_string(), + resume_identity: None, + resume_token: None, }; let worker_options_bytes = match serde_json::to_vec_pretty(&worker_options) { Ok(bytes) => bytes, @@ -1483,10 +1510,9 @@ fn list_codex_multi_instances_blocking() -> CommandResult CommandResult CommandResult CommandResult { match tauri::async_runtime::spawn_blocking(move || { - stop_codex_multi_instance_blocking(&request.instance_id) + stop_codex_multi_instance_blocking(&request.instance_id, request.generation) }) .await { @@ -1571,6 +1602,7 @@ pub async fn stop_codex_multi_instance( fn stop_codex_multi_instance_blocking( instance_id: &str, + expected_generation: Option, ) -> CommandResult { let operation = multi_instance_mutation_lock() .lock() @@ -1600,6 +1632,14 @@ fn stop_codex_multi_instance_blocking( }, ); } + if expected_generation.is_some_and(|generation| generation != instance.generation) { + return failed( + "实例已重新启动，请刷新列表后操作；未结束新一代进程。", + CodexMultiInstancesPayload { + instances: Vec::new(), + }, + ); + } let Some(identity) = instance .identity .as_ref() @@ -1612,24 +1652,12 @@ fn stop_codex_multi_instance_blocking( }, ); }; - if !multi_instance_is_running(&instance) - && !matches!( - codex_plus_core::watcher::inspect_process_instance(identity.process_id), - codex_plus_core::watcher::ProcessInstanceState::NotRunning - ) - { - return failed( - "无法重新确认该实例的完整身份，已拒绝结束进程。", - CodexMultiInstancesPayload { - instances: Vec::new(), - }, - ); - } - if multi_instance_is_running(&instance) { + let mut cleaned_by_manager = false; + if !codex_plus_core::multi_worker::codex_instance_process_exited(identity) { match tauri::async_runtime::block_on( codex_plus_core::multi_worker::terminate_verified_codex_instance(identity), ) { - Ok(true) => {} + Ok(true) => cleaned_by_manager = true, Ok(false) => { return failed( "实例身份在停止前已改变，未结束任何进程。", @@ -1640,7 +1668,7 @@ fn stop_codex_multi_instance_blocking( } Err(error) => { return failed( - &format!("安全停止实例失败：{error}"), + &format!("安全停止实例失败：{error:#}"), CodexMultiInstancesPayload { instances: Vec::new(), }, @@ -1648,10 +1676,12 @@ fn stop_codex_multi_instance_blocking( } } let deadline = std::time::Instant::now() + std::time::Duration::from_secs(10); - while multi_instance_is_running(&instance) && std::time::Instant::now() < deadline { + while !codex_plus_core::multi_worker::codex_instance_process_exited(identity) + && std::time::Instant::now() < deadline + { std::thread::sleep(std::time::Duration::from_millis(100)); } - if multi_instance_is_running(&instance) { + if !codex_plus_core::multi_worker::codex_instance_process_exited(identity) { return failed( "实例仍未退出，请稍后重试。", CodexMultiInstancesPayload { @@ -1660,10 +1690,63 @@ fn stop_codex_multi_instance_blocking( ); } } - // 保留该代结果作为身份记录，下次启动用新 generation 排除旧结果。 - if let Some(pid) = instance.pid { - let options_path = path.join("worker-options.json"); - let _ = terminate_multi_worker(pid, &options_path, instance.worker_birth_id); + // 根退出后 worker 仍可能持有残留子进程；先等待该代清理，不能抢先结束它。 + let cleanup_deadline = std::time::Instant::now() + std::time::Duration::from_secs(12); + loop { + let confirmed = load_multi_launch_result(&path).is_some_and(|result| { + result.instance_id == instance.instance_id + && result.generation == instance.generation + && result.identity.as_ref() == Some(identity) + && result.status == "stopped" + }); + if confirmed || std::time::Instant::now() >= cleanup_deadline { + if !confirmed && !cleaned_by_manager { + return failed( + "Codex 已退出，但该代后台尚未确认子进程清理完成；请稍后重试。", + CodexMultiInstancesPayload { + instances: Vec::new(), + }, + ); + } + break; + } + std::thread::sleep(std::time::Duration::from_millis(100)); + } + if let Some(pid) = instance.pid + && let Err(error) = terminate_multi_worker( + pid, + &path.join("worker-options.json"), + instance.worker_birth_id, + ) + { + return failed( + &format!("Codex 已退出，但后台尚未确认停止：{error}"), + CodexMultiInstancesPayload { + instances: Vec::new(), + }, + ); + } + if cleaned_by_manager { + let result = CodexMultiLaunchResult { + instance_id: instance.instance_id.clone(), + generation: instance.generation, + helper_port: Some(identity.helper_port), + identity: Some(identity.clone()), + status: "stopped".into(), + message: "Codex 及其自有后台已停止。".into(), + finished_at_ms: current_timestamp_ms(), + debug_port: Some(identity.debug_port), + pid: Some(identity.process_id), + }; + if let Err(error) = write_multi_launch_result(&path.join(MULTI_LAUNCH_RESULT_FILE), &result) + { + return failed( + &format!("实例已停止，但记录清理结果失败：{error}"), + CodexMultiInstancesPayload { + instances: Vec::new(), + }, + ); + } } let _ = codex_plus_core::diagnostic_log::append_diagnostic_log( "manager.multi_instance_stopped", @@ -1678,7 +1761,7 @@ pub async fn restart_codex_multi_instance( request: CodexMultiInstanceRequest, ) -> CommandResult { match tauri::async_runtime::spawn_blocking(move || { - restart_codex_multi_instance_blocking(&request.instance_id) + restart_codex_multi_instance_blocking(&request.instance_id, request.generation) }) .await { @@ -1687,7 +1770,10 @@ pub async fn restart_codex_multi_instance( } } -fn restart_codex_multi_instance_blocking(instance_id: &str) -> CommandResult { +fn restart_codex_multi_instance_blocking( + instance_id: &str, + expected_generation: Option, +) -> CommandResult { if instance_id.trim() == "main" { return failed( "主进程请使用顶部的重启 Codex。", @@ -1701,7 +1787,13 @@ fn restart_codex_multi_instance_blocking(instance_id: &str) -> CommandResult bool { .unwrap_or(false) } +fn multi_instance_snapshot_status(instance: &CodexMultiInstance, running: bool) -> &'static str { + if running { + return multi_instance_status(true, instance.launch_result.as_ref()); + } + if instance + .launch_result + .as_ref() + .is_some_and(|result| result.status == "cleanup_failed") + { + return "unknown"; + } + if let Some(identity) = instance.identity.as_ref() { + if !multi_manifest_identity_matches(instance, identity) + || !codex_plus_core::multi_worker::codex_instance_process_exited(identity) + { + return "unknown"; + } + if !instance.launch_result.as_ref().is_some_and(|result| { + result.instance_id == instance.instance_id + && result.generation == instance.generation + && result.identity.as_ref() == Some(identity) + && result.status == "stopped" + }) { + return "unknown"; + } + } else if instance.pid.is_some_and(|pid| { + !matches!( + codex_plus_core::watcher::inspect_process_instance(pid), + codex_plus_core::watcher::ProcessInstanceState::NotRunning + ) + }) { + return if instance.launch_result.is_none() { + "starting" + } else { + "unknown" + }; + } + multi_instance_status(false, instance.launch_result.as_ref()) +} + /// 列表里显示的状态：能跑就是运行中；跑不起来且 worker 明确报了失败才是失败， /// 否则只是没在运行（例如用户自己关掉了窗口）。 fn multi_instance_status(running: bool, result: Option<&CodexMultiLaunchResult>) -> &'static str { @@ -1823,7 +1955,7 @@ fn terminate_multi_worker( worker_birth_id: Option, ) -> anyhow::Result<()> { if pid == std::process::id() { - return Ok(()); + anyhow::bail!("拒绝结束管理器自身"); } #[cfg(not(windows))] { @@ -1840,9 +1972,21 @@ fn terminate_multi_worker( worker_birth_id, )?; if !terminated { - let _ = codex_plus_core::diagnostic_log::append_diagnostic_log( - "manager.multi_worker_stop_skipped", - json!({ "pid": pid, "reason": "identity_unverified" }), + let original_exited = match codex_plus_core::watcher::inspect_process_instance(pid) { + codex_plus_core::watcher::ProcessInstanceState::NotRunning => true, + codex_plus_core::watcher::ProcessInstanceState::Running { + birth_id: Some(actual), + .. + } => actual + .parse::() + .ok() + .zip(worker_birth_id) + .is_some_and(|(actual, expected)| actual != expected), + _ => false, + }; + anyhow::ensure!( + original_exited, + "无法核验原代 worker 已退出，未结束未验证的进程" ); } } @@ -1863,6 +2007,9 @@ fn terminate_multi_worker( /// 让界面显示「启动失败」而不是「运行中」。 pub fn run_codex_multi_worker(options_path: &Path) -> anyhow::Result<()> { let options: CodexMultiWorkerOptions = serde_json::from_slice(&std::fs::read(options_path)?)?; + // 更新续接必须在任何 launcher 副作用之前核验原桌面；失败绝不退回 start。 + multi_instance_resume::validate_worker_resume(&options)?; + let worker_binary = multi_instance_resume::BinaryStamp::current()?; // 状态目录以 worker 自己的 options 为准，并且**不用** `set_var` 改进程环境： // 多开 worker 复用管理器进程，启动时已是多线程，edition 2024 下 `set_var` 属未定义 @@ -1892,7 +2039,14 @@ pub fn run_codex_multi_worker(options_path: &Path) -> anyhow::Result<()> { codex_home: std::env::var_os("CODEX_HOME").map(PathBuf::from), sqlite_home: std::env::var_os("CODEX_SQLITE_HOME").map(PathBuf::from), }; - let receipt = match codex_plus_launcher::start_with_receipt(launch_options) { + let started = if options.resume_identity.is_some() { + codex_plus_launcher::resume_with_receipt(launch_options).and_then(|receipt| { + receipt.ok_or_else(|| anyhow::anyhow!("原 Codex 已不可续接，未启动新窗口")) + }) + } else { + codex_plus_launcher::start_with_receipt(launch_options) + }; + let receipt = match started { Ok(receipt) => receipt, Err(error) => { let _ = write_multi_launch_result( @@ -1912,9 +2066,12 @@ pub fn run_codex_multi_worker(options_path: &Path) -> anyhow::Result<()> { return Ok(()); } }; - let endpoint_ready = - wait_for_multi_instance_ready(receipt.debug_port, MULTI_LAUNCH_READY_TIMEOUT); - let captured = if endpoint_ready { + let endpoint_ready = options.resume_identity.is_some() + || wait_for_multi_instance_ready(receipt.debug_port, MULTI_LAUNCH_READY_TIMEOUT); + let captured = if let Some(identity) = &options.resume_identity { + multi_instance_resume::confirm_resumed_identity(identity, &receipt) + .map(|()| identity.clone()) + } else if endpoint_ready { tauri::async_runtime::block_on( codex_plus_core::multi_worker::capture_codex_instance_identity( &options.instance_id, @@ -1963,14 +2120,81 @@ pub fn run_codex_multi_worker(options_path: &Path) -> anyhow::Result<()> { let _ = codex_plus_launcher::shutdown(); return Ok(()); }; + let process_checkpoint = options_path.with_file_name("worker-process-tree.json"); + let mut process_tree = + codex_plus_core::multi_worker::CodexInstanceProcessTree::capture(&identity).ok(); + if options.resume_identity.is_some() && process_checkpoint.exists() { + if let Some(tree) = process_tree.as_mut() { + if let Err(error) = tree.restore_handoff_checkpoint(&process_checkpoint) { + let _ = codex_plus_launcher::shutdown(); + return Err(error.context("恢复原代进程树归属失败，未发布 worker 就绪")); + } + } + } + let handoff = match multi_instance_resume::publish_worker_ready( + options_path, + &options, + &identity, + worker_binary, + ) { + Ok(ready) => ready, + Err(error) => { + let shutdown = codex_plus_launcher::shutdown(); + shutdown?; + return Err(error); + } + }; while tauri::async_runtime::block_on( - codex_plus_core::multi_worker::codex_instance_identity_matches(&identity), + codex_plus_core::multi_worker::codex_instance_needs_worker(&identity), ) - .unwrap_or(false) + .unwrap_or(true) { + if multi_instance_resume::handoff_requested(options_path, &handoff, || { + process_tree.as_mut() + .ok_or_else(|| anyhow::anyhow!("未捕获完整进程树，拒绝丢弃原代归属")) + .and_then(|tree| tree.save_handoff_checkpoint(&process_checkpoint)) + }) { + // 新 worker 已接管服务和进程树归属，旧 worker 才能退出。 + return codex_plus_launcher::shutdown(); + } + if let Some(tree) = process_tree.as_mut() + && let Err(error) = tree.refresh() + { + let _ = codex_plus_core::diagnostic_log::append_diagnostic_log( + "manager.multi_instance_tree_refresh_failed", + json!({ "instance_id": options.instance_id, "error": error.to_string() }), + ); + } std::thread::sleep(std::time::Duration::from_secs(1)); } - codex_plus_launcher::shutdown() + let cleanup = process_tree + .ok_or_else(|| anyhow::anyhow!("未捕获完整的自有桌面进程，无法确认清理完成")) + .and_then(|tree| tree.cleanup_after_exit()); + let shutdown = codex_plus_launcher::shutdown(); + let cleanup = cleanup.and(shutdown); + write_multi_launch_result( + &result_path, + &CodexMultiLaunchResult { + instance_id: options.instance_id, + generation: options.generation, + helper_port: Some(receipt.helper_port), + identity: Some(identity.clone()), + status: if cleanup.is_ok() { + "stopped" + } else { + "cleanup_failed" + } + .into(), + message: cleanup.as_ref().err().map_or_else( + || "Codex 多开实例已退出。".to_string(), + |error| format!("Codex 已退出，但清理未完成：{error:#}"), + ), + finished_at_ms: current_timestamp_ms(), + debug_port: Some(receipt.debug_port), + pid: Some(identity.process_id), + }, + )?; + cleanup } /// 多开实例启动结果文件名（与 `worker-options.json` 同目录）。 @@ -2230,8 +2454,7 @@ pub fn run_auto_takeover_launch(request: LaunchRequest) -> Result<(), String> { ..Default::default() }; // 自动接管也走常驻 helper：Codex 之后退出时本地代理仍可被外部客户端使用。 - sync_resident_helper(); - codex_plus_launcher::start(launch_options).map_err(|error| error.to_string()) + start_main_launcher(launch_options).map_err(|error| error.to_string()) } #[tauri::command] @@ -2259,34 +2482,392 @@ pub fn pause_auto_takeover_for_instance() -> CommandResult, + identity: codex_plus_core::multi_worker::CodexInstanceIdentity, +} + +fn main_identity_path() -> PathBuf { + codex_plus_core::paths::default_app_state_dir().join("main-instance-identity.json") } -fn restart_disposition(sync_active_relay: bool, app_running: bool) -> RestartDisposition { - if !sync_active_relay && !app_running { - RestartDisposition::LaunchOnly - } else { - RestartDisposition::StopAndRestart +fn main_identity_in_scope( + receipt: &MainCodexIdentity, + state_dir: &Path, + codex_home: &Path, +) -> bool { + receipt.version == 1 + && receipt.identity.instance_id == "main" + && receipt.identity.generation != 0 + && receipt.identity.process_id != 0 + && receipt.identity.process_id != std::process::id() + && receipt.identity.process_birth_id != 0 + && receipt.identity.debug_port != 0 + && receipt.identity.helper_port != 0 + && !receipt.identity.browser_id.is_empty() + && receipt.identity.executable.is_absolute() + && (receipt.identity.user_data_dir.is_absolute() + || receipt.identity.user_data_dir.as_os_str().is_empty()) + && state_dir.canonicalize().ok().as_ref() == Some(&receipt.state_dir) + && codex_home.canonicalize().ok().as_ref() == Some(&receipt.codex_home) +} + +fn load_main_identity() -> anyhow::Result> { + let bytes = match fs::read(main_identity_path()) { + Ok(bytes) => bytes, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None), + Err(error) => return Err(error.into()), + }; + decode_main_identity( + &bytes, + &codex_plus_core::paths::default_app_state_dir(), + &codex_plus_core::relay_config::default_codex_home_dir(), + ) +} + +fn decode_main_identity( + bytes: &[u8], + state_dir: &Path, + codex_home: &Path, +) -> anyhow::Result> { + let Some(receipt) = serde_json::from_slice::>(bytes)? else { + return Ok(None); + }; + anyhow::ensure!( + main_identity_in_scope(&receipt, state_dir, codex_home), + "主实例身份记录不属于当前管理器目录" + ); + Ok(Some(receipt)) +} + +/// 窗口连通性与进程存活分开判断；绝不借任意同名进程补足主实例身份。 +pub(crate) fn main_codex_running(latest: Option<&LaunchStatus>) -> Option { + let session = codex_plus_launcher::session_snapshot(); + let receipt = load_main_identity().ok()?; + main_running_with( + &session, + receipt.as_ref().map(|receipt| &receipt.identity), + |identity| { + if codex_plus_core::multi_worker::codex_instance_process_exited(identity) { + return Some(false); + } + tauri::async_runtime::block_on( + codex_plus_core::multi_worker::codex_instance_needs_worker(identity), + ) + .ok() + }, + || { + let debug_port = latest + .and_then(|status| status.debug_port) + .unwrap_or(default_debug_port()); + main_target_absent(debug_port).ok() + }, + ) +} + +fn main_running_with( + session: &codex_plus_launcher::SessionSnapshot, + identity: Option<&codex_plus_core::multi_worker::CodexInstanceIdentity>, + process_running: impl FnOnce(&codex_plus_core::multi_worker::CodexInstanceIdentity) -> Option, + target_absent: impl FnOnce() -> Option, +) -> Option { + match identity { + Some(identity) => { + if session.owned && session.started_at_ms != Some(identity.generation) { + return None; + } + process_running(identity) + } + None => { + if session.owned { + return None; + } + target_absent().and_then(|absent| absent.then_some(false)) + } + } +} + +fn verified_main_neighbors() -> Vec { + let root = codex_plus_core::paths::default_app_state_dir().join("codex-multi"); + let Ok(entries) = fs::read_dir(root) else { + return Vec::new(); + }; + entries + .flatten() + .filter_map(|entry| { + if !entry.file_type().ok()?.is_dir() { + return None; + } + // 准入探测只读，不调用可能刷新清单的 load_multi_instance_manifest。 + let mut instance: CodexMultiInstance = serde_json::from_slice( + &fs::read(multi_instance_manifest_path(&entry.path())).ok()?, + ) + .ok()?; + if instance.is_main + || entry.file_name() + != std::ffi::OsStr::new(&multi_instance_slug(&instance.instance_id)) + { + return None; + } + if let Some(result) = load_multi_launch_result(&entry.path()).filter(|result| { + result.instance_id == instance.instance_id + && result.generation == instance.generation + }) { + if let Some(identity) = result.identity { + instance.debug_port = result.debug_port.unwrap_or(instance.debug_port); + instance.helper_port = result.helper_port.unwrap_or(instance.helper_port); + instance.identity = Some(identity); + } + } + let identity = instance.identity.as_ref()?; + if identity.instance_id == "main" + || !multi_manifest_identity_matches(&instance, identity) + { + return None; + } + tauri::async_runtime::block_on( + codex_plus_core::multi_worker::codex_instance_needs_worker(identity), + ) + .ok() + .filter(|running| *running) + .map(|_| identity.clone()) + }) + .collect() +} + +fn main_command_is_electron_child(arguments: &[String]) -> bool { + let mut arguments = arguments.iter().skip(1); + while let Some(argument) = arguments.next() { + let role = if argument == "--type" { + arguments.next().map(String::as_str) + } else { + argument.strip_prefix("--type=") + }; + if role.is_some_and(|role| { + matches!( + role, + "renderer" | "gpu-process" | "utility" | "zygote" | "crashpad-handler" + ) + }) { + return true; + } + } + false +} + +fn main_command_is_bundled_app_server(executable: &Path, arguments: &[String]) -> bool { + let bundled = executable + .parent() + .and_then(Path::file_name) + .is_some_and(|parent| parent.eq_ignore_ascii_case("resources")); + if !bundled { + return false; + } + let mut arguments = arguments.iter().skip(1); + while let Some(argument) = arguments.next() { + // CLI 全局参数可位于子命令之前；未知参数保持保守。 + match argument.as_str() { + "-c" | "--config" | "--enable" | "--disable" | "-p" | "--profile" | "-C" | "--cd" + | "-m" | "--model" => { + if arguments.next().is_none() { + return false; + } + } + "app-server" => return true, + value + if [ + "--config=", + "--enable=", + "--disable=", + "--profile=", + "--cd=", + "--model=", + ] + .iter() + .any(|prefix| value.starts_with(prefix)) => {} + _ => return false, + } + } + false +} + +#[cfg(windows)] +fn main_candidate_is_verified_neighbor( + process: &codex_plus_core::WindowsProcessInfo, + neighbors: &[codex_plus_core::multi_worker::CodexInstanceIdentity], + birth: Option, +) -> bool { + neighbors.iter().any(|identity| { + identity.process_id == process.process_id + && birth == Some(identity.process_birth_id) + && process + .executable_path + .as_ref() + .and_then(|path| path.canonicalize().ok()) + .as_ref() + == Some(&identity.executable) + }) +} + +fn main_target_absent(debug_port: u16) -> anyhow::Result { + let neighbors = verified_main_neighbors(); + if codex_plus_core::cdp::endpoint_available(debug_port) + && !neighbors + .iter() + .any(|identity| identity.debug_port == debug_port) + { + return Ok(false); + } + // 无身份时仅用于启动准入，不按这个候选集合终止任何进程。 + #[cfg(windows)] + { + let processes = codex_plus_core::windows_enumerate_processes(); + anyhow::ensure!(!processes.is_empty(), "无法确认主实例进程已退出"); + let candidates = codex_plus_core::watcher::find_codex_processes_from_snapshot(&processes); + for process in processes + .iter() + .filter(|process| candidates.contains(&process.process_id)) + { + if main_candidate_is_verified_neighbor( + process, + &neighbors, + codex_plus_core::windows_process_birth_id(process.process_id), + ) { + continue; + } + if let Some(arguments) = + codex_plus_core::windows_process_arguments(process.process_id) + { + if main_command_is_electron_child(&arguments) { + continue; + } + let internal_cli = process + .executable_path + .as_ref() + .is_some_and(|path| main_command_is_bundled_app_server(path, &arguments)); + if internal_cli { + continue; + } + } + return Ok(false); + } + Ok(true) + } + #[cfg(not(windows))] + { + Ok(codex_plus_core::watcher::find_codex_processes() + .into_iter() + .all(|pid| { + neighbors.iter().any(|identity| { + identity.process_id == pid + && !codex_plus_core::multi_worker::codex_instance_process_exited(identity) + }) + })) + } +} + +fn main_control_request(mut request: LaunchRequest) -> LaunchRequest { + let session = codex_plus_launcher::session_snapshot(); + let latest = StatusStore::default().load_latest().ok().flatten(); + request = main_control_request_from_status(request, &session, latest.as_ref()); + normalize_launch_request(request) +} + +fn main_control_request_from_status( + mut request: LaunchRequest, + session: &codex_plus_launcher::SessionSnapshot, + latest: Option<&LaunchStatus>, +) -> LaunchRequest { + request.debug_port = session + .owned + .then_some(session.debug_port) + .flatten() + .or_else(|| latest.and_then(|status| status.debug_port)) + .unwrap_or(request.debug_port); + if let Some(latest) = latest { + request.helper_port = latest.helper_port.unwrap_or(request.helper_port); + if let Some(app) = &latest.codex_app { + request.app_path = app.clone(); + } + } + request +} + +struct MainCodexStopTarget(Option); + +impl MainCodexStopTarget { + fn capture(request: &LaunchRequest) -> anyhow::Result { + let session = codex_plus_launcher::session_snapshot(); + if let Some(receipt) = load_main_identity()? { + anyhow::ensure!( + receipt.identity.debug_port == request.debug_port + && (!session.owned + || session.started_at_ms == Some(receipt.identity.generation)), + "主实例身份记录与当前会话代际或端口不匹配" + ); + return Ok(Self(Some(receipt.identity))); + } + anyhow::ensure!( + !session.owned && main_target_absent(request.debug_port)?, + "缺少可信主实例身份，未执行停止或重新启动；请手动关闭旧实例后启动" + ); + Ok(Self(None)) + } + + fn stop_and_wait(self) -> anyhow::Result<()> { + let Some(identity) = self.0 else { + return Ok(()); + }; + if codex_plus_core::multi_worker::codex_instance_process_exited(&identity) { + return Ok(()); + } + anyhow::ensure!( + tauri::async_runtime::block_on( + codex_plus_core::multi_worker::terminate_verified_codex_instance(&identity), + )?, + "主实例身份已改变，未执行停止" + ); + { + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); + loop { + if codex_plus_core::multi_worker::codex_instance_process_exited(&identity) { + break; + } + anyhow::ensure!(std::time::Instant::now() < deadline, "主实例尚未确认退出"); + std::thread::sleep(std::time::Duration::from_millis(100)); + } + } + Ok(()) } } fn restart_codex_plus_blocking(request: LaunchRequest) -> CommandResult { + if let Err(error) = prepare_main_launch_options(Default::default()) { + return failed( + &format!("显式启动目录无效，未执行重启：{error:#}"), + json!({}), + ); + } let _launch_guard = match try_acquire_launch_guard() { Ok(guard) => guard, Err(message) => return failed(message, json!({})), }; - let request = normalize_launch_request(request); - // 按本实例的调试端口判定是否在跑。 - // 早先用 `find_codex_processes_for_debug_port`，但它在非 macOS 上返回全部 - // Codex 进程：Windows 上只要有多开实例在跑，就会把「主 Codex 没开」误判成 - // 「已运行」，进而走进重启分支去停进程。 - let app_running = codex_plus_core::cdp::endpoint_available(request.debug_port); - if restart_disposition(request.sync_active_relay, app_running) == RestartDisposition::LaunchOnly + let request = main_control_request(request); + let target = match MainCodexStopTarget::capture(&request) { + Ok(target) => target, + Err(error) => return failed(&format!("无法确认重启目标：{error}"), json!({})), + }; + if codex_plus_core::app_paths::resolve_codex_app_dir( + (!request.app_path.trim().is_empty()).then(|| Path::new(request.app_path.trim())), + ) + .is_none() { - return spawn_codex_plus_launch(request, "Codex 未运行，已按直接启动方式在后台唤起。"); + return failed("无法定位 Codex 应用，未执行重启。", json!({})); } let Ok(_guard) = relay_switch_mutex().try_lock() else { return failed( @@ -2314,6 +2895,12 @@ fn restart_codex_plus_blocking(request: LaunchRequest) -> CommandResult { Ok(guard) => guard, Err(message) => return failed(&message, json!({})), }; + if let Err(error) = preflight_restart_helper(settings.as_ref(), request.helper_port) { + return failed( + &format!("代理端口尚未就绪，未停止 Codex：{error:#}"), + json!({"debugPort": request.debug_port, "helperPort": request.helper_port}), + ); + } // 停止 Codex 后清理同进程后台服务，再确认浏览器恢复结果。 // 先记录停止阶段，让界面在等待期间显示进度（issue #2403）。 // 该状态带的是当前时间戳，早于后面返回给前端的 launch_started_at_ms， @@ -2327,14 +2914,17 @@ fn restart_codex_plus_blocking(request: LaunchRequest) -> CommandResult { 10, )); let native_browser_shutdown = match stop_codex_plus_for_restart( - || stop_main_codex_process(request.debug_port), + || target.stop_and_wait(), codex_plus_launcher::stop, wait_native_browser_shutdown, ) { Ok(shutdown) => shutdown, Err(error) => { + let message = restart_stop_failure_message(&error); + let _ = + save_requested_launch_status(&request, "failed", &message, current_timestamp_ms()); return failed( - &restart_stop_failure_message(&error), + &message, json!({"debugPort": request.debug_port, "helperPort": request.helper_port}), ); } @@ -2350,8 +2940,10 @@ fn restart_codex_plus_blocking(request: LaunchRequest) -> CommandResult { ); } if let Err(error) = prepare_fixed_helper_port_for_restart(settings.as_ref()) { + let message = format!("旧 Codex 已停止，替代实例尚未启动：{error:#}"); + let _ = save_requested_launch_status(&request, "failed", &message, current_timestamp_ms()); return failed( - &format!("重启 Codex 失败：{error:#}"), + &message, json!({"debugPort": request.debug_port, "helperPort": request.helper_port}), ); } @@ -2373,7 +2965,7 @@ fn restart_codex_plus_blocking(request: LaunchRequest) -> CommandResult { launch_started_at_ms, ) { return failed( - &format!("记录重启状态失败，未执行重启：{error}"), + &format!("旧 Codex 已停止，但记录启动状态失败，替代实例未启动：{error}"), json!({ "debugPort": request.debug_port, "helperPort": request.helper_port, @@ -2417,26 +3009,10 @@ fn restart_codex_plus_blocking(request: LaunchRequest) -> CommandResult { "syncActiveRelay": request.sync_active_relay, "launchStartedAtMs": launch_started_at_ms, "nativeBrowserRestoreFailed": native_browser_restore_failed - }), - ) - } - } -} - -/// 结束主 Codex 进程并确认调试端口已关闭。重启和「停止 Codex」共用。 -fn stop_main_codex_process(debug_port: u16) -> anyhow::Result<()> { - // 主实例在 Windows 上是 AUMID 激活的，命令行未必带调试端口； - // 这里允许「按端口没匹配到就退回全局结束」，否则会卡在 - // 「Codex 进程尚未退出」。多开页的结束走的是不允许退回的那个版本。 - #[cfg(windows)] - codex_plus_core::watcher::stop_codex_processes_for_debug_port_with_fallback(debug_port, true); - #[cfg(not(windows))] - codex_plus_core::watcher::stop_codex_processes_for_debug_port_and_wait(debug_port); - anyhow::ensure!( - !codex_plus_core::cdp::endpoint_available(debug_port), - "Codex 进程尚未退出" - ); - Ok(()) + }), + ) + } + } } fn wait_native_browser_shutdown() @@ -2459,10 +3035,11 @@ fn stop_codex_plus_blocking(request: LaunchRequest) -> CommandResult { Ok(guard) => guard, Err(message) => return failed(message, json!({})), }; - let request = normalize_launch_request(request); - if !codex_plus_core::cdp::endpoint_available(request.debug_port) { - return ok("Codex 没有在运行。", json!({})); - } + let request = main_control_request(request); + let target = match MainCodexStopTarget::capture(&request) { + Ok(target) => target, + Err(error) => return failed(&format!("无法确认停止目标：{error}"), json!({})), + }; let Ok(_guard) = relay_switch_mutex().try_lock() else { return failed("供应商切换正在进行，未停止 Codex；请稍后重试。", json!({})); }; @@ -2479,22 +3056,39 @@ fn stop_codex_plus_blocking(request: LaunchRequest) -> CommandResult { 10, )); match stop_codex_plus_for_restart( - || stop_main_codex_process(request.debug_port), + || target.stop_and_wait(), codex_plus_launcher::stop, wait_native_browser_shutdown, ) { - Ok(_) => { + Ok(shutdown) => { + let restore_failed = matches!( + shutdown, + codex_plus_core::native_browser::NativeBrowserShutdown::RestoreFailed + ); + let message = if restore_failed { + "Codex 已停止，但原生浏览器文件恢复失败，请检查恢复状态。" + } else { + "Codex 已停止。" + }; let _ = StatusStore::default().save_latest(&requested_launch_status_with_phase( &request, "stopped", - "Codex 已停止。", + message, current_timestamp_ms(), "user_stopped", 100, )); - ok("Codex 已停止。", json!({})) + ok( + message, + json!({"nativeBrowserRestoreFailed": restore_failed}), + ) + } + Err(error) => { + let message = stop_failure_message(&error); + let _ = + save_requested_launch_status(&request, "failed", &message, current_timestamp_ms()); + failed(&message, json!({})) } - Err(error) => failed(&stop_failure_message(&error), json!({})), } } @@ -2656,6 +3250,32 @@ fn probe_loopback_port(port: u16) -> std::io::Result<()> { Ok(()) } +fn preflight_restart_helper( + settings: Option<&BackendSettings>, + helper_port: u16, +) -> anyhow::Result<()> { + let port = settings + .and_then(codex_plus_core::launcher::required_fixed_helper_port) + .unwrap_or(helper_port); + if codex_plus_core::resident_helper::resident_helper_port() == Some(port) + || probe_loopback_port(port).is_ok() + { + return Ok(()); + } + let healthy = tauri::async_runtime::block_on(async { + let client = codex_plus_core::http_client::local_helper_client()?; + let response = client.get(format!("http://127.0.0.1:{port}/backend/status")).send().await?; + anyhow::ensure!(response.status().is_success(), "占用端口的服务未就绪"); + let body: Value = response.json().await?; + anyhow::ensure!( + body["status"] == "ok" && body["transport"] == "http-helper", + "占用端口的服务不是可用 helper" + ); + Ok::<_, anyhow::Error>(()) + }); + healthy +} + fn spawn_after_provider_guard_release( guard: codex_plus_data::ProviderSyncLifecycleGuard, request: &LaunchRequest, @@ -2852,6 +3472,12 @@ fn normalize_launch_request(mut request: LaunchRequest) -> LaunchRequest { } fn spawn_codex_plus_launch(request: LaunchRequest, accepted_message: &str) -> CommandResult { + if let Err(error) = prepare_main_launch_options(Default::default()) { + return failed( + &format!("显式启动目录无效，未执行启动：{error:#}"), + json!({}), + ); + } let request = normalize_launch_request(request); let debug_port = request.debug_port; let helper_port = request.helper_port; @@ -2864,19 +3490,23 @@ fn spawn_codex_plus_launch(request: LaunchRequest, accepted_message: &str) -> Co "app_path": request.app_path.trim() }), ); - if let Err(error) = save_requested_launch_status( - &request, - "starting", - "Codex++ launcher is starting", - launch_started_at_ms, - ) { - return failed( - &format!("记录启动状态失败，未执行启动：{error}"), - json!({ - "debugPort": debug_port, - "helperPort": helper_port - }), - ); + // 重复激活保留原代健康状态，launcher 据此保留 running_degraded。 + let session = codex_plus_launcher::session_snapshot(); + if !(session.owned && session.ready) { + if let Err(error) = save_requested_launch_status( + &request, + "starting", + "Codex++ launcher is starting", + launch_started_at_ms, + ) { + return failed( + &format!("记录启动状态失败，未执行启动：{error}"), + json!({ + "debugPort": debug_port, + "helperPort": helper_port + }), + ); + } } match start_embedded_launcher(&request) { Ok(()) => { @@ -2992,10 +3622,7 @@ fn current_timestamp_ms() -> u64 { fn start_embedded_launcher(request: &LaunchRequest) -> anyhow::Result<()> { anyhow::ensure!(!crate::app_is_exiting(), "Codex++ 正在退出，未启动 Codex"); - // 先让常驻 helper 占住协议代理端口：启动链路会识别并复用，Codex 会话退出 - // 时就不再关闭它，从而实现「Codex++ 在跑，本地代理就一直可用」。 - sync_resident_helper(); - codex_plus_launcher::start(codex_plus_launcher::LaunchOptions { + start_main_launcher(codex_plus_launcher::LaunchOptions { app_dir: (!request.app_path.trim().is_empty()) .then(|| PathBuf::from(request.app_path.trim())), debug_port: request.debug_port, @@ -3005,16 +3632,294 @@ fn start_embedded_launcher(request: &LaunchRequest) -> anyhow::Result<()> { }) } +fn record_main_launcher_identity( + receipt: &codex_plus_launcher::LaunchReceipt, +) -> anyhow::Result<()> { + let options = prepare_main_launch_options(Default::default())?; + let home = codex_plus_core::relay_config::default_codex_home_dir().canonicalize()?; + let sqlite_home = options.sqlite_home.clone().unwrap_or_else(|| home.clone()); + if let Ok(Some(previous)) = load_main_identity() { + if previous.identity.generation == receipt.started_at_ms + && previous.identity.debug_port == receipt.debug_port + && previous.sqlite_home.as_ref() == Some(&sqlite_home) + { + return Ok(()); + } + } + // 新会话确认前先使旧回执失效；失败时保留 unknown，不能控制旧身份。 + codex_plus_core::settings::atomic_write(&main_identity_path(), b"null\n")?; + let settings = SettingsStore::default().load()?; + let user_data_dir = main_explicit_user_data_dir(&settings.codex_extra_args)?; + let executable = codex_plus_core::app_paths::build_codex_executable(&receipt.app_dir); + let identity = tauri::async_runtime::block_on( + codex_plus_core::multi_worker::capture_main_codex_instance_identity( + receipt.started_at_ms, + receipt.debug_port, + receipt.helper_port, + user_data_dir.as_deref(), + &executable, + ), + )?; + let receipt = MainCodexIdentity { + version: 1, + state_dir: codex_plus_core::paths::default_app_state_dir().canonicalize()?, + codex_home: home, + sqlite_home: Some(sqlite_home), + identity, + }; + codex_plus_core::settings::atomic_write( + &main_identity_path(), + &serde_json::to_vec_pretty(&receipt)?, + ) +} + +fn main_explicit_user_data_dir(arguments: &[String]) -> anyhow::Result> { + let arguments = codex_plus_core::settings::normalize_codex_extra_args(arguments); + let mut path = None; + let mut tokens = arguments.iter(); + while let Some(token) = tokens.next() { + let value = if token == "--user-data-dir" { + Some( + tokens + .next() + .ok_or_else(|| anyhow::anyhow!("缺少 Electron 数据目录参数"))? + .as_str(), + ) + } else { + token.strip_prefix("--user-data-dir=") + }; + if let Some(value) = value { + anyhow::ensure!( + path.is_none() && !value.is_empty(), + "Electron 数据目录参数为空或重复" + ); + path = Some(PathBuf::from(value)); + } + } + // None 由 core 核验 argv 确实没有目录参数；不推断 Electron 的默认目录。 + Ok(path) +} + +fn remember_main_launcher_identity(receipt: &codex_plus_launcher::LaunchReceipt) { + if let Err(error) = record_main_launcher_identity(receipt) { + // 应用已启动；身份记录失败单独报告，不能伪装成应用启动失败。 + let _ = codex_plus_core::diagnostic_log::append_diagnostic_log( + "manager.main_identity_unavailable", + json!({"message": error.to_string()}), + ); + } +} + +fn validated_main_directory(name: &str, path: PathBuf) -> anyhow::Result { + anyhow::ensure!( + path.to_str().is_some_and(|value| !value.trim().is_empty()) && path.is_absolute(), + "{name} 必须是非空的绝对目录，已拒绝回落默认目录" + ); + let canonical = path + .canonicalize() + .map_err(|error| anyhow::anyhow!("{name} 目录不可访问，已拒绝回落默认目录：{error}"))?; + anyhow::ensure!(canonical.is_dir(), "{name} 必须指向已存在的目录"); + Ok(canonical) +} + +fn apply_main_launch_directories( + mut options: codex_plus_launcher::LaunchOptions, + codex_home: Option, + sqlite_home: Option, +) -> anyhow::Result { + options.codex_home = options + .codex_home + .or(codex_home) + .map(|path| validated_main_directory("CODEX_HOME", path)) + .transpose()?; + options.sqlite_home = options + .sqlite_home + .or(sqlite_home) + .map(|path| validated_main_directory("CODEX_SQLITE_HOME", path)) + .transpose()?; + if options.codex_home.is_some() || options.sqlite_home.is_some() { + options + .state_dir + .get_or_insert_with(codex_plus_core::paths::default_app_state_dir); + } + Ok(options) +} + +fn prepare_main_launch_options( + options: codex_plus_launcher::LaunchOptions, +) -> anyhow::Result { + // 包激活不继承管理器环境，显式目录必须随启动请求进入包内 wrapper。 + apply_main_launch_directories( + options, + std::env::var_os("CODEX_HOME").map(PathBuf::from), + std::env::var_os("CODEX_SQLITE_HOME").map(PathBuf::from), + ) +} + +fn main_launch_has_explicit_directories(options: &codex_plus_launcher::LaunchOptions) -> bool { + options.codex_home.is_some() || options.sqlite_home.is_some() +} + +fn validate_main_resume_scope( + options: &codex_plus_launcher::LaunchOptions, + settings: &BackendSettings, + receipt: &MainCodexIdentity, + session: &codex_plus_launcher::SessionSnapshot, +) -> anyhow::Result<()> { + let state_dir = options + .state_dir + .clone() + .unwrap_or_else(codex_plus_core::paths::default_app_state_dir); + let home = options + .codex_home + .clone() + .unwrap_or_else(codex_plus_core::relay_config::default_codex_home_dir) + .canonicalize()?; + let sqlite_home = options.sqlite_home.as_ref().unwrap_or(&home); + anyhow::ensure!( + main_identity_in_scope(receipt, &state_dir, &home) + && receipt.sqlite_home.as_ref() == Some(sqlite_home), + "显式目录与持久主实例身份不一致或缺少 SQLite 归属，拒绝恢复" + ); + let identity = &receipt.identity; + anyhow::ensure!( + identity.debug_port == options.debug_port && identity.helper_port == options.helper_port, + "主实例端口与持久身份不一致，拒绝恢复" + ); + let app_dir = codex_plus_core::app_paths::resolve_codex_app_dir_with_saved( + options.app_dir.as_deref(), + Some(settings.codex_app_path.as_str()), + ) + .ok_or_else(|| anyhow::anyhow!("缺少主实例应用路径，拒绝恢复"))?; + let executable = codex_plus_core::app_paths::build_codex_executable(&app_dir).canonicalize()?; + let user_data = main_explicit_user_data_dir(&settings.codex_extra_args)? + .map(|path| path.canonicalize()) + .transpose()? + .unwrap_or_default(); + anyhow::ensure!( + executable == identity.executable && user_data == identity.user_data_dir, + "主实例映像或 Electron 数据目录与持久身份不一致，拒绝恢复" + ); + anyhow::ensure!( + !session.owned + || (session.ready + && session.started_at_ms == Some(identity.generation) + && session.debug_port == Some(identity.debug_port)), + "当前 launcher 会话与持久主实例身份不一致，拒绝复用" + ); + Ok(()) +} + +fn verified_main_resume_target( + options: &codex_plus_launcher::LaunchOptions, + settings: &BackendSettings, +) -> anyhow::Result { + let session = codex_plus_launcher::session_snapshot(); + let Some(receipt) = load_main_identity()? else { + validate_missing_main_identity(&session, || main_target_absent(options.debug_port))?; + return Ok(false); + }; + if codex_plus_core::multi_worker::codex_instance_process_exited(&receipt.identity) { + validate_missing_main_identity(&session, || main_target_absent(options.debug_port))?; + return Ok(false); + } + validate_main_resume_scope( + options, + settings, + &receipt, + &session, + )?; + anyhow::ensure!( + tauri::async_runtime::block_on( + codex_plus_core::multi_worker::codex_instance_identity_matches(&receipt.identity) + )?, + "无法确认原代主实例的进程及 CDP 归属，拒绝恢复或复用" + ); + Ok(true) +} + +fn validate_missing_main_identity( + session: &codex_plus_launcher::SessionSnapshot, + target_absent: impl FnOnce() -> anyhow::Result, +) -> anyhow::Result<()> { + anyhow::ensure!( + !session.owned && target_absent()?, + "显式目录缺少持久主实例身份，且无法确认目标不存在，拒绝恢复或复用" + ); + Ok(()) +} + +fn start_main_launcher(options: codex_plus_launcher::LaunchOptions) -> anyhow::Result<()> { + let options = prepare_main_launch_options(options)?; + if main_launch_has_explicit_directories(&options) + && codex_plus_launcher::session_snapshot().owned + { + let settings = SettingsStore::default().load()?; + anyhow::ensure!( + verified_main_resume_target(&options, &settings)?, + "原代主实例已退出，拒绝复用旧 launcher 会话" + ); + } + // 校验完成后才绑定 helper，供启动链路识别并复用。 + sync_resident_helper(); + let receipt = codex_plus_launcher::start_with_receipt(options)?; + remember_main_launcher_identity(&receipt); + Ok(()) +} + +fn remember_main_ready_session() { + let session = codex_plus_launcher::session_snapshot(); + if !session.owned || !session.ready { + return; + } + let Some(latest) = StatusStore::default().load_latest().ok().flatten() else { + return; + }; + let Some(app_dir) = latest.codex_app.as_deref().map(PathBuf::from) else { + return; + }; + let (Some(debug_port), Some(helper_port), Some(started_at_ms)) = ( + session.debug_port, + latest.helper_port, + session.started_at_ms, + ) else { + return; + }; + remember_main_launcher_identity(&codex_plus_launcher::LaunchReceipt { + debug_port, + helper_port, + app_dir, + started_at_ms, + }); +} + /// 重新打开管理界面时恢复已有 Codex 的 Helper、代理和注入，不启动官方应用。 pub fn resume_existing_codex_background(app: tauri::AppHandle, start_if_missing: bool) { tauri::async_runtime::spawn_blocking(move || { if crate::app_is_exiting() { return; } - // 必须先于 resume：resume 会走 launch_and_inject，如果它先 bind 了 helper， - // Codex 退出时就把端口带走了；这里先托管常驻实例，会话会复用同一监听。 - sync_resident_helper(); - let settings = SettingsStore::default().load().unwrap_or_default(); + let directory_options = match prepare_main_launch_options(Default::default()) { + Ok(options) => options, + Err(error) => { + let _ = codex_plus_core::diagnostic_log::append_diagnostic_log( + "manager.resume_existing_failed", + json!({ "message": format!("{error:#}") }), + ); + return; + } + }; + let settings = match SettingsStore::default().load() { + Ok(settings) => settings, + Err(error) if main_launch_has_explicit_directories(&directory_options) => { + let _ = codex_plus_core::diagnostic_log::append_diagnostic_log( + "manager.resume_existing_failed", + json!({ "message": format!("无法读取隔离实例设置，拒绝恢复：{error:#}") }), + ); + return; + } + Err(_) => BackendSettings::default(), + }; let latest = StatusStore::default().load_latest().unwrap_or(None); let options = codex_plus_launcher::LaunchOptions { app_dir: codex_plus_core::app_paths::resolve_codex_app_dir_with_saved( @@ -3033,12 +3938,27 @@ pub fn resume_existing_codex_background(app: tauri::AppHandle, start_if_missing: .and_then(|status| status.helper_port) .unwrap_or_else(default_helper_port), status_store: StatusStore::default(), - ..Default::default() + ..directory_options }; - match codex_plus_launcher::resume_if_running(options.clone()) { - Ok(true) => {} + let resumed = (|| -> anyhow::Result { + if main_launch_has_explicit_directories(&options) + && !verified_main_resume_target(&options, &settings)? + { + return Ok(false); + } + // 归属校验完成后才绑定 helper 或进入 launcher 的恢复操作。 + sync_resident_helper(); + let resumed = codex_plus_launcher::resume_if_running(options.clone())?; + anyhow::ensure!( + resumed || !main_launch_has_explicit_directories(&options), + "身份校验后主实例连接已失联，拒绝回退启动" + ); + Ok(resumed) + })(); + match resumed { + Ok(true) => remember_main_ready_session(), Ok(false) if start_if_missing => { - if let Err(error) = codex_plus_launcher::start(options) { + if let Err(error) = start_main_launcher(options) { let _ = codex_plus_core::diagnostic_log::append_diagnostic_log( "manager.autostart_codex_failed", json!({ "message": error.to_string() }), @@ -9473,6 +10393,213 @@ fn default_log_lines() -> usize { mod tests { use super::*; + #[test] + fn explicit_main_directories_reject_invalid_paths_without_fallback() { + let temp = tempfile::tempdir().unwrap(); + let file = temp.path().join("file"); + fs::write(&file, b"not a directory").unwrap(); + for path in [ + PathBuf::new(), + PathBuf::from(" "), + PathBuf::from("relative-home"), + temp.path().join("missing"), + file, + ] { + for sqlite in [false, true] { + let (home, sqlite_home) = if sqlite { + (None, Some(path.clone())) + } else { + (Some(path.clone()), None) + }; + assert!( + apply_main_launch_directories(Default::default(), home, sqlite_home).is_err(), + "accepted invalid explicit directory: {path:?}" + ); + } + } + } + + #[test] + fn explicit_main_directories_preserve_default_and_options_precedence() { + let default = apply_main_launch_directories(Default::default(), None, None).unwrap(); + assert!(default.codex_home.is_none()); + assert!(default.sqlite_home.is_none()); + assert!(default.state_dir.is_none()); + assert!(!main_launch_has_explicit_directories(&default)); + + let temp = tempfile::tempdir().unwrap(); + let home = temp.path().join("home"); + let sqlite = temp.path().join("sqlite"); + fs::create_dir(&home).unwrap(); + fs::create_dir(&sqlite).unwrap(); + let options = apply_main_launch_directories( + codex_plus_launcher::LaunchOptions { + state_dir: Some(temp.path().to_path_buf()), + codex_home: Some(home.clone()), + sqlite_home: Some(sqlite.clone()), + ..Default::default() + }, + Some(PathBuf::from("invalid-lower-priority-home")), + Some(PathBuf::from("invalid-lower-priority-sqlite")), + ) + .unwrap(); + assert_eq!(options.codex_home, Some(home.canonicalize().unwrap())); + assert_eq!(options.sqlite_home, Some(sqlite.canonicalize().unwrap())); + assert_eq!(options.state_dir.as_deref(), Some(temp.path())); + assert!(main_launch_has_explicit_directories(&options)); + for sqlite_only in [false, true] { + let options = apply_main_launch_directories( + codex_plus_launcher::LaunchOptions { + state_dir: Some(temp.path().to_path_buf()), + ..Default::default() + }, + (!sqlite_only).then(|| home.clone()), + sqlite_only.then(|| sqlite.clone()), + ) + .unwrap(); + assert!(main_launch_has_explicit_directories(&options)); + assert_eq!(options.codex_home.is_some(), !sqlite_only); + assert_eq!(options.sqlite_home.is_some(), sqlite_only); + } + } + + fn explicit_main_resume_fixture( + root: &Path, + ) -> ( + codex_plus_launcher::LaunchOptions, + BackendSettings, + MainCodexIdentity, + codex_plus_launcher::SessionSnapshot, + ) { + let executable = codex_plus_core::app_paths::build_codex_executable(root); + fs::create_dir_all(executable.parent().unwrap()).unwrap(); + fs::write(&executable, b"fixture only").unwrap(); + fs::create_dir(root.join("resources")).unwrap(); + fs::write(root.join("resources/app.asar"), b"fixture only").unwrap(); + let root = root.canonicalize().unwrap(); + let options = codex_plus_launcher::LaunchOptions { + app_dir: Some(root.clone()), + debug_port: 49101, + helper_port: 49102, + state_dir: Some(root.clone()), + codex_home: Some(root.clone()), + sqlite_home: Some(root.clone()), + ..Default::default() + }; + let receipt = MainCodexIdentity { + version: 1, + state_dir: root.clone(), + codex_home: root.clone(), + sqlite_home: Some(root), + identity: codex_plus_core::multi_worker::CodexInstanceIdentity { + instance_id: "main".into(), + generation: 7, + debug_port: options.debug_port, + helper_port: options.helper_port, + user_data_dir: PathBuf::new(), + browser_id: "fixture-browser".into(), + process_id: 123, + process_birth_id: 456, + executable: executable.canonicalize().unwrap(), + }, + }; + let session = codex_plus_launcher::SessionSnapshot { + owned: true, + ready: true, + status: Some("running".into()), + debug_port: Some(options.debug_port), + started_at_ms: Some(7), + }; + (options, BackendSettings::default(), receipt, session) + } + + #[test] + fn explicit_main_first_start_requires_confirmed_absence_without_a_receipt() { + let mut session = codex_plus_launcher::SessionSnapshot { + owned: false, ready: false, status: None, debug_port: None, started_at_ms: None, + }; + validate_missing_main_identity(&session, || Ok(true)).unwrap(); + assert!(validate_missing_main_identity(&session, || Ok(false)).is_err()); + assert!(validate_missing_main_identity(&session, || anyhow::bail!("probe unavailable")).is_err()); + session.owned = true; + assert!(validate_missing_main_identity(&session, || panic!("must not probe an owned session")).is_err()); + } + + #[test] + fn explicit_main_resume_preserves_default_electron_directory() { + let temp = tempfile::tempdir().unwrap(); + let (options, settings, receipt, mut session) = explicit_main_resume_fixture(temp.path()); + validate_main_resume_scope(&options, &settings, &receipt, &session).unwrap(); + session.owned = false; + session.ready = false; + session.started_at_ms = None; + validate_main_resume_scope(&options, &settings, &receipt, &session).unwrap(); + } + + #[test] + fn explicit_main_resume_rejects_incomplete_or_foreign_receipts() { + let temp = tempfile::tempdir().unwrap(); + let foreign_dir = tempfile::tempdir().unwrap(); + let foreign = foreign_dir.path().canonicalize().unwrap(); + let (options, settings, receipt, session) = explicit_main_resume_fixture(temp.path()); + for change in 0..9 { + let mut wrong = receipt.clone(); + match change { + 0 => wrong.sqlite_home = None, + 1 => wrong.sqlite_home = Some(foreign.clone()), + 2 => wrong.codex_home = foreign.clone(), + 3 => wrong.state_dir = foreign.clone(), + 4 => wrong.identity.executable = foreign.join("other.exe"), + 5 => wrong.identity.user_data_dir = foreign.clone(), + 6 => wrong.identity.debug_port += 1, + 7 => wrong.identity.helper_port += 1, + 8 => wrong.identity.generation += 1, + _ => unreachable!(), + } + assert!( + validate_main_resume_scope(&options, &settings, &wrong, &session).is_err(), + "accepted conflicting receipt case {change}" + ); + } + let mut old = serde_json::to_value(&receipt).unwrap(); + old.as_object_mut().unwrap().remove("sqlite_home"); + let old: MainCodexIdentity = serde_json::from_value(old).unwrap(); + assert!(validate_main_resume_scope(&options, &settings, &old, &session).is_err()); + for change in 0..3 { + let mut wrong = session.clone(); + match change { + 0 => wrong.ready = false, + 1 => wrong.started_at_ms = None, + 2 => wrong.debug_port = Some(49103), + _ => unreachable!(), + } + assert!(validate_main_resume_scope(&options, &settings, &receipt, &wrong).is_err()); + } + } + + #[test] + fn explicit_main_resume_requires_matching_electron_directory_arguments() { + let temp = tempfile::tempdir().unwrap(); + let (options, mut settings, mut receipt, session) = + explicit_main_resume_fixture(temp.path()); + let data = temp.path().join("user-data"); + fs::create_dir(&data).unwrap(); + receipt.identity.user_data_dir = data.canonicalize().unwrap(); + assert!(validate_main_resume_scope(&options, &settings, &receipt, &session).is_err()); + for args in [ + vec![ + "--user-data-dir".to_string(), + data.to_string_lossy().into_owned(), + ], + vec![format!("--user-data-dir={}", data.display())], + ] { + settings.codex_extra_args = args; + validate_main_resume_scope(&options, &settings, &receipt, &session).unwrap(); + } + settings.codex_extra_args.push("--user-data-dir=".into()); + assert!(validate_main_resume_scope(&options, &settings, &receipt, &session).is_err()); + } + #[test] fn skills_payload_hides_disabled_repository_catalog() { let temp = tempfile::tempdir().unwrap(); @@ -9968,17 +11095,109 @@ base_url = "https://example.invalid/v1" } #[test] - fn restart_absent_app_uses_launch_but_explicit_sync_still_restarts() { + fn main_without_owned_identity_requires_confirmed_absence() { + let session = codex_plus_launcher::SessionSnapshot { + owned: false, + ready: false, + status: Some("failed".into()), + debug_port: Some(9229), + started_at_ms: Some(1), + }; assert_eq!( - restart_disposition(false, false), - RestartDisposition::LaunchOnly + main_running_with(&session, None, |_| panic!("no identity"), || Some(true)), + Some(false) ); - for (sync, running) in [(true, false), (false, true), (true, true)] { - assert_eq!( - restart_disposition(sync, running), - RestartDisposition::StopAndRestart - ); - } + assert_eq!( + main_running_with(&session, None, |_| panic!("no identity"), || Some(false)), + None + ); + assert_eq!( + main_running_with(&session, None, |_| panic!("no identity"), || None), + None + ); + } + + #[test] + fn main_identity_null_is_absent_and_stale_ready_does_not_override_exit() { + let temp = tempfile::tempdir().unwrap(); + assert!( + decode_main_identity(b"null\n", temp.path(), temp.path()) + .unwrap() + .is_none() + ); + let identity = codex_plus_core::multi_worker::CodexInstanceIdentity { + instance_id: "main".into(), + generation: 7, + debug_port: 49101, + helper_port: 49102, + user_data_dir: PathBuf::new(), + browser_id: "browser".into(), + process_id: 123, + process_birth_id: 456, + executable: temp.path().join("Codex.exe"), + }; + let receipt = MainCodexIdentity { + version: 1, + state_dir: temp.path().canonicalize().unwrap(), + codex_home: temp.path().canonicalize().unwrap(), + sqlite_home: None, + identity: identity.clone(), + }; + let bytes = serde_json::to_vec(&receipt).unwrap(); + assert!( + decode_main_identity(&bytes, temp.path(), temp.path()) + .unwrap() + .is_some() + ); + let foreign = tempfile::tempdir().unwrap(); + assert!(decode_main_identity(&bytes, foreign.path(), temp.path()).is_err()); + let session = codex_plus_launcher::SessionSnapshot { + owned: true, + ready: true, + status: Some("running".into()), + debug_port: Some(49101), + started_at_ms: Some(7), + }; + assert_eq!( + main_running_with( + &session, + Some(&identity), + |_| Some(false), + || panic!("has identity") + ), + Some(false) + ); + assert_eq!( + main_running_with( + &session, + Some(&identity), + |_| None, + || panic!("has identity") + ), + None + ); + } + + #[test] + fn main_control_uses_recorded_ports_instead_of_default_form() { + let session = codex_plus_launcher::SessionSnapshot { + owned: true, + ready: true, + status: Some("running_degraded".into()), + debug_port: Some(49101), + started_at_ms: Some(7), + }; + let latest = LaunchStatus { + debug_port: Some(49101), + helper_port: Some(49102), + codex_app: Some("saved-app".into()), + ..LaunchStatus::default() + }; + let request = + main_control_request_from_status(launch_request(false), &session, Some(&latest)); + assert_eq!(request.debug_port, 49101); + assert_eq!(request.helper_port, 49102); + assert_eq!(request.app_path, "saved-app"); } #[test] @@ -11949,6 +13168,22 @@ model_reasoning_effort = "high" wrong.user_data_dir = temp.path().canonicalize().unwrap(); assert!(!multi_manifest_identity_matches(¤t, &wrong)); assert!(!multi_instance_is_running(&instance)); + let mut uncertain = current.clone(); + uncertain.generation += 1; + assert_eq!(multi_instance_snapshot_status(&uncertain, false), "unknown"); + uncertain.identity = None; + uncertain.launch_result.as_mut().unwrap().status = "cleanup_failed".into(); + assert_eq!(multi_instance_snapshot_status(&uncertain, false), "unknown"); + } + + #[test] + fn multi_instance_requests_preserve_generation_and_allow_legacy_clients() { + let current: CodexMultiInstanceRequest = + serde_json::from_value(json!({ "instanceId": "fixture", "generation": 7 })).unwrap(); + assert_eq!(current.generation, Some(7)); + let legacy: CodexMultiInstanceRequest = + serde_json::from_value(json!({ "instanceId": "fixture" })).unwrap(); + assert_eq!(legacy.generation, None); } #[cfg(unix)] @@ -12099,12 +13334,11 @@ model_reasoning_effort = "high" assert_eq!(instance.debug_port, 9456); } - #[cfg(unix)] #[test] fn multi_worker_stop_never_terminates_current_process() { let options = std::env::temp_dir().join("codex-plus-worker-options-test.json"); let pid = std::process::id(); - terminate_multi_worker(pid, &options, None).unwrap(); + assert!(terminate_multi_worker(pid, &options, None).is_err()); } #[test] diff --git a/apps/codex-plus-manager/src-tauri/src/commands/multi_instance_resume.rs b/apps/codex-plus-manager/src-tauri/src/commands/multi_instance_resume.rs new file mode 100644 index 000000000..8bf021516 --- /dev/null +++ b/apps/codex-plus-manager/src-tauri/src/commands/multi_instance_resume.rs @@ -0,0 +1,1679 @@ +use super::*; +use anyhow::{Context, ensure}; +use codex_plus_core::multi_worker::CodexInstanceIdentity; +use std::process::{Child, Command, Stdio}; +use std::sync::mpsc; +use std::time::{Duration, Instant}; + +const OPTIONS: &str = "worker-options.json"; +const READY: &str = "worker-ready.json"; +const REQUEST: &str = "worker-handoff.json"; +const RELEASE: &str = "worker-handoff-release.json"; +const COMMIT: &str = "worker-handoff-commit.json"; +const CANDIDATE: &str = "worker-handoff-candidate.json"; +const HANDOFF_ENV: &str = "CODEX_PLUS_WORKER_HANDOFF"; +const RESUME_LOCK: &str = "worker-resume.lock"; +const OUTCOME: &str = "worker-resume-result.json"; +const STAGE_TIMEOUT: Duration = Duration::from_secs(90); +const HANDOFF_LIFETIME: Duration = Duration::from_secs(180); +const HANDOFF_PROTOCOL: u32 = 2; +const POLL: Duration = Duration::from_millis(100); + +/// 记录 worker 启动时的文件属性；原路径被替换后不能据路径相同复用旧构建。 +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub(super) struct BinaryStamp { + executable: PathBuf, + size: u64, + modified_ns: String, + version: String, +} + +impl BinaryStamp { + pub(super) fn current() -> anyhow::Result { + Self::at(&std::env::current_exe()?) + } + + fn at(path: &Path) -> anyhow::Result { + let executable = path.canonicalize().context("无法定位新版 worker 二进制")?; + let file = fs::File::open(&executable).context("新版 worker 二进制不可读")?; + let metadata = file.metadata()?; + ensure!( + metadata.is_file() && metadata.len() != 0, + "新版 worker 二进制无效" + ); + Ok(Self { + executable, + size: metadata.len(), + modified_ns: metadata + .modified()? + .duration_since(UNIX_EPOCH)? + .as_nanos() + .to_string(), + version: env!("CARGO_PKG_VERSION").into(), + }) + } +} + +/// 交接始终绑定完整 ready；v1 worker 不理解撤销，不能向其发送退出请求。 +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub(super) struct WorkerReady { + protocol: u32, + instance_id: String, + generation: u64, + pid: u32, + worker_birth_id: Option, + token: String, + options_path: PathBuf, + identity: CodexInstanceIdentity, + binary: BinaryStamp, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +struct HandoffRequest { + protocol: u32, + token: String, + worker: WorkerReady, + requester_pid: u32, + requester_birth: String, + issued_at_ms: u64, + expires_at_ms: u64, + successor_token: String, + successor_binary: BinaryStamp, +} + +#[derive(Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +struct HandoffRelease { + request: HandoffRequest, + options: CodexMultiWorkerOptions, + released: bool, +} + +#[derive(Serialize, Deserialize)] +struct HandoffCommit { + request: HandoffRequest, + successor: WorkerReady, +} + +fn read_json(path: &Path) -> Option { + serde_json::from_slice(&fs::read(path).ok()?).ok() +} + +fn open_resume_lock(root: &Path) -> anyhow::Result { + let path = root.join(RESUME_LOCK); + if let Ok(metadata) = fs::symlink_metadata(&path) { + ensure!( + !metadata.file_type().is_symlink(), + "worker 交接锁不能重定向" + ); + } + Ok(fs::OpenOptions::new() + .create(true) + .truncate(false) + .read(true) + .write(true) + .open(path)?) +} + +fn process_birth(pid: u32) -> Option { + #[cfg(windows)] + { + codex_plus_core::windows_process_birth_id(pid).map(|birth| birth.to_string()) + } + #[cfg(not(windows))] + { + match codex_plus_core::watcher::inspect_process_instance(pid) { + codex_plus_core::watcher::ProcessInstanceState::Running { birth_id, .. } => birth_id, + _ => None, + } + } +} + +fn request_matches( + request: &HandoffRequest, + ready: &WorkerReady, + options_path: &Path, + now: u64, +) -> bool { + request.protocol == HANDOFF_PROTOCOL + && ready.protocol == HANDOFF_PROTOCOL + && request.worker == *ready + && options_path.canonicalize().ok().as_ref() == Some(&ready.options_path) + && request.requester_pid != 0 + && request.requester_pid != ready.pid + && !request.requester_birth.is_empty() + && !request.token.is_empty() + && !request.successor_token.is_empty() + && request.successor_token != ready.token + && request.issued_at_ms <= now + && now < request.expires_at_ms + && request.expires_at_ms.saturating_sub(request.issued_at_ms) + <= HANDOFF_LIFETIME.as_millis() as u64 +} + +fn request_active(root: &Path, request: &HandoffRequest, ready: &WorkerReady) -> bool { + if !request_matches(request, ready, &root.join(OPTIONS), current_timestamp_ms()) + || read_json::(&root.join(REQUEST)).as_ref() != Some(request) + || process_birth(request.requester_pid).as_ref() != Some(&request.requester_birth) + { + return false; + } + // 请求必须仍由发起 manager 持锁；崩溃会释放锁，即使磁盘请求尚未来得及撤销。 + let Ok(lock) = open_resume_lock(root) else { + return false; + }; + matches!(lock.try_lock(), Err(std::fs::TryLockError::WouldBlock)) +} + +struct PendingHandoff { + root: PathBuf, + request: HandoffRequest, +} + +impl PendingHandoff { + fn revoke(&self) -> anyhow::Result<()> { + // 只撤销本次请求，不覆盖另一代交接；用 tombstone 避免按路径删除的竞争。 + if read_json::(&self.root.join(REQUEST)).as_ref() == Some(&self.request) { + write_json(&self.root.join(REQUEST), &serde_json::Value::Null)?; + } + Ok(()) + } +} + +impl Drop for PendingHandoff { + fn drop(&mut self) { + if let Err(error) = self.revoke() { + let _ = codex_plus_core::diagnostic_log::append_diagnostic_log( + "manager.multi_worker_revoke_failed", + json!({"message": format!("{error:#}")}), + ); + } + } +} + +fn new_token() -> String { + static SERIAL: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0); + format!( + "{}-{}-{}", + std::process::id(), + SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_nanos(), + SERIAL.fetch_add(1, Ordering::Relaxed), + ) +} + +fn write_json(path: &Path, value: &impl Serialize) -> anyhow::Result<()> { + if let Ok(metadata) = fs::symlink_metadata(path) { + ensure!(!metadata.file_type().is_symlink(), "续接文件不能是符号链接"); + } + codex_plus_core::settings::atomic_write(path, &serde_json::to_vec_pretty(value)?) +} + +pub(super) fn publish_worker_ready( + options_path: &Path, + options: &CodexMultiWorkerOptions, + identity: &CodexInstanceIdentity, + binary: BinaryStamp, +) -> anyhow::Result { + let ready = worker_ready(options_path, options, identity, binary)?; + validate_candidate(&ready, false)?; + let root = ready.options_path.parent().context("缺少 worker 目录")?; + write_json(&root.join(READY), &ready)?; + Ok(ready) +} + +fn worker_ready( + options_path: &Path, + options: &CodexMultiWorkerOptions, + identity: &CodexInstanceIdentity, + binary: BinaryStamp, +) -> anyhow::Result { + ensure!( + identity.instance_id == options.instance_id && identity.generation == options.generation, + "worker 就绪身份不匹配" + ); + let options_path = options_path.canonicalize()?; + let ready = WorkerReady { + protocol: HANDOFF_PROTOCOL, + instance_id: options.instance_id.clone(), + generation: options.generation, + pid: std::process::id(), + #[cfg(windows)] + worker_birth_id: codex_plus_core::windows_process_birth_id(std::process::id()), + #[cfg(not(windows))] + worker_birth_id: None, + token: options.resume_token.clone().unwrap_or_else(new_token), + options_path: options_path.clone(), + identity: identity.clone(), + binary, + }; + ensure!(!ready.token.is_empty(), "worker token 为空"); + Ok(ready) +} + +fn validate_candidate(ready: &WorkerReady, register: bool) -> anyhow::Result<()> { + let Some(request) = std::env::var_os(HANDOFF_ENV) else { + return Ok(()); + }; + let request: HandoffRequest = + serde_json::from_str(request.to_str().context("候选 worker 交接标记无效")?)?; + validate_candidate_request(&request, ready, register) +} + +fn validate_candidate_request( + request: &HandoffRequest, + ready: &WorkerReady, + register: bool, +) -> anyhow::Result<()> { + ensure!( + successor_matches(request, ready), + "候选 worker 与交接身份不一致" + ); + let root = ready.options_path.parent().context("缺少 worker 目录")?; + // 提交绑定完整 PID/birth，不能用过期请求或另一候选的 commit 放行。 + if read_json::(&root.join(COMMIT)) + .is_some_and(|commit| commit.request == *request && commit.successor == *ready) + { + return Ok(()); + } + if register { + ensure!( + request_active(root, request, &request.worker), + "候选 worker 启动前交接已失效" + ); + // 先注册再复验：manager 在 spawn/manifest 之间退出时，回滚仍能发现候选。 + write_json(&root.join(CANDIDATE), ready)?; + } + ensure!( + request_active(root, request, &request.worker), + "交接已取消或发起 manager 已退出，候选 worker 不得启动服务或发布就绪" + ); + Ok(()) +} + +pub(super) fn handoff_requested( + options_path: &Path, + ready: &WorkerReady, + checkpoint: impl FnOnce() -> anyhow::Result<()>, +) -> bool { + if ready.pid != std::process::id() { + return false; + } + let Some(root) = options_path.parent() else { + return false; + }; + let release = read_json::(&root.join(RELEASE)) + .filter(|release| release.request.worker == *ready); + if let Some(release) = release { + let committed = committed_successor(root, &release.request); + let active = !committed && request_active(root, &release.request, ready); + match finish_handoff(committed, active, || { + restore_released_worker(root, ready, &release) + }) { + Ok(true) => return true, + Err(error) => { + // 保留 release 回执，下轮继续恢复；不能丢掉旧 worker 的恢复责任。 + let _ = codex_plus_core::diagnostic_log::append_diagnostic_log( + "manager.multi_worker_rollback_pending", + json!({"message": format!("{error:#}")}), + ); + } + Ok(false) => {} + } + if active && !release.released { + release_worker_services(root, ready, release, checkpoint); + } + return false; + } + let Some(request) = read_json::(&root.join(REQUEST)) + .filter(|request| request_active(root, request, ready)) + else { + return false; + }; + let Some(options) = read_json::(options_path) else { + return false; + }; + let release = HandoffRelease { + request, + options, + released: false, + }; + // 先持久化恢复材料，stop 或回执写入失败后仍能在下一轮恢复。 + if write_json(&root.join(RELEASE), &release).is_ok() { + release_worker_services(root, ready, release, checkpoint); + } + false +} + +fn finish_handoff( + committed: bool, + active: bool, + restore: impl FnOnce() -> anyhow::Result<()>, +) -> anyhow::Result { + if committed { + return Ok(true); + } + if !active { + restore()?; + } + Ok(false) +} + +fn release_worker_services( + root: &Path, + ready: &WorkerReady, + mut release: HandoffRelease, + checkpoint: impl FnOnce() -> anyhow::Result<()>, +) { + let result = (|| -> anyhow::Result<()> { + ensure!( + request_active(root, &release.request, ready), + "交接请求已失效" + ); + checkpoint()?; + ensure!( + request_active(root, &release.request, ready), + "保存 checkpoint 期间交接已取消" + ); + codex_plus_launcher::stop()?; + ensure!( + request_active(root, &release.request, ready), + "服务释放期间交接已取消" + ); + release.released = true; + write_json(&root.join(RELEASE), &release) + })(); + if let Err(error) = result { + let _ = codex_plus_core::diagnostic_log::append_diagnostic_log( + "manager.multi_worker_release_failed", + json!({"message": format!("{error:#}")}), + ); + } +} + +fn successor_matches(request: &HandoffRequest, successor: &WorkerReady) -> bool { + successor.protocol == HANDOFF_PROTOCOL + && successor.instance_id == request.worker.instance_id + && successor.generation == request.worker.generation + && successor.identity == request.worker.identity + && successor.options_path == request.worker.options_path + && successor.pid != 0 + && successor.pid != request.worker.pid + && successor.token == request.successor_token + && successor.binary == request.successor_binary +} + +fn committed_successor(root: &Path, request: &HandoffRequest) -> bool { + let Some(commit) = read_json::(&root.join(COMMIT)).filter(|commit| { + commit.request == *request && successor_matches(request, &commit.successor) + }) else { + return false; + }; + let Ok(instance) = load_multi_instance_manifest(root) else { + return false; + }; + ready_matches(&commit.successor, &instance, &root.join(OPTIONS)) + && worker_present(&instance, &root.join(OPTIONS)).unwrap_or(false) +} + +fn restore_released_worker( + root: &Path, + ready: &WorkerReady, + release: &HandoffRelease, +) -> anyhow::Result<()> { + restore_released_worker_with(root, ready, release, terminate_multi_worker, |options| { + let receipt = + codex_plus_launcher::resume_with_receipt(codex_plus_launcher::LaunchOptions { + app_dir: Some(PathBuf::from(&options.app_path)), + debug_port: ready.identity.debug_port, + helper_port: ready.identity.helper_port, + status_store: StatusStore::new( + Path::new(&options.state_dir).join("latest-status.json"), + ), + state_dir: Some(PathBuf::from(&options.state_dir)), + codex_home: std::env::var_os("CODEX_HOME").map(PathBuf::from), + sqlite_home: std::env::var_os("CODEX_SQLITE_HOME").map(PathBuf::from), + })? + .context("原桌面暂不可续接，保留旧 worker 稍后重试")?; + confirm_resumed_identity(&ready.identity, &receipt) + }) +} + +fn restore_released_worker_with( + root: &Path, + ready: &WorkerReady, + release: &HandoffRelease, + stop_candidate: impl FnOnce(u32, &Path, Option) -> anyhow::Result<()>, + resume: impl FnOnce(&CodexMultiWorkerOptions) -> anyhow::Result<()>, +) -> anyhow::Result<()> { + // manager 先回收自己持有 Child 的候选 worker 并退出事务；旧 worker 随后恢复。 + let lock = open_resume_lock(root)?; + lock.try_lock().context("等待 manager 结束交接事务")?; + let mut instance = load_multi_instance_manifest(root)?; + ensure!( + instance.identity.as_ref() == Some(&ready.identity) + && instance.instance_id == ready.instance_id + && instance.generation == ready.generation, + "恢复时桌面身份已变化，未触动其他代次" + ); + let registered = read_json::(&root.join(CANDIDATE)) + .filter(|candidate| successor_matches(&release.request, candidate)); + let candidate = registered + .as_ref() + .map(|candidate| (candidate.pid, candidate.worker_birth_id)) + .or_else(|| { + instance + .pid + .filter(|pid| *pid != ready.pid) + .map(|pid| (pid, instance.worker_birth_id)) + }); + if let Some((pid, birth)) = candidate { + ensure!( + instance + .pid + .is_none_or(|current| current == ready.pid || current == pid), + "清单指向其他 worker，拒绝停止" + ); + let options: CodexMultiWorkerOptions = + read_json(&root.join(OPTIONS)).context("缺少候选 worker 参数")?; + ensure!( + options.resume_token.as_ref() == Some(&release.request.successor_token) + && options.resume_identity.as_ref() == Some(&ready.identity), + "候选 worker 不属于本次交接，拒绝停止" + ); + // 发起 manager 崩溃时 Child 不会自动退出；仅清理本次完整身份的候选 worker。 + stop_candidate(pid, &root.join(OPTIONS), birth)?; + write_json(&root.join(CANDIDATE), &serde_json::Value::Null)?; + } + write_json(&root.join(OPTIONS), &release.options)?; + instance.pid = Some(ready.pid); + instance.worker_birth_id = ready.worker_birth_id; + save_multi_instance_manifest(root, &instance)?; + resume(&release.options)?; + write_json(&root.join(READY), ready)?; + write_json(&root.join(RELEASE), &serde_json::Value::Null)?; + Ok(()) +} + +fn desktop_matches(identity: &CodexInstanceIdentity, deadline: Instant) -> anyhow::Result<()> { + // 外部 CDP 验证只读；调用方即使超时也不会留下延迟执行的 stop/spawn。 + let identity = identity.clone(); + let (sender, receiver) = mpsc::sync_channel(1); + std::thread::spawn(move || { + let result = tauri::async_runtime::block_on( + codex_plus_core::multi_worker::codex_instance_identity_matches(&identity), + ); + let _ = sender.send(result); + }); + let remaining = deadline.saturating_duration_since(Instant::now()); + ensure!(!remaining.is_zero(), "验证原 Codex 身份超时，可重试"); + ensure!( + receiver + .recv_timeout(remaining) + .context("验证原 Codex 身份超时，可重试")??, + "原 Codex 身份已变化或连接暂不可用，已保留桌面，可重试" + ); + Ok(()) +} + +pub(super) fn validate_worker_resume(options: &CodexMultiWorkerOptions) -> anyhow::Result<()> { + let candidate_request = std::env::var_os(HANDOFF_ENV) + .map(|value| -> anyhow::Result { + Ok(serde_json::from_str( + value.to_str().context("候选 worker 交接标记无效")?, + )?) + }) + .transpose()?; + let candidate = candidate_request + .as_ref() + .map(|request| { + worker_ready( + &request.worker.options_path, + options, + options + .resume_identity + .as_ref() + .context("候选缺少桌面身份")?, + BinaryStamp::current()?, + ) + }) + .transpose()?; + if let Some(ready) = &candidate { + validate_candidate(ready, true)?; + } + if let Some(identity) = &options.resume_identity { + ensure!( + options + .resume_token + .as_ref() + .is_some_and(|token| !token.is_empty()) + && identity.instance_id == options.instance_id + && identity.generation == options.generation + && identity.debug_port == options.debug_port + && identity.helper_port == options.helper_port + && Path::new(&options.user_data_dir).canonicalize()? == identity.user_data_dir + && desktop_executable(Path::new(&options.app_path)) == identity.executable, + "续接参数与原 Codex 身份不一致" + ); + desktop_matches(identity, Instant::now() + STAGE_TIMEOUT)?; + } + if let Some(ready) = &candidate { + validate_candidate(ready, false)?; + } + Ok(()) +} + +pub(super) fn confirm_resumed_identity( + identity: &CodexInstanceIdentity, + receipt: &codex_plus_launcher::LaunchReceipt, +) -> anyhow::Result<()> { + ensure!( + receipt.debug_port == identity.debug_port + && receipt.helper_port == identity.helper_port + && desktop_executable(&receipt.app_dir) == identity.executable, + "续接端口或应用发生变化，拒绝接管" + ); + desktop_matches(identity, Instant::now() + STAGE_TIMEOUT) +} + +fn desktop_executable(app_dir: &Path) -> PathBuf { + let executable = codex_plus_core::app_paths::build_codex_executable(app_dir); + executable.canonicalize().unwrap_or(executable) +} + +fn scoped_options( + root: &Path, + parent: &Path, + instance: &CodexMultiInstance, + legacy_parent: Option<&Path>, +) -> anyhow::Result { + let root = root.canonicalize()?; + let parent = parent.canonicalize()?; + ensure!( + root.parent() == Some(parent.as_path()) + && root.file_name().and_then(|name| name.to_str()) + == Some(multi_instance_slug(&instance.instance_id).as_str()), + "多开目录归属不匹配" + ); + let identity = instance + .identity + .as_ref() + .context("缺少原 Codex 完整身份")?; + let options_path = root.join(OPTIONS); + ensure!( + options_path.canonicalize()? == options_path, + "worker options 路径发生重定向" + ); + let mut options: CodexMultiWorkerOptions = serde_json::from_slice(&fs::read(options_path)?)?; + let state = root + .join("worker-home/.codex-session-delete") + .canonicalize()?; + let user_data = identity.user_data_dir.canonicalize()?; + let legacy_user_data = legacy_parent + .map(|parent| parent.join(multi_instance_slug(&instance.instance_id))) + .and_then(|path| path.canonicalize().ok()); + let configured_user_data = std::env::var_os("CODEX_ELECTRON_USER_DATA_PATH") + .filter(|value| !value.is_empty()) + .map(PathBuf::from) + .map(|parent| { + parent + .join("multi") + .join(multi_instance_slug(&instance.instance_id)) + }) + .and_then(|path| path.canonicalize().ok()); + ensure!( + multi_manifest_identity_matches(instance, identity) + && options.instance_id == instance.instance_id + && options.generation == instance.generation + && instance.generation != 0 + && state.starts_with(&root) + && Path::new(&options.state_dir).canonicalize()? == state + && Path::new(&options.user_data_dir).canonicalize()? == user_data + && (user_data.starts_with(&root) + || legacy_user_data.as_ref() == Some(&user_data) + || configured_user_data.as_ref() == Some(&user_data)) + && instance + .codex_home + .as_ref() + .map(Path::new) + .map(Path::canonicalize) + .transpose()? + == Some(root.join("codex-home").canonicalize()?) + && instance + .sqlite_home + .as_ref() + .map(Path::new) + .map(Path::canonicalize) + .transpose()? + == Some(root.join("sqlite-home").canonicalize()?), + "manifest、窗口或隔离路径不完整，未触动旧 worker" + ); + ensure!( + root.join("codex-home").canonicalize()?.starts_with(&root) + && root.join("sqlite-home").canonicalize()?.starts_with(&root) + && root.join("worker-home").canonicalize()?.starts_with(&root), + "多开隔离目录发生重定向" + ); + // 不读取供应商内容或主配置，更不从主实例重建 worker 的 provider。 + ensure!( + state.join("settings.json").is_file(), + "缺少原 worker 设置，已保留旧 worker" + ); + let _settings = fs::File::open(state.join("settings.json"))?; + options.app_path = instance.app_path.clone().context("缺少原应用目录")?; + // WindowsApps 映像可能拒绝 metadata；实际 exe 归属由桌面身份核验。 + let _app_entries = fs::read_dir(&options.app_path).context("原 Codex 应用目录不可访问")?; + options.debug_port = identity.debug_port; + options.helper_port = identity.helper_port; + options.resume_identity = Some(identity.clone()); + options.resume_token = Some(new_token()); + Ok(options) +} + +fn legacy_parent(parent: &Path) -> Option { + let dirs = directories::BaseDirs::new()?; + let production = dirs.home_dir().join(".codex-session-delete/codex-multi"); + // 隔离 manager 不能以旧布局为由借用生产窗口。 + (parent.canonicalize().ok()? == production.canonicalize().ok()?) + .then(|| dirs.data_local_dir().join("CodexPlusPlus/multi")) +} + +fn ready_matches(ready: &WorkerReady, instance: &CodexMultiInstance, options_path: &Path) -> bool { + matches!(ready.protocol, 1 | HANDOFF_PROTOCOL) + && ready.pid != 0 + && Some(ready.pid) == instance.pid + && ready.worker_birth_id == instance.worker_birth_id + && ready.instance_id == instance.instance_id + && ready.generation == instance.generation + && !ready.token.is_empty() + && instance.identity.as_ref() == Some(&ready.identity) + && options_path.canonicalize().ok().as_ref() == Some(&ready.options_path) +} + +fn read_ready(root: &Path) -> anyhow::Result> { + match fs::read(root.join(READY)) { + Ok(bytes) => Ok(Some( + serde_json::from_slice(&bytes).context("worker ready 协议损坏")?, + )), + Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(None), + Err(error) => Err(error.into()), + } +} + +fn worker_present(instance: &CodexMultiInstance, options_path: &Path) -> anyhow::Result { + let Some(pid) = instance.pid else { + return Ok(false); + }; + #[cfg(windows)] + { + codex_plus_core::multi_worker::verified_worker_is_running( + pid, + options_path, + instance.worker_birth_id, + ) + } + #[cfg(not(windows))] + { + let _ = options_path; + match codex_plus_core::watcher::inspect_process_instance(pid) { + codex_plus_core::watcher::ProcessInstanceState::NotRunning => Ok(false), + _ => anyhow::bail!("当前系统缺少精确 worker 核验接口，已保留旧 worker"), + } + } +} + +fn verify_legacy_services( + instance: &CodexMultiInstance, + options_path: &Path, +) -> anyhow::Result<()> { + verify_legacy_services_with( + || worker_present(instance, options_path), + || { + tauri::async_runtime::block_on(async { + let client = codex_plus_core::http_client::local_helper_client()?; + let response = client + .get(format!( + "http://127.0.0.1:{}/backend/status", + instance.helper_port + )) + .send() + .await? + .error_for_status()?; + let status: serde_json::Value = response.json().await?; + ensure!( + status["status"] == "ok" && status["transport"] == "http-helper", + "旧 worker 的 helper 尚未确认健康，已保留并延期交接" + ); + Ok::<_, anyhow::Error>(()) + }) + }, + ) +} + +fn verify_legacy_services_with( + mut present: impl FnMut() -> anyhow::Result, + healthy: impl FnOnce() -> anyhow::Result<()>, +) -> anyhow::Result<()> { + ensure!(present()?, "旧 worker 已退出"); + healthy()?; + ensure!(present()?, "健康检查期间旧 worker 已变化"); + Ok(()) +} + +fn generation_may_be_alive( + expected_birth: Option, + state: codex_plus_core::watcher::ProcessInstanceState, +) -> bool { + use codex_plus_core::watcher::ProcessInstanceState; + match state { + ProcessInstanceState::NotRunning => false, + ProcessInstanceState::Running { + birth_id: Some(birth), + .. + } => !expected_birth + .zip(birth.parse::().ok()) + .is_some_and(|(a, b)| a != b), + _ => true, + } +} + +fn worker_could_hold_handles(root: &Path) -> bool { + #[derive(Deserialize)] + #[serde(rename_all = "camelCase")] + struct WorkerHint { + #[serde(default)] + pid: Option, + #[serde(default)] + worker_birth_id: Option, + } + let Ok(bytes) = fs::read(multi_instance_manifest_path(root)) else { + return false; + }; + let Ok(hint) = serde_json::from_slice::(&bytes) else { + return false; + }; + hint.pid.is_some_and(|pid| { + generation_may_be_alive( + hint.worker_birth_id, + codex_plus_core::watcher::inspect_process_instance(pid), + ) + }) +} + +fn check_active(cancel: &AtomicBool, deadline: Instant) -> anyhow::Result<()> { + ensure!( + !crate::app_is_exiting() && !cancel.load(Ordering::Acquire), + "管理器恢复已取消，原 Codex 保持运行，可重试" + ); + ensure!( + Instant::now() < deadline, + "worker 交接阶段超过 90 秒，可重试" + ); + Ok(()) +} + +fn same_generation(current: &CodexMultiInstance, original: &CodexMultiInstance) -> bool { + current.instance_id == original.instance_id + && current.generation == original.generation + && current.identity == original.identity +} + +struct PendingWorker { + child: Child, + armed: bool, +} + +impl PendingWorker { + fn reclaim(&mut self) -> anyhow::Result<()> { + if self.child.try_wait()?.is_some() { + self.armed = false; + return Ok(()); + } + // 持有本次 spawn 的 Child 句柄；绝不按 PID 或程序名结束其他进程。 + self.child.kill().context("回收本次未就绪 worker 失败")?; + let deadline = Instant::now() + Duration::from_secs(5); + while Instant::now() < deadline { + if self.child.try_wait()?.is_some() { + self.armed = false; + return Ok(()); + } + std::thread::sleep(POLL); + } + anyhow::bail!("本次未就绪 worker 尚未确认退出") + } +} + +impl Drop for PendingWorker { + fn drop(&mut self) { + if self.armed { + let _ = self.reclaim(); + } + } +} + +fn spawn_resume( + root: &Path, + binary: &BinaryStamp, + options: &CodexMultiWorkerOptions, + instance: &CodexMultiInstance, + handoff: Option<&HandoffRequest>, +) -> anyhow::Result { + ensure!( + BinaryStamp::at(&binary.executable)? == *binary, + "新版二进制预检后发生变化，可重试" + ); + let mut command = Command::new(&binary.executable); + command.env_remove(HANDOFF_ENV); + if let Some(request) = handoff { + command.env(HANDOFF_ENV, serde_json::to_string(request)?); + } + command + .arg("--codex-plus-multi-worker") + .arg(root.join(OPTIONS)) + .env("HOME", root.join("worker-home")) + .env("CODEX_HOME", root.join("codex-home")) + .env("CODEX_SQLITE_HOME", root.join("sqlite-home")) + .env("CODEX_ELECTRON_USER_DATA_PATH", &options.user_data_dir) + .env("CODEX_PLUS_MULTI_INSTANCE", "1") + .env("CODEX_PLUS_INSTANCE_ID", &instance.instance_id) + .env("CODEX_PLUS_INSTANCE_LABEL", &instance.profile_name) + .env( + "CODEX_PLUS_LAUNCHER_GUARD_PORT", + select_multi_instance_port(0, &format!("{}-guard", instance.instance_id)).to_string(), + ) + .env( + "CODEX_PLUS_PROTOCOL_PROXY_PORT", + options.helper_port.to_string(), + ) + .env(codex_plus_core::paths::STATE_DIR_ENV, &options.state_dir) + .env( + "XDG_CONFIG_HOME", + directories::BaseDirs::new() + .context("无法定位共享脚本目录")? + .home_dir() + .join(".config"), + ) + .stdin(Stdio::null()) + .stdout(Stdio::null()) + .stderr(Stdio::null()); + #[cfg(windows)] + { + use std::os::windows::process::CommandExt; + command.creation_flags(codex_plus_core::windows_create_no_window()); + } + Ok(PendingWorker { + child: command.spawn()?, + armed: true, + }) +} + +fn resume_one(root: &Path, parent: &Path, cancel: &AtomicBool) -> anyhow::Result<&'static str> { + let root = root.canonicalize()?; + ensure!( + root.parent() == Some(parent.canonicalize()?.as_path()), + "实例目录越界" + ); + let lock = open_resume_lock(&root)?; + lock.try_lock() + .context("另一个管理器正在交接此 worker，可重试")?; + let mut instance = load_multi_instance_manifest(&root)?; + let options = scoped_options(&root, parent, &instance, legacy_parent(parent).as_deref())?; + let binary = BinaryStamp::current()?; + let options_path = root.join(OPTIONS); + let identity = options.resume_identity.as_ref().context("缺少续接身份")?; + let preflight_deadline = Instant::now() + STAGE_TIMEOUT; + check_active(cancel, preflight_deadline)?; + desktop_matches(identity, preflight_deadline)?; + let mut handoff = None; + if worker_present(&instance, &options_path)? { + if let Some(ready) = read_ready(&root)? { + ensure!( + ready_matches(&ready, &instance, &options_path), + "旧 worker 协议身份不一致，已保留原 worker，可重试" + ); + if ready.protocol == 1 { + verify_legacy_services(&instance, &options_path) + .context("legacy worker 健康状态不明确，已保留并延期交接")?; + desktop_matches(identity, preflight_deadline)?; + check_active(cancel, preflight_deadline)?; + return Ok("reused_legacy"); + } + if ready.binary == binary + && !read_json::(&root.join(RELEASE)) + .is_some_and(|release| release.request.worker == ready) + { + return Ok("reused"); + } + ensure!( + BinaryStamp::current()? == binary, + "新版二进制已变化，保留旧 worker" + ); + check_active(cancel, preflight_deadline)?; + let issued_at_ms = current_timestamp_ms(); + let request = HandoffRequest { + protocol: HANDOFF_PROTOCOL, + token: new_token(), + worker: ready, + requester_pid: std::process::id(), + requester_birth: process_birth(std::process::id()) + .context("无法核验发起 manager 的创建身份，未请求交接")?, + issued_at_ms, + expires_at_ms: issued_at_ms.saturating_add(HANDOFF_LIFETIME.as_millis() as u64), + successor_token: options.resume_token.clone().context("缺少续接 token")?, + successor_binary: binary.clone(), + }; + handoff = Some(PendingHandoff { + root: root.clone(), + request, + }); + let request = &handoff.as_ref().unwrap().request; + write_json(&root.join(REQUEST), request)?; + let release_deadline = Instant::now() + STAGE_TIMEOUT; + loop { + check_active(cancel, release_deadline)?; + ensure!( + worker_present(&instance, &options_path)?, + "旧 worker 在交接提交前退出，未继续启动" + ); + if read_json::(&root.join(RELEASE)) + .is_some_and(|release| release.request == *request && release.released) + { + break; + } + std::thread::sleep(POLL); + } + } else { + verify_legacy_services(&instance, &options_path) + .context("无 ready 的 legacy worker 尚未确认健康，已保留并延期交接")?; + desktop_matches(identity, preflight_deadline)?; + check_active(cancel, preflight_deadline)?; + return Ok("reused_legacy"); + } + } + let ready_deadline = Instant::now() + STAGE_TIMEOUT; + check_active(cancel, ready_deadline)?; + desktop_matches(identity, ready_deadline)?; + { + let _guard = multi_instance_mutation_lock() + .lock() + .unwrap_or_else(|e| e.into_inner()); + let current = load_multi_instance_manifest(&root)?; + ensure!( + same_generation(¤t, &instance) && current.pid == instance.pid, + "交接期间实例代次或 worker 发生变化，可重试" + ); + write_json(&options_path, &options)?; + } + check_active(cancel, ready_deadline)?; + let mut pending = spawn_resume( + &root, + &binary, + &options, + &instance, + handoff.as_ref().map(|handoff| &handoff.request), + )?; + let new_pid = pending.child.id(); + let new_birth = codex_plus_core::multi_worker::worker_birth_id(&pending.child); + let result = (|| { + { + let _guard = multi_instance_mutation_lock() + .lock() + .unwrap_or_else(|e| e.into_inner()); + let current = load_multi_instance_manifest(&root)?; + ensure!( + same_generation(¤t, &instance) && current.pid == instance.pid, + "启动续接 worker 时清单已变化" + ); + instance.pid = Some(new_pid); + instance.worker_birth_id = new_birth; + save_multi_instance_manifest(&root, &instance)?; + } + loop { + check_active(cancel, ready_deadline)?; + ensure!( + pending.child.try_wait()?.is_none(), + "新版 worker 提前退出，原桌面保持运行，可重试" + ); + if let Ok(Some(ready)) = read_ready(&root) { + if ready_matches(&ready, &instance, &options_path) + && options.resume_token.as_ref() == Some(&ready.token) + && ready.binary == binary + { + desktop_matches(identity, ready_deadline)?; + let _guard = multi_instance_mutation_lock() + .lock() + .unwrap_or_else(|e| e.into_inner()); + let current = load_multi_instance_manifest(&root)?; + ensure!( + same_generation(¤t, &instance) && current.pid == Some(new_pid), + "续接期间清单已变化，已回收本次 worker" + ); + check_active(cancel, ready_deadline)?; + if let Some(handoff) = &handoff { + ensure!( + request_active(&root, &handoff.request, &handoff.request.worker), + "交接请求已过期或撤销，未提交候选 worker" + ); + write_json( + &root.join(COMMIT), + &HandoffCommit { + request: handoff.request.clone(), + successor: ready, + }, + )?; + } + return Ok("resumed"); + } + } + std::thread::sleep(POLL); + } + })(); + match result { + Ok(outcome) => { + pending.armed = false; + Ok(outcome) + } + Err(error) => { + let cleanup = pending.reclaim(); + if cleanup.is_ok() { + let _guard = multi_instance_mutation_lock() + .lock() + .unwrap_or_else(|e| e.into_inner()); + if let Ok(mut current) = load_multi_instance_manifest(&root) { + if same_generation(¤t, &instance) && current.pid == Some(new_pid) { + current.pid = None; + current.worker_birth_id = None; + // 保留原 desktop identity 与隔离配置，重开 manager 即可重试 attach。 + save_multi_instance_manifest(&root, ¤t)?; + } + } + } + match cleanup { + Ok(()) => Err(error), + Err(cleanup) => Err(anyhow::anyhow!( + "{error:#}；未就绪 worker 回收失败：{cleanup:#}" + )), + } + } + } +} + +fn worker_resume_roots(parent: &Path, windows: bool) -> Vec { + if !windows { + return Vec::new(); + } + fs::read_dir(parent) + .into_iter() + .flatten() + .flatten() + .filter(|entry| entry.file_type().is_ok_and(|kind| kind.is_dir())) + .map(|entry| entry.path()) + .filter(|root| root.join("instance.json").is_file()) + .collect() +} + +/// Windows worker 并行交接后恢复主服务；其他平台保留原 worker，不设全局门槛。 +pub(crate) fn resume_on_manager_start(app: tauri::AppHandle, start_if_missing: bool) { + tauri::async_runtime::spawn_blocking(move || { + let parent = codex_plus_core::paths::default_app_state_dir().join("codex-multi"); + let cancel = Arc::new(AtomicBool::new(false)); + let (sender, receiver) = mpsc::channel(); + let mut count = 0; + for root in worker_resume_roots(&parent, cfg!(windows)) { + count += 1; + let sender = sender.clone(); + let parent = parent.clone(); + let cancel = cancel.clone(); + std::thread::spawn(move || { + let result = + (|| { + let instance = load_multi_instance_manifest(&root)?; + // 确认桌面退出才跳过；缺身份/临时 CDP 失联必须报告而非误认停止。 + if instance.identity.as_ref().is_some_and( + codex_plus_core::multi_worker::codex_instance_process_exited, + ) || (instance.identity.is_none() && !worker_could_hold_handles(&root)) + { + return Ok("not_running"); + } + resume_one(&root, &parent, &cancel) + })(); + let outcome = json!({ + "status": result.as_ref().copied().unwrap_or("deferred"), + "retryable": result.is_err(), + "message": result.as_ref().err().map(|error| format!("{error:#}")), + "finishedAtMs": current_timestamp_ms(), + }); + let _ = write_json(&root.join(OUTCOME), &outcome); + let _ = codex_plus_core::diagnostic_log::append_diagnostic_log( + "manager.multi_worker_resume", + &outcome, + ); + // 已退出的坏历史不阻断主恢复；仍活着的失败 worker 可能遗留监听。 + let _ = sender.send(result.is_ok() || !worker_could_hold_handles(&root)); + }); + } + drop(sender); + // 三个阶段各 90s；实例并行，不按实例数量累加等待。 + let deadline = Instant::now() + STAGE_TIMEOUT * 3 + Duration::from_secs(5); + let mut complete = true; + for _ in 0..count { + match receiver.recv_timeout(deadline.saturating_duration_since(Instant::now())) { + Ok(ok) => complete &= ok, + Err(_) => { + complete = false; + cancel.store(true, Ordering::Release); + break; + } + } + } + if !complete { + let _ = codex_plus_core::diagnostic_log::append_diagnostic_log( + "manager.background_restore_deferred", + json!({"message": "多开交接未完成；保留原桌面，主 helper/Gateway 恢复已延期，可重开管理器重试"}), + ); + } else if !crate::app_is_exiting() { + resume_existing_codex_background(app.clone(), start_if_missing); + crate::gateway_commands::start_from_saved_state(); + crate::gateway_commands::start_model_watch(); + crate::auto_takeover::start_supervisor(); + } + let _ = app.emit("runtime-health-changed", ()); + }); +} + +#[cfg(test)] +mod tests { + use super::*; + + fn identity(root: &Path) -> CodexInstanceIdentity { + CodexInstanceIdentity { + instance_id: "test".into(), + generation: 7, + debug_port: 49101, + helper_port: 49102, + user_data_dir: root.into(), + browser_id: "browser".into(), + process_id: 123, + process_birth_id: 456, + executable: root.join("Codex.exe"), + } + } + + #[test] + fn handoff_requires_every_ready_field_and_exact_options_path() { + let temp = tempfile::tempdir().unwrap(); + let options = temp.path().join(OPTIONS); + fs::write(&options, b"{}").unwrap(); + let ready = WorkerReady { + protocol: HANDOFF_PROTOCOL, + instance_id: "test".into(), + generation: 7, + pid: 100, + worker_birth_id: Some(1000), + token: "fresh".into(), + options_path: options.canonicalize().unwrap(), + identity: identity(temp.path()), + binary: BinaryStamp { + executable: temp.path().join("manager.exe"), + size: 100, + modified_ns: "2".into(), + version: "test".into(), + }, + }; + assert!(!handoff_requested(&options, &ready, || panic!( + "no request" + ))); + let request = request_fixture(&ready); + assert!(request_matches(&request, &ready, &options, 1000)); + let value = serde_json::to_value(&ready).unwrap(); + for field in [ + "pid", + "generation", + "token", + "optionsPath", + "identity", + "workerBirthId", + "binary", + ] { + let mut invalid = value.clone(); + invalid[field] = serde_json::Value::Null; + let mut invalid_request = serde_json::to_value(&request).unwrap(); + invalid_request["worker"] = invalid; + write_json(&temp.path().join(REQUEST), &invalid_request).unwrap(); + assert!( + read_json::(&temp.path().join(REQUEST)) + .is_none_or(|request| !request_matches(&request, &ready, &options, 1000)), + "{field}" + ); + } + // v1 是直接比较完整 WorkerReady；v2 使用不同 envelope，旧版无法误消费。 + write_json(&temp.path().join(REQUEST), &ready).unwrap(); + assert!(read_json::(&temp.path().join(REQUEST)).is_none()); + assert!( + serde_json::from_value::(serde_json::to_value(&request).unwrap()).is_err() + ); + let mut wrong = ready.clone(); + wrong.token = "stale".into(); + assert!(!request_matches(&request, &wrong, &options, 1000)); + assert!(!request_matches( + &request, + &ready, + &temp.path().join("other"), + 1000 + )); + } + + fn request_fixture(ready: &WorkerReady) -> HandoffRequest { + HandoffRequest { + protocol: HANDOFF_PROTOCOL, + token: "request".into(), + worker: ready.clone(), + requester_pid: 999, + requester_birth: "9999".into(), + issued_at_ms: 1000, + expires_at_ms: 1000 + HANDOFF_LIFETIME.as_millis() as u64, + successor_token: "successor".into(), + successor_binary: ready.binary.clone(), + } + } + + fn ready_fixture(root: &Path) -> WorkerReady { + let options_path = root.join(OPTIONS); + fs::write(&options_path, b"{}").unwrap(); + WorkerReady { + protocol: HANDOFF_PROTOCOL, + instance_id: "test".into(), + generation: 7, + pid: 100, + worker_birth_id: Some(1000), + token: "original".into(), + options_path: options_path.canonicalize().unwrap(), + identity: identity(root), + binary: BinaryStamp { + executable: root.join("manager.exe"), + size: 100, + modified_ns: "2".into(), + version: "test".into(), + }, + } + } + + #[test] + fn expired_future_overlong_and_legacy_requests_never_release_services() { + let temp = tempfile::tempdir().unwrap(); + let ready = ready_fixture(temp.path()); + let request = request_fixture(&ready); + assert!(!request_matches(&request, &ready, &ready.options_path, 999)); + assert!(!request_matches( + &request, + &ready, + &ready.options_path, + request.expires_at_ms + )); + let mut overlong = request.clone(); + overlong.expires_at_ms += 1; + assert!(!request_matches( + &overlong, + &ready, + &ready.options_path, + 1000 + )); + let mut legacy = ready.clone(); + legacy.protocol = 1; + let legacy_request = request_fixture(&legacy); + assert!(!request_matches( + &legacy_request, + &legacy, + &legacy.options_path, + 1000 + )); + write_json(&temp.path().join(REQUEST), &request).unwrap(); + assert!(!handoff_requested(&ready.options_path, &ready, || panic!( + "expired request" + ))); + assert!(!temp.path().join(RELEASE).exists()); + } + + #[test] + fn cancellation_revokes_only_its_request_and_process_or_lock_loss_invalidates_it() { + let temp = tempfile::tempdir().unwrap(); + let ready = ready_fixture(temp.path()); + let mut request = request_fixture(&ready); + request.requester_pid = std::process::id(); + request.requester_birth = process_birth(request.requester_pid).unwrap(); + request.issued_at_ms = current_timestamp_ms(); + request.expires_at_ms = request.issued_at_ms + HANDOFF_LIFETIME.as_millis() as u64; + let lock = open_resume_lock(temp.path()).unwrap(); + lock.try_lock().unwrap(); + write_json(&temp.path().join(REQUEST), &request).unwrap(); + assert!(request_active(temp.path(), &request, &ready)); + let pending = PendingHandoff { + root: temp.path().to_path_buf(), + request: request.clone(), + }; + drop(pending); + assert!(!request_active(temp.path(), &request, &ready)); + write_json(&temp.path().join(REQUEST), &request).unwrap(); + drop(lock); + assert!(!request_active(temp.path(), &request, &ready)); + let lock = open_resume_lock(temp.path()).unwrap(); + lock.try_lock().unwrap(); + let mut recycled = request.clone(); + recycled.requester_birth.push('0'); + write_json(&temp.path().join(REQUEST), &recycled).unwrap(); + assert!(!request_active(temp.path(), &recycled, &ready)); + let mut exited = request.clone(); + exited.requester_pid = u32::MAX; + write_json(&temp.path().join(REQUEST), &exited).unwrap(); + assert!(!request_active(temp.path(), &exited, &ready)); + let mut replacement = request.clone(); + replacement.token = "different-request".into(); + write_json(&temp.path().join(REQUEST), &replacement).unwrap(); + drop(PendingHandoff { + root: temp.path().to_path_buf(), + request, + }); + assert_eq!( + read_json::(&temp.path().join(REQUEST)), + Some(replacement) + ); + } + + #[test] + fn cancellation_after_release_reclaims_only_its_candidate_and_retries_restore() { + let temp = tempfile::tempdir().unwrap(); + let root = temp.path(); + let ready = ready_fixture(root); + let request = request_fixture(&ready); + let original: CodexMultiWorkerOptions = serde_json::from_value(json!({ + "instanceId": "test", "generation": 7, "appPath": root, + "userDataDir": root, "stateDir": root, + "debugPort": 49101, "helperPort": 49102, + })) + .unwrap(); + let mut candidate = original.clone(); + candidate.resume_token = Some(request.successor_token.clone()); + candidate.resume_identity = Some(ready.identity.clone()); + write_json(&root.join(OPTIONS), &candidate).unwrap(); + let instance: CodexMultiInstance = serde_json::from_value(json!({ + "instanceId": "test", "generation": 7, "identity": ready.identity, + "profileId": "test", "profileName": "test", "pid": ready.pid, + "workerBirthId": ready.worker_birth_id, "debugPort": 49101, "helperPort": 49102, + "status": "running", "running": true, "startedAtMs": 1, "isMain": false, + })) + .unwrap(); + save_multi_instance_manifest(root, &instance).unwrap(); + // manager 在 spawn 后、清单写入前退出：仅候选自己的持久化注册可发现它。 + let mut registered = ready.clone(); + registered.pid = 200; + registered.worker_birth_id = Some(2000); + registered.token = request.successor_token.clone(); + write_json(&root.join(CANDIDATE), ®istered).unwrap(); + let release = HandoffRelease { + request, + options: original, + released: true, + }; + write_json(&root.join(RELEASE), &release).unwrap(); + let lock = open_resume_lock(root).unwrap(); + lock.try_lock().unwrap(); + assert!( + restore_released_worker_with( + root, + &ready, + &release, + |_, _, _| panic!("manager still owns transaction"), + |_| panic!("manager still owns transaction") + ) + .is_err() + ); + drop(lock); + let stopped = std::cell::Cell::new(false); + let result = restore_released_worker_with( + root, + &ready, + &release, + |pid, path, birth| { + assert_eq!((pid, birth), (200, Some(2000))); + assert_eq!(path, root.join(OPTIONS)); + stopped.set(true); + Ok(()) + }, + |options| { + assert!(stopped.get()); + assert!(options.resume_token.is_none()); + assert_eq!(load_multi_instance_manifest(root)?.pid, Some(ready.pid)); + anyhow::bail!("temporary attach failure") + }, + ); + assert!(result.is_err()); + assert!(read_json::(&root.join(RELEASE)).is_some()); + restore_released_worker_with( + root, + &ready, + &release, + |_, _, _| panic!("candidate already reclaimed"), + |_| Ok(()), + ) + .unwrap(); + assert!(read_json::(&root.join(RELEASE)).is_none()); + assert_eq!(read_ready(root).unwrap(), Some(ready)); + } + + #[test] + fn candidate_registration_rejects_lost_requester_and_commit_survives_revoke() { + let temp = tempfile::tempdir().unwrap(); + let root = temp.path(); + let old = ready_fixture(root); + let mut request = request_fixture(&old); + request.requester_pid = std::process::id(); + request.requester_birth = process_birth(request.requester_pid).unwrap(); + request.issued_at_ms = current_timestamp_ms(); + request.expires_at_ms = request.issued_at_ms + HANDOFF_LIFETIME.as_millis() as u64; + let mut candidate = old.clone(); + candidate.pid = 200; + candidate.worker_birth_id = Some(2000); + candidate.token = request.successor_token.clone(); + let lock = open_resume_lock(root).unwrap(); + lock.try_lock().unwrap(); + write_json(&root.join(REQUEST), &request).unwrap(); + validate_candidate_request(&request, &candidate, true).unwrap(); + assert_eq!( + read_json::(&root.join(CANDIDATE)), + Some(candidate.clone()) + ); + // 模拟注册后 manager 退出/释放事务锁，发布 ready 必须失败。 + drop(lock); + assert!(validate_candidate_request(&request, &candidate, false).is_err()); + let lock = open_resume_lock(root).unwrap(); + lock.try_lock().unwrap(); + for change in 0..3 { + let mut invalid = request.clone(); + match change { + 0 => invalid.requester_birth.push('0'), + 1 => invalid.requester_pid = u32::MAX, + _ => invalid.expires_at_ms = current_timestamp_ms(), + } + write_json(&root.join(REQUEST), &invalid).unwrap(); + assert!(validate_candidate_request(&invalid, &candidate, true).is_err()); + } + write_json(&root.join(REQUEST), &serde_json::Value::Null).unwrap(); + assert!(validate_candidate_request(&request, &candidate, false).is_err()); + write_json( + &root.join(COMMIT), + &HandoffCommit { + request: request.clone(), + successor: candidate.clone(), + }, + ) + .unwrap(); + // 已提交代次不再依赖 manager 的请求、生命周期或 deadline。 + drop(lock); + validate_candidate_request(&request, &candidate, false).unwrap(); + candidate.worker_birth_id = Some(2001); + assert!(validate_candidate_request(&request, &candidate, false).is_err()); + } + + #[test] + fn successor_commit_requires_exact_generation_token_binary_and_identity() { + let temp = tempfile::tempdir().unwrap(); + let ready = ready_fixture(temp.path()); + let request = request_fixture(&ready); + let mut successor = ready.clone(); + successor.pid = 200; + successor.worker_birth_id = Some(2000); + successor.token = request.successor_token.clone(); + assert!(successor_matches(&request, &successor)); + for change in 0..5 { + let mut wrong = successor.clone(); + match change { + 0 => wrong.generation += 1, + 1 => wrong.token = "foreign".into(), + 2 => wrong.binary.size += 1, + 3 => wrong.identity.process_birth_id += 1, + _ => wrong.pid = ready.pid, + } + assert!(!successor_matches(&request, &wrong)); + } + } + + #[test] + fn committed_handoff_never_restores_old_services_but_revoked_release_does() { + assert!( + finish_handoff(true, false, || panic!("committed successor owns services")).unwrap() + ); + assert!(!finish_handoff(false, true, || panic!("manager still holds lease")).unwrap()); + let restored = std::cell::Cell::new(false); + assert!( + !finish_handoff(false, false, || { + restored.set(true); + Ok(()) + }) + .unwrap() + ); + assert!(restored.get()); + assert!(finish_handoff(false, false, || anyhow::bail!("retry restore")).is_err()); + } + + #[test] + fn legacy_reuse_requires_healthy_helper_between_two_identity_checks() { + let checks = std::cell::Cell::new(0); + verify_legacy_services_with( + || { + checks.set(checks.get() + 1); + Ok(true) + }, + || { + assert_eq!(checks.get(), 1); + Ok(()) + }, + ) + .unwrap(); + assert_eq!(checks.get(), 2); + assert!( + verify_legacy_services_with( + || Ok(false), + || panic!("unverified process cannot authorize helper reuse"), + ) + .is_err() + ); + assert!( + verify_legacy_services_with( + || Ok(true), + || anyhow::bail!("unhealthy helper; keep worker and defer"), + ) + .is_err() + ); + let checks = std::cell::Cell::new(0); + assert!( + verify_legacy_services_with( + || { + checks.set(checks.get() + 1); + Ok(checks.get() == 1) + }, + || Ok(()), + ) + .is_err() + ); + } + + #[test] + fn non_windows_never_waits_on_existing_workers_before_starting_main_services() { + let temp = tempfile::tempdir().unwrap(); + let worker = temp.path().join("live-worker"); + fs::create_dir(&worker).unwrap(); + // 即使存在损坏/不可核验的历史清单，非 Windows 的等待集合也必须为空。 + fs::write(worker.join("instance.json"), b"unknown legacy worker").unwrap(); + assert_eq!(worker_resume_roots(temp.path(), true), vec![worker]); + assert!(worker_resume_roots(temp.path(), false).is_empty()); + assert!(worker_resume_roots(&temp.path().join("missing"), false).is_empty()); + } + + #[test] + fn binary_stamp_does_not_reuse_replaced_same_path() { + let temp = tempfile::tempdir().unwrap(); + let exe = temp.path().join("manager.exe"); + fs::write(&exe, b"old").unwrap(); + let before = BinaryStamp::at(&exe).unwrap(); + fs::write(&exe, b"new worker build").unwrap(); + assert_ne!(before, BinaryStamp::at(&exe).unwrap()); + } + + #[test] + fn legacy_options_default_to_normal_start_and_resume_requires_token() { + let options: CodexMultiWorkerOptions = serde_json::from_value(json!({ + "appPath": "old", "debugPort": 1, "helperPort": 2, + })) + .unwrap(); + assert!(options.resume_identity.is_none()); + assert!(options.resume_token.is_none()); + let mut resume = options; + resume.resume_identity = Some(identity(Path::new("unused"))); + assert!(validate_worker_resume(&resume).is_err()); + } + + #[test] + fn dead_and_reused_worker_generations_do_not_block_main_restore() { + use codex_plus_core::watcher::ProcessInstanceState; + assert!(!generation_may_be_alive( + Some(1), + ProcessInstanceState::NotRunning + )); + assert!(!generation_may_be_alive( + Some(1), + ProcessInstanceState::Running { + started_at_secs: None, + birth_id: Some("2".into()), + } + )); + assert!(generation_may_be_alive( + Some(1), + ProcessInstanceState::Running { + started_at_secs: None, + birth_id: Some("1".into()), + } + )); + assert!(generation_may_be_alive( + Some(1), + ProcessInstanceState::Unknown + )); + let temp = tempfile::tempdir().unwrap(); + fs::write( + temp.path().join("instance.json"), + b"invalid historical json", + ) + .unwrap(); + assert!(!worker_could_hold_handles(temp.path())); + } + + #[test] + fn cancelled_or_expired_stage_stops_before_next_side_effect() { + assert!(check_active(&AtomicBool::new(true), Instant::now() + STAGE_TIMEOUT).is_err()); + assert!(check_active(&AtomicBool::new(false), Instant::now()).is_err()); + } +} diff --git a/apps/codex-plus-manager/src-tauri/src/lib.rs b/apps/codex-plus-manager/src-tauri/src/lib.rs index 1a4026b38..b69d099cd 100644 --- a/apps/codex-plus-manager/src-tauri/src/lib.rs +++ b/apps/codex-plus-manager/src-tauri/src/lib.rs @@ -241,15 +241,10 @@ pub fn run() { cu_pip::start(app.handle().clone()); commands::start_weixin_connect_from_saved_settings(); register_main_window_events(main_window); - commands::resume_existing_codex_background( + commands::multi_instance_resume::resume_on_manager_start( app.handle().clone(), start_codex_at_login || start_codex_on_open, ); - // 网关模式：有 Agent 走网关时常驻 helper 必须在线。 - gateway_commands::start_from_saved_state(); - gateway_commands::start_model_watch(); - // 自动接管官方 App（默认关闭）：设置里开着才起守护线程。 - auto_takeover::start_supervisor(); if let Err(error) = global_dictation::apply_config() { let _ = codex_plus_core::diagnostic_log::append_diagnostic_log( "dictation.global_hotkey_start_failed", diff --git a/apps/codex-plus-manager/src-tauri/src/runtime_health.rs b/apps/codex-plus-manager/src-tauri/src/runtime_health.rs index 1293e4e15..60718af20 100644 --- a/apps/codex-plus-manager/src-tauri/src/runtime_health.rs +++ b/apps/codex-plus-manager/src-tauri/src/runtime_health.rs @@ -39,13 +39,19 @@ impl RuntimeHealth { /// 仅在读取概览时检查本地端点，不启动服务、不请求供应商，也不创建后台监控线程。 pub fn inspect(settings: &BackendSettings, latest: Option<&LaunchStatus>) -> RuntimeHealth { - inspect_with( + let running = crate::commands::main_codex_running(latest); + let mut health = inspect_with( settings, latest, - !codex_plus_core::watcher::find_codex_processes().is_empty(), + running.unwrap_or(false), helper_available, codex_plus_core::cdp::endpoint_available, - ) + ); + if running.is_none() { + health.codex_app.status = "not_checked"; + health.codex_app.message = "无法确认主 Codex 身份，请重新连接或手动关闭后启动"; + } + health } fn inspect_with( diff --git a/apps/codex-plus-manager/src/App.tsx b/apps/codex-plus-manager/src/App.tsx index 61cf0a29c..ae32d694e 100644 --- a/apps/codex-plus-manager/src/App.tsx +++ b/apps/codex-plus-manager/src/App.tsx @@ -2719,7 +2719,7 @@ export function App() { // 界面要给出「启动失败 + 原因」，只留运行中的会让它彻底消失。 setMultiLaunchInstances( Array.isArray(result.instances) - ? result.instances.filter((instance) => multiLaunchInstanceRunning(instance) || Boolean(multiLaunchInstanceFailure(instance))) + ? result.instances.filter((instance) => multiLaunchInstanceStatus(instance) !== "stopped") : [], ); if (!silent && !isSuccessStatus(result.status)) showNotice(t("当前运行实例"), result.message, result.status); @@ -2736,7 +2736,7 @@ export function App() { setMultiLaunchActionPending(instance.instanceId); try { const result = await run(() => call("restart_codex_multi_instance", { - request: { instanceId: instance.instanceId }, + request: { instanceId: instance.instanceId, generation: instance.generation }, })); if (result) showNotice(t("重启 Codex 实例"), result.message, result.status); await refreshMultiLaunchInstances(true); @@ -2752,7 +2752,7 @@ export function App() { setMultiLaunchActionPending(instance.instanceId); try { const result = await run(() => call>("stop_codex_multi_instance", { - request: { instanceId: instance.instanceId }, + request: { instanceId: instance.instanceId, generation: instance.generation }, })); if (result) showNotice(t("结束 Codex 实例"), result.message, result.status); await refreshMultiLaunchInstances(true); @@ -2835,12 +2835,13 @@ export function App() { }; const launchCommand = async (command: "launch_codex_plus" | "restart_codex_plus" | "stop_codex_plus", syncActiveRelay = false) => { + const currentLaunch = command === "launch_codex_plus" ? null : overview?.latest_launch; const result = await run(() => call(command, { request: { - appPath: launchForm.appPath, - debugPort: numberOrDefault(launchForm.debugPort, 9229), - helperPort: numberOrDefault(launchForm.helperPort, 57321), + appPath: currentLaunch?.codex_app || launchForm.appPath, + debugPort: currentLaunch?.debug_port ?? numberOrDefault(launchForm.debugPort, 9229), + helperPort: currentLaunch?.helper_port ?? numberOrDefault(launchForm.helperPort, 57321), syncActiveRelay, }, }), @@ -7273,6 +7274,7 @@ function CodexMultiLaunchScreen({ {actions.multiLaunchInstances.map((instance) => { const status = multiLaunchInstanceStatus(instance); const running = multiLaunchInstanceRunning(instance); + const canStop = running || status === "unknown"; const pending = actions.multiLaunchActionPending === instance.instanceId; const label = formatMultiLaunchInstanceLabel(instance, { main: t("主 Codex"), fallback: t("Codex 实例") }); // 启动失败的原因必须让用户看得见：Windows 上多开曾经完全不吭声。 @@ -7282,7 +7284,7 @@ function CodexMultiLaunchScreen({ typeof instance.pid === "number" ? `PID ${instance.pid}` : "", typeof instance.debugPort === "number" ? `CDP ${instance.debugPort}` : "", typeof instance.helperPort === "number" ? `API ${instance.helperPort}` : "", - failure ? `${t("启动失败")}：${failure}` : "", + failure ? `${instance.launchResult?.status === "cleanup_failed" ? t("失败") : t("启动失败")}：${failure}` : "", ].filter(Boolean).join(" · "); return (
@@ -7303,10 +7305,10 @@ function CodexMultiLaunchScreen({ {pending ? t("处理中") : t("重启")}