Stabilize runtime control transport

nebula_app/src/atomic_file.rs

@@ -4,7 +4,7 @@
 //! so a crash cannot leave a half-written JSON document. A best-effort lock
 //! prevents two Nebula processes from compacting the same store at once.
 
-use std::fs::OpenOptions;
+use std::fs::{File, OpenOptions};
 use std::io::{self, Write};
 use std::path::{Path, PathBuf};
 use std::sync::atomic::{AtomicU64, Ordering};
@@ -64,6 +64,50 @@ pub(crate) fn try_lock(path: &Path) -> io::Result<Option<FileLock>> {
     }
 }
 
+/// Try to own a lock for the full lifetime of a resident process.
+///
+/// 与短状态写锁不同，这里依赖 OS 句柄锁：进程崩溃时内核立即释放，健康进程
+/// 无论运行多久都不会被“超时回收”。锁文件本身可以保留，所有权只看句柄。
+pub(crate) fn try_lifetime_lock(path: &Path) -> io::Result<Option<LifetimeFileLock>> {
+    let lock_path = lock_path(path);
+    if let Some(parent) = lock_path.parent() {
+        std::fs::create_dir_all(parent)?;
+    }
+
+    #[cfg(windows)]
+    let mut file = {
+        use std::os::windows::fs::OpenOptionsExt as _;
+
+        match OpenOptions::new().read(true).write(true).create(true).share_mode(0).open(&lock_path)
+        {
+            Ok(file) => file,
+            Err(error) if matches!(error.raw_os_error(), Some(32) | Some(33)) => return Ok(None),
+            Err(error) => return Err(error),
+        }
+    };
+
+    #[cfg(unix)]
+    let mut file = {
+        use std::os::fd::AsRawFd as _;
+
+        let file = OpenOptions::new().read(true).write(true).create(true).open(&lock_path)?;
+        let result = unsafe { libc::flock(file.as_raw_fd(), libc::LOCK_EX | libc::LOCK_NB) };
+        if result != 0 {
+            let error = io::Error::last_os_error();
+            if error.kind() == io::ErrorKind::WouldBlock {
+                return Ok(None);
+            }
+            return Err(error);
+        }
+        file
+    };
+
+    file.set_len(0)?;
+    write!(file, "{}", std::process::id())?;
+    file.sync_data()?;
+    Ok(Some(LifetimeFileLock { file }))
+}
+
 fn create_lock(path: &Path) -> io::Result<FileLock> {
     if let Some(parent) = path.parent() {
         std::fs::create_dir_all(parent)?;
@@ -99,6 +143,20 @@ impl Drop for FileLock {
     }
 }
 
+pub(crate) struct LifetimeFileLock {
+    file: File,
+}
+
+impl Drop for LifetimeFileLock {
+    fn drop(&mut self) {
+        #[cfg(unix)]
+        {
+            use std::os::fd::AsRawFd as _;
+            let _ = unsafe { libc::flock(self.file.as_raw_fd(), libc::LOCK_UN) };
+        }
+    }
+}
+
 #[cfg(windows)]
 fn replace(source: &Path, destination: &Path) -> io::Result<()> {
     use std::os::windows::ffi::OsStrExt;
@@ -131,7 +189,7 @@ fn replace(source: &Path, destination: &Path) -> io::Result<()> {
 
 #[cfg(test)]
 mod tests {
-    use super::{try_lock, write};
+    use super::{try_lifetime_lock, try_lock, write};
 
     #[test]
     fn atomic_write_replaces_and_lock_is_exclusive() {
@@ -145,5 +203,10 @@ mod tests {
         assert!(try_lock(&path).unwrap().is_none());
         drop(first);
         assert!(try_lock(&path).unwrap().is_some());
+
+        let owner = try_lifetime_lock(&path).unwrap().expect("first lifetime lock");
+        assert!(try_lifetime_lock(&path).unwrap().is_none());
+        drop(owner);
+        assert!(try_lifetime_lock(&path).unwrap().is_some());
     }
 }

nebula_app/src/gpui_shell/mod.rs

@@ -15,9 +15,9 @@ pub mod config;
 pub mod doc_tabs;
 pub mod http;
 pub mod math_view;
+pub mod network_settings;
 pub mod prelude;
 pub mod session_restore;
-pub mod network_settings;
 pub mod settings_pane;
 pub mod ssh_hosts;
 pub mod ssh_settings;
@@ -64,7 +64,7 @@ pub fn run_shell() {
     });
     // 驻留进程写 runtime.port：第二份 `nebula --gpui` 经 ATTACH + tab.new
     // 并入当前窗口，而不是再拉一套进程。Drop 会删 port 文件。
-    let _runtime_server = crate::runtime_api::RuntimeServer::spawn_callback(
+    let runtime_server = crate::runtime_api::RuntimeServer::spawn_callback(
         {
             let tx = shell_tx;
             move |callback| {
@@ -79,6 +79,18 @@ pub fn run_shell() {
         },
         runtime_hub.clone(),
     );
+    if runtime_server.is_none() {
+        // 与另一份进程同时启动时，对方可能已经拿到 owner lock、但尚未来得及
+        // 发布 endpoint。短暂等待并交接，避免继续打开一个无控制面的空窗口。
+        for _ in 0..40 {
+            if crate::runtime_api::try_open_default_tab_existing() {
+                crate::tray::shutdown();
+                return;
+            }
+            std::thread::sleep(std::time::Duration::from_millis(25));
+        }
+    }
+    let _runtime_server = runtime_server;
     gpui::Application::new().with_assets(Assets).run(move |cx| {
         // GPUI is the only event loop in `--gpui` mode, so it owns the same
         // per-process hook pipe before the first TerminalView spawns.
@@ -173,9 +185,8 @@ fn open_main_window(
     };
 
     cx.open_window(options, move |window, cx| {
-        let workspace = cx.new(|cx| {
-            NebulaWorkspace::new(window, ai_events, shell_events, 1, runtime_hub, cx)
-        });
+        let workspace =
+            cx.new(|cx| NebulaWorkspace::new(window, ai_events, shell_events, 1, runtime_hub, cx));
         // 调试/验收后门：启动即打开指定文档（见 open_document_at_startup）。
         if let Ok(path) = std::env::var("NEBULA_GPUI_OPEN_DOC")
             && !path.is_empty()

nebula_app/src/input/terminal_input.rs

@@ -282,33 +282,10 @@ pub(crate) fn build_runtime_sequence(
     result
 }
 
-/// Encode text injected by the Runtime API for the pane's active input protocol.
-///
-/// ConPTY's Win32 input mode deserializes `CSI Vk;Sc;Uc;Kd;Cs;Rc_` into
-/// `KEY_EVENT_RECORD`s. Mixing raw UTF-8 text and a serialized Enter record in
-/// one pipe batch can leave the control record in an interactive TUI's text
-/// input batch. Encode the text as `VK_PACKET` records as well, so a submitted
-/// prompt is one homogeneous protocol stream and needs no timing boundary.
-pub(crate) fn build_runtime_text_sequence(text: &str, mode: TermMode) -> Vec<u8> {
-    if use_win32_input_mode(mode) {
-        #[cfg(target_os = "windows")]
-        {
-            use windows_sys::Win32::UI::Input::KeyboardAndMouse::VK_PACKET;
-
-            let mut sequence = String::with_capacity(text.len().saturating_mul(40));
-            for unicode_char in text.encode_utf16() {
-                for key_down in [1, 0] {
-                    // VK_PACKET is the documented virtual key for Unicode
-                    // supplied by a non-keyboard input source. A zero scan
-                    // code and the UTF-16 unit preserve arbitrary prompt text.
-                    let _ =
-                        write!(sequence, "\x1b[{};0;{};{key_down};0;1_", VK_PACKET, unicode_char);
-                }
-            }
-            return sequence.into_bytes();
-        }
-    }
-
+/// Runtime 文本始终走终端的原生文本路径；ConPTY 会在 `WriteString` 中按当前
+/// Console/键盘布局生成事件。强制伪造成 `VK_PACKET` 会被旧 PSReadLine 丢弃。
+/// Enter 等控制键由 submit barrier 在下一批按当前协议单独编码。
+pub(crate) fn build_runtime_text_sequence(text: &str, _mode: TermMode) -> Vec<u8> {
     text.as_bytes().to_vec()
 }
 
@@ -1122,28 +1099,14 @@ mod vt_tests {
 
     #[cfg(target_os = "windows")]
     #[test]
-    fn runtime_win32_text_and_enter_form_one_record_stream() {
-        use windows_sys::Win32::UI::Input::KeyboardAndMouse::VK_PACKET;
-
+    fn runtime_win32_text_uses_conpty_native_text_path() {
         let mode = TermMode::WIN32_INPUT_MODE;
-        let mut bytes = build_runtime_text_sequence("A中", mode);
-        bytes.extend(build_runtime_sequence(
-            RuntimeKey::Enter,
-            RuntimeKeyModifiers::default(),
-            1,
-            mode,
-        ));
+        assert_eq!(build_runtime_text_sequence("A中", mode), "A中".as_bytes());
 
-        let text = String::from_utf8(bytes).unwrap();
-        let records: Vec<_> = text.split_inclusive('_').collect();
-        assert_eq!(records.len(), 6);
-        assert_eq!(records[0], format!("\x1b[{VK_PACKET};0;65;1;0;1_"));
-        assert_eq!(records[1], format!("\x1b[{VK_PACKET};0;65;0;0;1_"));
-        assert_eq!(records[2], format!("\x1b[{VK_PACKET};0;20013;1;0;1_"));
-        assert_eq!(records[3], format!("\x1b[{VK_PACKET};0;20013;0;0;1_"));
-        assert!(records[4].starts_with("\x1b[13;"));
-        assert!(records[4].contains(";13;1;0;1_"));
-        assert!(records[5].contains(";13;0;0;1_"));
+        let enter =
+            build_runtime_sequence(RuntimeKey::Enter, RuntimeKeyModifiers::default(), 1, mode);
+        assert_ne!(enter, b"\r");
+        assert!(String::from_utf8(enter).unwrap().starts_with("\x1b[13;"));
     }
 
     #[test]

nebula_app/src/runtime_api.rs

@@ -8,12 +8,22 @@
 mod agent_api;
 mod cli;
 mod command;
+mod server;
 #[cfg(test)]
 mod tests;
 
 pub use cli::run_cli;
-pub(crate) use command::{capture_process_tree, capture_terminal_tail, validate_command_line, validate_prompt};
-use command::{RuntimeWaitState, SubscribeParams, WaitParams, default_read_lines, default_true, parse_params, validate_agent_name, validate_agent_selector, wait_matches};
+use command::{
+    RuntimeWaitState, SubscribeParams, WaitParams, default_read_lines, default_true, parse_params,
+    validate_agent_name, validate_agent_selector, wait_matches,
+};
+pub(crate) use command::{
+    capture_process_tree, capture_terminal_tail, validate_command_line, validate_prompt,
+};
+use server::{Endpoint, endpoint_addr, read_endpoint};
+pub use server::{
+    RuntimeServer, dispatch_prompt, try_open_default_tab_existing, try_open_directory_existing,
+};
 
 use std::error::Error;
 use std::fmt;
@@ -1379,164 +1389,6 @@ fn run_result_or_capability_error(result: RuntimeRunResult) -> Result<RuntimeRun
         .details(serde_json::to_value(result).unwrap_or_else(|_| json!({ "reason": reason }))))
 }
 
-
-#[derive(Debug, Clone, PartialEq, Eq)]
-struct Endpoint {
-    port: u16,
-    token: String,
-}
-
-fn port_file() -> PathBuf {
-    crate::display::nebula_data_dir().join("runtime.port")
-}
-
-fn legacy_port_file() -> PathBuf {
-    crate::display::nebula_data_dir().join("mux.port")
-}
-
-fn read_endpoint() -> Option<Endpoint> {
-    read_endpoint_from(port_file())
-}
-
-fn read_endpoint_from(path: PathBuf) -> Option<Endpoint> {
-    let data = std::fs::read_to_string(path).ok()?;
-    let mut parts = data.split_whitespace();
-    Some(Endpoint { port: parts.next()?.parse().ok()?, token: parts.next()?.to_owned() })
-}
-
-fn endpoint_addr(endpoint: &Endpoint) -> SocketAddr {
-    SocketAddr::from((Ipv4Addr::LOCALHOST, endpoint.port))
-}
-
-fn fresh_token() -> String {
-    use std::hash::{BuildHasher, Hasher, RandomState};
-    let mut a = RandomState::new().build_hasher();
-    let mut b = RandomState::new().build_hasher();
-    a.write_u32(std::process::id());
-    b.write_u128(
-        std::time::SystemTime::now()
-            .duration_since(std::time::UNIX_EPOCH)
-            .map(|duration| duration.as_nanos())
-            .unwrap_or(0),
-    );
-    format!("{:016x}{:016x}", a.finish(), b.finish())
-}
-
-/// 普通二次启动并入驻留实例：先恢复/聚焦窗口，再新建一个默认 shell 标签页。
-pub fn try_open_default_tab_existing() -> bool {
-    try_open_tab_existing(None)
-}
-
-/// 后台任务把一行文本作为输入敲进某个 pane（不回车）。复用 runtime Prompt
-/// 的窗口定位与写入路径（`window_id=None` 按 pane 自动定位）；响应通道即弃
-/// ——调用方不关心结果，失败在事件线程侧留日志即可。SFTP 图片粘贴用它回粘
-/// 远端路径。
-pub fn dispatch_prompt(proxy: &EventLoopProxy<Event>, pane_id: u64, text: String) {
-    let (dispatch, _receiver) = RuntimeDispatch::new(RuntimeCommand::Prompt {
-        window_id: None,
-        pane_id,
-        text,
-        submit: false,
-    });
-    if proxy.send_event(Event::new(EventType::RuntimeControl(dispatch), None)).is_err() {
-        warn!("prompt dispatch failed: event loop is gone");
-    }
-}
-
-/// Explorer 右键「在 Nebula 中打开」/ CLI 带 `--working-directory` 的启动并入
-/// 驻留实例：先经 ATTACH 恢复或聚焦既有窗口（与平启动同一套交接，detached
-/// 的标签因此先回来），再请求在该窗口打开定目录标签。任一步失败都返回
-/// false，调用方回落为独立启动——绝不吞掉用户的手势。
-pub fn try_open_directory_existing(dir: &std::path::Path) -> bool {
-    try_open_tab_existing(Some(dir))
-}
-
-fn try_open_tab_existing(dir: Option<&std::path::Path>) -> bool {
-    if legacy_request("ATTACH").is_none() {
-        return false;
-    }
-    // ATTACH 与 tab.new 都落到同一条 winit 事件队列上，先后有序：窗口先
-    // 恢复，新标签随后落在恢复出来的窗口里。
-    let params = dir.map_or_else(|| json!({}), |dir| json!({ "cwd": dir }));
-    cli::request_once("tab.new", params, IO_TIMEOUT).map(|response| response.ok).unwrap_or(false)
-}
-
-fn legacy_request(verb: &str) -> Option<()> {
-    read_endpoint()
-        .and_then(|endpoint| legacy_request_to(verb, &endpoint))
-        // Upgrade compatibility: an already-running pre-v1 Nebula still
-        // publishes mux.port. A plain launch can attach to it and exit; a new
-        // resident process writes runtime.port after the old process ends.
-        .or_else(|| {
-            read_endpoint_from(legacy_port_file())
-                .and_then(|endpoint| legacy_request_to(verb, &endpoint))
-        })
-}
-
-fn legacy_request_to(verb: &str, endpoint: &Endpoint) -> Option<()> {
-    let mut stream = TcpStream::connect_timeout(&endpoint_addr(&endpoint), CONNECT_TIMEOUT).ok()?;
-    stream.set_read_timeout(Some(IO_TIMEOUT)).ok()?;
-    stream.set_write_timeout(Some(IO_TIMEOUT)).ok()?;
-    stream.write_all(format!("{verb} {}\n", endpoint.token).as_bytes()).ok()?;
-    let mut line = String::new();
-    BufReader::new(stream).read_line(&mut line).ok()?;
-    (line.trim() == "OK").then_some(())
-}
-
-/// Resident versioned runtime API server.
-pub struct RuntimeServer {
-    endpoint: Endpoint,
-    port_file: PathBuf,
-}
-
-impl RuntimeServer {
-    pub fn spawn(proxy: EventLoopProxy<Event>, hub: RuntimeHub) -> Option<Self> {
-        Self::spawn_with_sink(EventSink::Winit(proxy), hub)
-    }
-
-    /// Same discovery file and ATTACH/PING/JSON protocol as [`Self::spawn`],
-    /// but attach/control are delivered through a callback (GPUI has no winit
-    /// `EventLoopProxy`).
-    pub fn spawn_callback(
-        on_event: impl Fn(RuntimeCallback) + Send + Sync + 'static,
-        hub: RuntimeHub,
-    ) -> Option<Self> {
-        Self::spawn_with_sink(EventSink::Callback(Arc::new(on_event)), hub)
-    }
-
-    fn spawn_with_sink(sink: EventSink, hub: RuntimeHub) -> Option<Self> {
-        if read_endpoint().and_then(|endpoint| legacy_request_to("PING", &endpoint)).is_some() {
-            info!("Runtime API server already running; this instance stays client-only");
-            return None;
-        }
-
-        let listener = TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).ok()?;
-        let endpoint = Endpoint { port: listener.local_addr().ok()?.port(), token: fresh_token() };
-        let path = port_file();
-        let contents = format!("{} {} {}\n", endpoint.port, endpoint.token, PROTOCOL_VERSION);
-        if crate::atomic_file::write(&path, contents.as_bytes()).is_err() {
-            warn!("Runtime API: cannot write {path:?}; control plane disabled");
-            return None;
-        }
-
-        let server_token = endpoint.token.clone();
-        let spawned = std::thread::Builder::new()
-            .name("nebula-runtime-api".into())
-            .spawn(move || serve(listener, server_token, sink, hub))
-            .is_ok();
-        spawned.then(|| Self { endpoint, port_file: path })
-    }
-}
-
-impl Drop for RuntimeServer {
-    fn drop(&mut self) {
-        // Do not delete another process's newer discovery record.
-        if read_endpoint().as_ref() == Some(&self.endpoint) {
-            let _ = std::fs::remove_file(&self.port_file);
-        }
-    }
-}
-
 fn serve(listener: TcpListener, token: String, sink: EventSink, hub: RuntimeHub) {
     let active = Arc::new(AtomicUsize::new(0));
     for stream in listener.incoming() {

nebula_app/src/runtime_api/cli.rs

@@ -22,14 +22,18 @@ pub(super) fn request_once(
     params: Value,
     timeout: Duration,
 ) -> Result<ApiResponse, Box<dyn Error>> {
-    let endpoint =
-        read_endpoint().ok_or_else(|| CliError("no resident Nebula runtime found".into()))?;
+    let endpoint = read_endpoint()
+        .ok_or_else(|| CliError::new("runtime_unavailable", "no resident Nebula runtime found"))?;
     let request = ApiRequest::new(endpoint.token.clone(), method, params);
     let stream = client_stream(&endpoint, &request, Some(timeout))?;
     let mut line = String::new();
     BufReader::new(stream).read_line(&mut line)?;
     if line.is_empty() {
-        return Err(CliError("runtime closed the connection without a response".into()).into());
+        return Err(CliError::new(
+            "runtime_no_response",
+            "runtime closed the connection without a response",
+        )
+        .into());
     }
     Ok(serde_json::from_str(&line)?)
 }
@@ -45,15 +49,31 @@ fn print_response(response: &ApiResponse, pretty: bool) -> Result<(), Box<dyn Er
     } else {
         let error =
             response.error.as_ref().map_or("runtime request failed", |error| &error.message);
-        Err(CliError(error.to_owned()).into())
+        Err(PrintedCliError(error.to_owned()).into())
     }
 }
 
 /// CLI adapter. It deliberately speaks the same serialized protocol as any
 /// external client instead of calling Processor helpers in-process.
 pub fn run_cli(options: ControlOptions) -> Result<(), Box<dyn Error>> {
+    let pretty = options.pretty;
+    match run_cli_inner(options) {
+        result @ Ok(_) => result,
+        Err(error) if error.downcast_ref::<PrintedCliError>().is_some() => Err(error),
+        Err(error) => {
+            let code =
+                error.downcast_ref::<CliError>().map_or("cli_transport_error", |error| error.code);
+            let response = ApiResponse::failure("cli", ApiError::new(code, error.to_string()));
+            print_response(&response, pretty)
+        },
+    }
+}
+
+fn run_cli_inner(options: ControlOptions) -> Result<(), Box<dyn Error>> {
     if options.timeout_ms == 0 || Duration::from_millis(options.timeout_ms) > MAX_WAIT {
-        return Err(CliError("--timeout-ms must be between 1 and 86400000".into()).into());
+        return Err(
+            CliError::new("invalid_params", "--timeout-ms must be between 1 and 86400000").into()
+        );
     }
     let timeout = Duration::from_millis(options.timeout_ms);
     match options.command {
@@ -310,8 +330,8 @@ fn wait_state_name(state: ControlWaitState) -> &'static str {
 }
 
 fn subscribe_cli(since: Option<u64>, timeout: Duration) -> Result<(), Box<dyn Error>> {
-    let endpoint =
-        read_endpoint().ok_or_else(|| CliError("no resident Nebula runtime found".into()))?;
+    let endpoint = read_endpoint()
+        .ok_or_else(|| CliError::new("runtime_unavailable", "no resident Nebula runtime found"))?;
     let request = ApiRequest::new(
         endpoint.token.clone(),
         "events.subscribe",
@@ -321,9 +341,11 @@ fn subscribe_cli(since: Option<u64>, timeout: Duration) -> Result<(), Box<dyn Er
     let mut reader = BufReader::new(stream);
     let mut line = String::new();
     if reader.read_line(&mut line)? == 0 {
-        return Err(
-            CliError("runtime closed the subscription without an acknowledgement".into()).into()
-        );
+        return Err(CliError::new(
+            "runtime_no_response",
+            "runtime closed the subscription without an acknowledgement",
+        )
+        .into());
     }
     print!("{line}");
     std::io::stdout().flush()?;
@@ -339,12 +361,32 @@ fn subscribe_cli(since: Option<u64>, timeout: Duration) -> Result<(), Box<dyn Er
 }
 
 #[derive(Debug)]
-struct CliError(String);
+struct CliError {
+    code: &'static str,
+    message: String,
+}
+
+impl CliError {
+    fn new(code: &'static str, message: impl Into<String>) -> Self {
+        Self { code, message: message.into() }
+    }
+}
 
 impl fmt::Display for CliError {
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        formatter.write_str(&self.message)
+    }
+}
+
+impl Error for CliError {}
+
+#[derive(Debug)]
+struct PrintedCliError(String);
+
+impl fmt::Display for PrintedCliError {
     fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
         formatter.write_str(&self.0)
     }
 }
 
-impl Error for CliError {}
+impl Error for PrintedCliError {}

nebula_app/src/runtime_api/server.rs

@@ -0,0 +1,161 @@
+//! Runtime endpoint discovery and resident server ownership.
+
+use super::*;
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub(super) struct Endpoint {
+    pub(super) port: u16,
+    pub(super) token: String,
+}
+
+fn port_file() -> PathBuf {
+    crate::display::nebula_data_dir().join("runtime.port")
+}
+
+fn legacy_port_file() -> PathBuf {
+    crate::display::nebula_data_dir().join("mux.port")
+}
+
+pub(super) fn read_endpoint() -> Option<Endpoint> {
+    read_endpoint_from(port_file())
+}
+
+fn read_endpoint_from(path: PathBuf) -> Option<Endpoint> {
+    let data = std::fs::read_to_string(path).ok()?;
+    let mut parts = data.split_whitespace();
+    Some(Endpoint { port: parts.next()?.parse().ok()?, token: parts.next()?.to_owned() })
+}
+
+pub(super) fn endpoint_addr(endpoint: &Endpoint) -> SocketAddr {
+    SocketAddr::from((Ipv4Addr::LOCALHOST, endpoint.port))
+}
+
+fn fresh_token() -> String {
+    use std::hash::{BuildHasher, Hasher, RandomState};
+    let mut a = RandomState::new().build_hasher();
+    let mut b = RandomState::new().build_hasher();
+    a.write_u32(std::process::id());
+    b.write_u128(
+        std::time::SystemTime::now()
+            .duration_since(std::time::UNIX_EPOCH)
+            .map(|duration| duration.as_nanos())
+            .unwrap_or(0),
+    );
+    format!("{:016x}{:016x}", a.finish(), b.finish())
+}
+
+/// 普通二次启动并入驻留实例：先恢复/聚焦窗口，再新建一个默认 shell 标签页。
+pub fn try_open_default_tab_existing() -> bool {
+    try_open_tab_existing(None)
+}
+
+/// 后台任务把一行文本作为输入敲进某个 pane（不回车）。
+pub fn dispatch_prompt(proxy: &EventLoopProxy<Event>, pane_id: u64, text: String) {
+    let (dispatch, _receiver) = RuntimeDispatch::new(RuntimeCommand::Prompt {
+        window_id: None,
+        pane_id,
+        text,
+        submit: false,
+    });
+    if proxy.send_event(Event::new(EventType::RuntimeControl(dispatch), None)).is_err() {
+        warn!("prompt dispatch failed: event loop is gone");
+    }
+}
+
+/// Explorer 右键或带 `--working-directory` 的启动并入驻留实例。
+pub fn try_open_directory_existing(dir: &std::path::Path) -> bool {
+    try_open_tab_existing(Some(dir))
+}
+
+fn try_open_tab_existing(dir: Option<&std::path::Path>) -> bool {
+    if legacy_request("ATTACH").is_none() {
+        return false;
+    }
+    // ATTACH 与 tab.new 落到同一事件队列，窗口先恢复，新标签随后创建。
+    let params = dir.map_or_else(|| json!({}), |dir| json!({ "cwd": dir }));
+    cli::request_once("tab.new", params, IO_TIMEOUT).map(|response| response.ok).unwrap_or(false)
+}
+
+fn legacy_request(verb: &str) -> Option<()> {
+    read_endpoint()
+        .and_then(|endpoint| legacy_request_to(verb, &endpoint))
+        // 已运行的 pre-v1 版本只发布 mux.port；升级期间仍允许普通启动交接。
+        .or_else(|| {
+            read_endpoint_from(legacy_port_file())
+                .and_then(|endpoint| legacy_request_to(verb, &endpoint))
+        })
+}
+
+fn legacy_request_to(verb: &str, endpoint: &Endpoint) -> Option<()> {
+    let mut stream = TcpStream::connect_timeout(&endpoint_addr(endpoint), CONNECT_TIMEOUT).ok()?;
+    stream.set_read_timeout(Some(IO_TIMEOUT)).ok()?;
+    stream.set_write_timeout(Some(IO_TIMEOUT)).ok()?;
+    stream.write_all(format!("{verb} {}\n", endpoint.token).as_bytes()).ok()?;
+    let mut line = String::new();
+    BufReader::new(stream).read_line(&mut line).ok()?;
+    (line.trim() == "OK").then_some(())
+}
+
+/// Resident versioned runtime API server.
+pub struct RuntimeServer {
+    endpoint: Endpoint,
+    port_file: PathBuf,
+    _owner_lock: crate::atomic_file::LifetimeFileLock,
+}
+
+impl RuntimeServer {
+    pub fn spawn(proxy: EventLoopProxy<Event>, hub: RuntimeHub) -> Option<Self> {
+        Self::spawn_with_sink(EventSink::Winit(proxy), hub)
+    }
+
+    /// GPUI variant of [`Self::spawn`], delivered through a callback.
+    pub fn spawn_callback(
+        on_event: impl Fn(RuntimeCallback) + Send + Sync + 'static,
+        hub: RuntimeHub,
+    ) -> Option<Self> {
+        Self::spawn_with_sink(EventSink::Callback(Arc::new(on_event)), hub)
+    }
+
+    fn spawn_with_sink(sink: EventSink, hub: RuntimeHub) -> Option<Self> {
+        let path = port_file();
+        let owner_lock = match crate::atomic_file::try_lifetime_lock(&path) {
+            Ok(Some(lock)) => lock,
+            Ok(None) => {
+                info!("Runtime API owner is starting or already running; staying client-only");
+                return None;
+            },
+            Err(error) => {
+                warn!("Runtime API: cannot lock {path:?}: {error}; control plane disabled");
+                return None;
+            },
+        };
+        if read_endpoint().and_then(|endpoint| legacy_request_to("PING", &endpoint)).is_some() {
+            info!("Runtime API server already running; this instance stays client-only");
+            return None;
+        }
+
+        let listener = TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).ok()?;
+        let endpoint = Endpoint { port: listener.local_addr().ok()?.port(), token: fresh_token() };
+        let contents = format!("{} {} {}\n", endpoint.port, endpoint.token, PROTOCOL_VERSION);
+        if crate::atomic_file::write(&path, contents.as_bytes()).is_err() {
+            warn!("Runtime API: cannot write {path:?}; control plane disabled");
+            return None;
+        }
+
+        let server_token = endpoint.token.clone();
+        let spawned = std::thread::Builder::new()
+            .name("nebula-runtime-api".into())
+            .spawn(move || serve(listener, server_token, sink, hub))
+            .is_ok();
+        spawned.then(|| Self { endpoint, port_file: path, _owner_lock: owner_lock })
+    }
+}
+
+impl Drop for RuntimeServer {
+    fn drop(&mut self) {
+        // Do not delete another process's newer discovery record.
+        if read_endpoint().as_ref() == Some(&self.endpoint) {
+            let _ = std::fs::remove_file(&self.port_file);
+        }
+    }
+}