Add direct SSH terminal transport

nebula_app/Cargo.toml

@@ -46,6 +46,7 @@ polling = "3.11.0"
 # Markdown viewer: CommonMark + GFM (tables / task lists / strikethrough)
 # event parser; `display::markdown_view` renders the folded block AST.
 pulldown-cmark = { version = "0.13", default-features = false }
+russh = { version = "0.62.2", default-features = false, features = ["ring"] }
 # preserve_order: we rewrite the user's claude settings.json (ai_hook
 # self-heal) and must not alphabetize their keys in the process.
 serde_json = { version = "1", features = ["preserve_order"] }
@@ -55,6 +56,7 @@ signal-hook = { version = "0.4.3", default-features = false }
 tempfile = "3.12.0"
 toml.workspace = true
 toml_edit.workspace = true
+tokio = { version = "1", features = ["rt-multi-thread", "sync", "time"] }
 unicode-width = "0.2.0"
 winit = { version = "0.30.9", default-features = false, features = ["rwh_06", "serde"] }
 global-hotkey = "0.8.0"

nebula_app/src/main.rs

@@ -60,6 +60,8 @@ mod ssh_credentials;
 mod mux;
 #[cfg(windows)]
 mod ssh;
+#[cfg(windows)]
+mod ssh_session;
 mod string;
 mod window_context;
 

nebula_app/src/ssh_credentials.rs

@@ -304,6 +304,16 @@ pub fn store_password(destination: &str, password: &[u8]) -> std::io::Result<()>
     windows_store::save_password(destination, password)
 }
 
+#[cfg(windows)]
+pub fn load_stored_password(destination: &str) -> std::io::Result<Option<Vec<u8>>> {
+    windows_store::load_password(destination)
+}
+
+#[cfg(not(windows))]
+pub fn load_stored_password(_destination: &str) -> std::io::Result<Option<Vec<u8>>> {
+    Ok(None)
+}
+
 #[cfg(windows)]
 pub fn forget_password(destination: &str) -> std::io::Result<()> {
     windows_store::delete_password(destination)

nebula_app/src/ssh_session.rs

@@ -0,0 +1,340 @@
+//! 由 SSH 通道直接驱动的远端终端会话。
+//!
+//! 远端 Pane 不创建本地伪终端，但继续使用统一的输入、缩放和关闭消息协议，
+//! 从而让渲染与键盘处理保持传输层无关。
+
+use std::io;
+use std::sync::Arc;
+use std::sync::mpsc::Receiver;
+use std::time::Duration;
+
+use log::{error, warn};
+use nebula_terminal::event::{Event as TerminalEvent, WindowSize};
+use nebula_terminal::event_loop::{EventLoopSender, Msg, StreamProcessor};
+use nebula_terminal::sync::FairMutex;
+use nebula_terminal::term::Term;
+use russh::client;
+use russh::keys::ssh_key;
+use russh::{ChannelMsg, Disconnect};
+
+use crate::event::EventProxy;
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct SshDestination {
+    pub original: String,
+    pub user: String,
+    pub host: String,
+    pub port: u16,
+}
+
+impl SshDestination {
+    pub fn parse(value: &str) -> io::Result<Self> {
+        let original = value.trim().to_owned();
+        let address = original.strip_prefix("ssh://").unwrap_or(&original).to_owned();
+        let (user, host_port) = address.rsplit_once('@').ok_or_else(|| {
+            io::Error::new(io::ErrorKind::InvalidInput, "SSH destination requires user@host")
+        })?;
+        if user.is_empty() || host_port.is_empty() {
+            return Err(io::Error::new(
+                io::ErrorKind::InvalidInput,
+                "SSH destination is incomplete",
+            ));
+        }
+
+        let (host, port) = if let Some(rest) = host_port.strip_prefix('[') {
+            let (host, suffix) = rest.split_once(']').ok_or_else(|| {
+                io::Error::new(io::ErrorKind::InvalidInput, "invalid bracketed SSH address")
+            })?;
+            let port = suffix
+                .strip_prefix(':')
+                .map(str::parse)
+                .transpose()
+                .map_err(|_| io::Error::new(io::ErrorKind::InvalidInput, "invalid SSH port"))?
+                .unwrap_or(22);
+            (host.to_owned(), port)
+        } else if let Some((host, port)) = host_port.rsplit_once(':') {
+            if host.contains(':') {
+                (host_port.to_owned(), 22)
+            } else {
+                let port = port
+                    .parse()
+                    .map_err(|_| io::Error::new(io::ErrorKind::InvalidInput, "invalid SSH port"))?;
+                (host.to_owned(), port)
+            }
+        } else {
+            (host_port.to_owned(), 22)
+        };
+        if host.is_empty() {
+            return Err(io::Error::new(io::ErrorKind::InvalidInput, "SSH host is empty"));
+        }
+
+        Ok(Self { original, user: user.to_owned(), host, port })
+    }
+}
+
+struct ClientHandler {
+    host: String,
+    port: u16,
+}
+
+impl client::Handler for ClientHandler {
+    type Error = russh::Error;
+
+    async fn check_server_key(
+        &mut self,
+        server_public_key: &ssh_key::PublicKey,
+    ) -> Result<bool, Self::Error> {
+        match russh::keys::known_hosts::check_known_hosts(&self.host, self.port, server_public_key)
+        {
+            Ok(true) => Ok(true),
+            Ok(false) => {
+                if confirm_new_host(&self.host, self.port, server_public_key) {
+                    if let Err(err) = russh::keys::known_hosts::learn_known_hosts(
+                        &self.host,
+                        self.port,
+                        server_public_key,
+                    ) {
+                        warn!("Failed to store SSH host key: {err}");
+                    }
+                    Ok(true)
+                } else {
+                    Ok(false)
+                }
+            },
+            Err(err) => {
+                warn!("SSH host key verification failed: {err}");
+                show_host_key_changed(&self.host, self.port, &err.to_string());
+                Ok(false)
+            },
+        }
+    }
+}
+
+/// 启动密码认证的 SSH Pane。返回 `Unsupported` 时保留兼容路径，
+/// 避免尚未覆盖的认证方式在迁移期间失效。
+pub fn spawn_password_session(
+    destination: SshDestination,
+    initial_size: WindowSize,
+    terminal: Arc<FairMutex<Term<EventProxy>>>,
+    event_proxy: EventProxy,
+) -> io::Result<EventLoopSender> {
+    let Some(password) = crate::ssh_credentials::load_stored_password(&destination.original)?
+    else {
+        return Err(io::Error::new(
+            io::ErrorKind::Unsupported,
+            "no stored password for direct SSH authentication",
+        ));
+    };
+    let (sender, receiver) = EventLoopSender::standalone()?;
+    std::thread::Builder::new().name(format!("SSH {}", destination.host)).spawn(move || {
+        run_session(destination, password, initial_size, terminal, event_proxy, receiver)
+    })?;
+    Ok(sender)
+}
+
+fn run_session(
+    destination: SshDestination,
+    mut password: Vec<u8>,
+    initial_size: WindowSize,
+    terminal: Arc<FairMutex<Term<EventProxy>>>,
+    event_proxy: EventProxy,
+    receiver: Receiver<Msg>,
+) {
+    let runtime =
+        match tokio::runtime::Builder::new_multi_thread().enable_all().worker_threads(2).build() {
+            Ok(runtime) => runtime,
+            Err(err) => {
+                render_error(&terminal, &event_proxy, &format!("SSH runtime failed: {err}"));
+                return;
+            },
+        };
+    runtime.block_on(async move {
+        if let Err(err) = run_session_async(
+            &destination,
+            &password,
+            initial_size,
+            terminal.clone(),
+            event_proxy.clone(),
+            receiver,
+        )
+        .await
+        {
+            error!("Direct SSH session failed for {}: {err}", destination.original);
+            render_error(&terminal, &event_proxy, &format!("SSH connection failed: {err}"));
+        }
+        password.fill(0);
+        terminal.lock().exit();
+        event_proxy.send_event(TerminalEvent::Wakeup.into());
+    });
+}
+
+async fn run_session_async(
+    destination: &SshDestination,
+    password: &[u8],
+    initial_size: WindowSize,
+    terminal: Arc<FairMutex<Term<EventProxy>>>,
+    event_proxy: EventProxy,
+    receiver: Receiver<Msg>,
+) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+    let config = Arc::new(client::Config {
+        inactivity_timeout: Some(Duration::from_secs(30)),
+        keepalive_interval: Some(Duration::from_secs(5)),
+        keepalive_max: 6,
+        ..Default::default()
+    });
+    let handler = ClientHandler { host: destination.host.clone(), port: destination.port };
+    let mut session =
+        client::connect(config, (destination.host.as_str(), destination.port), handler).await?;
+
+    let password = String::from_utf8(password.to_vec())?;
+    let auth = session.authenticate_password(destination.user.clone(), password).await?;
+    if !auth.success() {
+        let _ = crate::ssh_credentials::forget_password(&destination.original);
+        return Err("password authentication was rejected".into());
+    }
+
+    let mut channel = session.channel_open_session().await?;
+    channel
+        .request_pty(
+            true,
+            "xterm-256color",
+            u32::from(initial_size.num_cols),
+            u32::from(initial_size.num_lines),
+            u32::from(initial_size.cell_width) * u32::from(initial_size.num_cols),
+            u32::from(initial_size.cell_height) * u32::from(initial_size.num_lines),
+            &[],
+        )
+        .await?;
+    channel.request_shell(true).await?;
+
+    let (input_tx, mut input_rx) = tokio::sync::mpsc::unbounded_channel();
+    tokio::task::spawn_blocking(move || {
+        while let Ok(message) = receiver.recv() {
+            if input_tx.send(message).is_err() {
+                break;
+            }
+        }
+    });
+
+    let mut stream = StreamProcessor::default();
+    stream.resize(initial_size);
+    loop {
+        tokio::select! {
+            message = input_rx.recv() => match message {
+                Some(Msg::Input(bytes)) => channel.data(bytes.as_ref()).await?,
+                Some(Msg::Resize(size)) => {
+                    stream.resize(size);
+                    channel.window_change(
+                        u32::from(size.num_cols),
+                        u32::from(size.num_lines),
+                        u32::from(size.cell_width) * u32::from(size.num_cols),
+                        u32::from(size.cell_height) * u32::from(size.num_lines),
+                    ).await?;
+                },
+                Some(Msg::Shutdown) | None => {
+                    let _ = channel.eof().await;
+                    break;
+                },
+            },
+            message = channel.wait() => match message {
+                Some(ChannelMsg::Data { data }) | Some(ChannelMsg::ExtendedData { data, .. }) => {
+                    stream.feed(&mut *terminal.lock(), &event_proxy, data.as_ref());
+                    event_proxy.send_event(TerminalEvent::Wakeup.into());
+                },
+                Some(ChannelMsg::ExitStatus { .. }) | None => break,
+                _ => {},
+            },
+        }
+    }
+    let _ = session.disconnect(Disconnect::ByApplication, "", "English").await;
+    Ok(())
+}
+
+fn render_error(
+    terminal: &Arc<FairMutex<Term<EventProxy>>>,
+    event_proxy: &EventProxy,
+    message: &str,
+) {
+    let mut stream = StreamProcessor::default();
+    let text = format!("\r\n\x1b[31m{message}\x1b[0m\r\n");
+    stream.feed(&mut *terminal.lock(), event_proxy, text.as_bytes());
+    event_proxy.send_event(TerminalEvent::Wakeup.into());
+}
+
+#[cfg(windows)]
+fn confirm_new_host(host: &str, port: u16, key: &ssh_key::PublicKey) -> bool {
+    use std::ptr::null_mut;
+    use windows_sys::Win32::UI::WindowsAndMessaging::{
+        IDYES, MB_ICONQUESTION, MB_SETFOREGROUND, MB_YESNO, MessageBoxW,
+    };
+
+    let fingerprint = key.fingerprint(ssh_key::HashAlg::Sha256);
+    let text = wide(&format!(
+        "首次连接到 {host}:{port}。\n\n主机密钥：{fingerprint}\n\n是否信任并保存此主机密钥？"
+    ));
+    let title = wide("Nebula SSH");
+    unsafe {
+        MessageBoxW(
+            null_mut(),
+            text.as_ptr(),
+            title.as_ptr(),
+            MB_YESNO | MB_ICONQUESTION | MB_SETFOREGROUND,
+        ) == IDYES
+    }
+}
+
+#[cfg(not(windows))]
+fn confirm_new_host(_host: &str, _port: u16, _key: &ssh_key::PublicKey) -> bool {
+    false
+}
+
+#[cfg(windows)]
+fn show_host_key_changed(host: &str, port: u16, detail: &str) {
+    use std::ptr::null_mut;
+    use windows_sys::Win32::UI::WindowsAndMessaging::{
+        MB_ICONERROR, MB_OK, MB_SETFOREGROUND, MessageBoxW,
+    };
+    let text = wide(&format!(
+        "{host}:{port} 的主机密钥与已保存记录不一致。\n\n连接已终止，以避免连接到错误的主机。\n\n{detail}"
+    ));
+    let title = wide("Nebula SSH");
+    unsafe {
+        MessageBoxW(
+            null_mut(),
+            text.as_ptr(),
+            title.as_ptr(),
+            MB_OK | MB_ICONERROR | MB_SETFOREGROUND,
+        );
+    }
+}
+
+#[cfg(not(windows))]
+fn show_host_key_changed(_host: &str, _port: u16, _detail: &str) {}
+
+#[cfg(windows)]
+fn wide(value: &str) -> Vec<u16> {
+    value.encode_utf16().chain(std::iter::once(0)).collect()
+}
+
+#[cfg(test)]
+mod tests {
+    use super::SshDestination;
+
+    #[test]
+    fn parses_saved_destinations() {
+        let plain = SshDestination::parse("root@example.com").unwrap();
+        assert_eq!(
+            (plain.user.as_str(), plain.host.as_str(), plain.port),
+            ("root", "example.com", 22)
+        );
+
+        let uri = SshDestination::parse("ssh://alice@example.com:2200").unwrap();
+        assert_eq!(
+            (uri.user.as_str(), uri.host.as_str(), uri.port),
+            ("alice", "example.com", 2200)
+        );
+
+        let ipv6 = SshDestination::parse("ssh://root@[2001:db8::1]:2222").unwrap();
+        assert_eq!((ipv6.host.as_str(), ipv6.port), ("2001:db8::1", 2222));
+    }
+}

nebula_app/src/window_context.rs

@@ -543,6 +543,49 @@ impl WindowContext {
         })
     }
 
+    #[cfg(windows)]
+    /// 创建由远端 PTY 通道驱动的 Pane，并复用本地终端的解析、渲染和事件协议。
+    /// 这样传输层只负责字节流，输入、缩放与终端状态无需维护两套实现。
+    fn create_ssh_pane(
+        size_info: &crate::display::SizeInfo,
+        window_id: WindowId,
+        config: &UiConfig,
+        proxy: &EventLoopProxy<Event>,
+        pane_id: PaneId,
+        destination: crate::ssh_session::SshDestination,
+    ) -> Result<Pane, Box<dyn Error>> {
+        let window_route = Arc::new(AtomicU64::new(window_id.into()));
+        let event_proxy = EventProxy::new_tab(proxy.clone(), window_route.clone(), pane_id);
+        let terminal = Arc::new(FairMutex::new(Term::new(
+            config.term_options(),
+            size_info,
+            event_proxy.clone(),
+        )));
+        let sender = crate::ssh_session::spawn_password_session(
+            destination,
+            (*size_info).into(),
+            terminal.clone(),
+            event_proxy.clone(),
+        )?;
+        if config.cursor.style().blinking {
+            event_proxy.send_event(TerminalEvent::CursorBlinkingChange.into());
+        }
+        Ok(Pane {
+            terminal,
+            notifier: Notifier(sender),
+            search_state: Default::default(),
+            inline_search_state: Default::default(),
+            id: pane_id,
+            title: String::from("ssh"),
+            nebula_state: NebulaPaneState::default(),
+            intro_cols: None,
+            shell_pid: 0,
+            window_route,
+            #[cfg(not(windows))]
+            master_fd: -1,
+        })
+    }
+
     /// A pane-shaped stub for document-viewer tabs: a real (empty) `Term` so
     /// the shared event pipeline has state to borrow, but NO PTY behind it —
     /// the notifier is a sink, so keystrokes routed here are swallowed
@@ -856,6 +899,47 @@ impl WindowContext {
     /// `nebula ssh` is typed into that shell's PTY so OpenSSH remains inside
     /// Nebula's ConPTY instead of becoming the pane's GUI-subsystem root.
     fn spawn_tab_ssh(&mut self, host: String) {
+        #[cfg(windows)]
+        if let Ok(destination) = crate::ssh_session::SshDestination::parse(&host) {
+            let pane_id = self.next_pane_id;
+            match Self::create_ssh_pane(
+                &self.display.size_info,
+                self.display.window.id(),
+                &self.config,
+                &self.proxy,
+                pane_id,
+                destination,
+            ) {
+                Ok(pane) => {
+                    self.next_pane_id += 1;
+                    self.panes.push(pane);
+                    let at = (self.active_tab + 1).min(self.tabs.len());
+                    self.tabs.insert(at, TabEntry {
+                        layout: Layout::Leaf(pane_id),
+                        active_pane: pane_id,
+                        has_bell: false,
+                        custom_name: Some(host),
+                        doc: None,
+                    });
+                    self.active_tab = at;
+                    self.resize_active_layout();
+                    self.dirty = true;
+                    return;
+                },
+                Err(err)
+                    if err
+                        .downcast_ref::<std::io::Error>()
+                        .is_some_and(|err| err.kind() == std::io::ErrorKind::Unsupported) =>
+                {
+                    log::info!("直连 SSH 暂无可用认证方式，回退到兼容连接路径");
+                },
+                Err(err) => {
+                    error!("创建直连 SSH Pane 失败: {err}");
+                    return;
+                },
+            }
+        }
+
         let Ok(exe) = std::env::current_exe() else {
             error!("Cannot locate nebula.exe for the SSH AskPass helper");
             return;

nebula_terminal/src/event_loop.rs

@@ -28,6 +28,83 @@ pub(crate) const READ_BUFFER_SIZE: usize = 0x10_0000;
 /// Max bytes to read from the PTY while the terminal is locked.
 const MAX_LOCKED_READ: usize = u16::MAX as usize;
 
+/// 本地 PTY 与远端传输共用的有状态终端字节流处理器。
+/// OSC 提取必须紧贴 VT 解析，避免两类会话产生不同的目录、命令和图片状态。
+#[derive(Default)]
+pub struct StreamProcessor {
+    parser: ansi::Processor,
+    cwd_sniffer: crate::osc_cwd::CwdSniffer,
+    window_size: Option<WindowSize>,
+}
+
+impl StreamProcessor {
+    pub fn resize(&mut self, window_size: WindowSize) {
+        self.window_size = Some(window_size);
+    }
+
+    pub fn next_sync_timeout(&self) -> Option<Instant> {
+        self.parser.sync_timeout().sync_timeout()
+    }
+
+    pub fn sync_bytes_count(&self) -> usize {
+        self.parser.sync_bytes_count()
+    }
+
+    pub fn stop_sync<U: EventListener>(&mut self, terminal: &mut Term<U>) {
+        self.parser.stop_sync(terminal);
+    }
+
+    pub fn feed<U: EventListener>(
+        &mut self,
+        terminal: &mut Term<U>,
+        event_proxy: &U,
+        bytes: &[u8],
+    ) {
+        let osc_events = self.cwd_sniffer.feed(bytes);
+        let mut latest_cwd = None;
+        let mut advanced = 0;
+        for (offset, event) in osc_events {
+            match event {
+                OscEvent::Cwd(cwd) => latest_cwd = Some(cwd),
+                OscEvent::CommandStart => event_proxy.send_event(Event::CommandStart),
+                OscEvent::CommandDone => event_proxy.send_event(Event::CommandDone),
+                OscEvent::Notify(text) => event_proxy.send_event(Event::Notify(text)),
+                OscEvent::PromptMark => {
+                    self.parser.advance(terminal, &bytes[advanced..offset]);
+                    advanced = offset;
+                    terminal.nebula_add_prompt_mark();
+                },
+                OscEvent::InlineImage { png, width, height } => {
+                    self.parser.advance(terminal, &bytes[advanced..offset]);
+                    advanced = offset;
+                    let (cell_w, cell_h) = self.window_size.map_or((9.0, 20.0), |ws| {
+                        (f32::from(ws.cell_width), f32::from(ws.cell_height))
+                    });
+                    let max_w = terminal.columns() as f32 * cell_w;
+                    let scale = (max_w / width as f32).min(1.0);
+                    let disp_w = width as f32 * scale;
+                    let disp_h = height as f32 * scale;
+                    let rows = (disp_h / cell_h).ceil().max(1.0) as usize;
+                    let abs_line = terminal.nebula_cursor_abs_line();
+                    for _ in 0..=rows {
+                        self.parser.advance(terminal, b"\r\n");
+                    }
+                    event_proxy.send_event(Event::InlineImage {
+                        png: std::sync::Arc::new(png),
+                        abs_line,
+                        width: disp_w,
+                        height: disp_h,
+                    });
+                },
+            }
+        }
+        self.parser.advance(terminal, &bytes[advanced..]);
+        if let Some(cwd) = latest_cwd {
+            event_proxy.send_event(Event::CwdReport(cwd));
+        }
+    }
+}
+
 /// Messages that may be sent to the `EventLoop`.
 #[derive(Debug)]
 pub enum Msg {
@@ -95,7 +172,7 @@ where
             match msg {
                 Msg::Input(input) => state.write_list.push_back(input),
                 Msg::Resize(window_size) => {
-                    state.window_size = Some(window_size);
+                    state.stream.resize(window_size);
                     self.pty.on_resize(window_size);
                 },
                 Msg::Shutdown => return false,
@@ -167,70 +244,7 @@ where
                 writer.write_all(&buf[..unprocessed]).unwrap();
             }
 
-            // Tee the bytes through the OSC sniffer first: vte drops OSC 7 /
-            // 9;9 (cwd) and 133;A (prompt marks). Prompt marks must be applied
-            // while the grid cursor sits exactly at the mark's byte position,
-            // so `parser.advance` is split at each mark offset.
-            let osc_events = state.cwd_sniffer.feed(&buf[..unprocessed]);
-            let mut latest_cwd = None;
-            let mut advanced = 0;
-            for (offset, event) in osc_events {
-                match event {
-                    OscEvent::Cwd(cwd) => latest_cwd = Some(cwd),
-                    // No cursor semantics — just relay upstream.
-                    OscEvent::CommandStart => self.event_proxy.send_event(Event::CommandStart),
-                    OscEvent::CommandDone => self.event_proxy.send_event(Event::CommandDone),
-                    OscEvent::Notify(text) => self.event_proxy.send_event(Event::Notify(text)),
-                    OscEvent::PromptMark => {
-                        state.parser.advance(&mut **terminal, &buf[advanced..offset]);
-                        advanced = offset;
-                        terminal.nebula_add_prompt_mark();
-                    },
-                    OscEvent::InlineImage { png, width, height } => {
-                        state.parser.advance(&mut **terminal, &buf[advanced..offset]);
-                        advanced = offset;
-
-                        // Convert image pixels into grid rows via the cached
-                        // cell size (default matches a typical 14px mono font
-                        // until the first resize lands, which in practice
-                        // happens before any image can arrive).
-                        let (cell_w, cell_h) = state
-                            .window_size
-                            .map_or((9.0, 20.0), |ws| {
-                                (f32::from(ws.cell_width), f32::from(ws.cell_height))
-                            });
-                        let max_w = terminal.columns() as f32 * cell_w;
-                        let scale = (max_w / width as f32).min(1.0);
-                        let disp_w = width as f32 * scale;
-                        let disp_h = height as f32 * scale;
-                        let rows = (disp_h / cell_h).ceil().max(1.0) as usize;
-
-                        // Anchor at the current cursor row, then feed real
-                        // newlines through the parser so the image's rows
-                        // exist in the grid (scrolling included) and the
-                        // prompt continues below the picture.
-                        let abs_line = terminal.nebula_cursor_abs_line();
-                        for _ in 0..=rows {
-                            state.parser.advance(&mut **terminal, b"\r\n");
-                        }
-
-                        self.event_proxy.send_event(Event::InlineImage {
-                            png: std::sync::Arc::new(png),
-                            abs_line,
-                            width: disp_w,
-                            height: disp_h,
-                        });
-                    },
-                }
-            }
-
-            // Parse the remaining bytes (or, without marks, the whole chunk).
-            state.parser.advance(&mut **terminal, &buf[advanced..unprocessed]);
-
-            // New tabs/splits inherit the newest reported directory.
-            if let Some(cwd) = latest_cwd {
-                self.event_proxy.send_event(Event::CwdReport(cwd));
-            }
+            state.stream.feed(&mut **terminal, &self.event_proxy, &buf[..unprocessed]);
 
             processed += unprocessed;
             unprocessed = 0;
@@ -242,7 +256,7 @@ where
         }
 
         // Queue terminal redraw unless all processed bytes were synchronized.
-        if state.parser.sync_bytes_count() < processed && processed > 0 {
+        if state.stream.sync_bytes_count() < processed && processed > 0 {
             self.event_proxy.send_event(Event::Wakeup);
         }
 
@@ -305,9 +319,10 @@ where
 
             'event_loop: loop {
                 // Wakeup the event loop when a synchronized update timeout was reached.
-                let handler = state.parser.sync_timeout();
-                let timeout =
-                    handler.sync_timeout().map(|st| st.saturating_duration_since(Instant::now()));
+                let timeout = state
+                    .stream
+                    .next_sync_timeout()
+                    .map(|st| st.saturating_duration_since(Instant::now()));
 
                 events.clear();
                 if let Err(err) = self.poll.wait(&mut events, timeout) {
@@ -322,7 +337,7 @@ where
 
                 // Handle synchronized update timeout.
                 if events.is_empty() && self.rx.peek().is_none() {
-                    state.parser.stop_sync(&mut *self.terminal.lock());
+                    state.stream.stop_sync(&mut *self.terminal.lock());
                     self.event_proxy.send_event(Event::Wakeup);
                     continue;
                 }
@@ -466,6 +481,12 @@ pub struct EventLoopSender {
 }
 
 impl EventLoopSender {
+    /// 为非 PTY 传输创建消息通道，继续复用应用既有的输入、缩放和关闭协议。
+    pub fn standalone() -> io::Result<(Self, Receiver<Msg>)> {
+        let (sender, receiver) = mpsc::channel();
+        Ok((Self { sender, poller: Arc::new(Poller::new()?) }, receiver))
+    }
+
     pub fn send(&self, msg: Msg) -> Result<(), EventLoopSendError> {
         self.sender.send(msg).map_err(EventLoopSendError::Send)?;
         self.poller.notify().map_err(EventLoopSendError::Io)
@@ -489,12 +510,7 @@ impl EventLoopSender {
 pub struct State {
     write_list: VecDeque<Cow<'static, [u8]>>,
     writing: Option<Writing>,
-    parser: ansi::Processor,
-    /// Tees OSC 7 / 9;9 / 133;A / 1337 events out of the byte stream.
-    cwd_sniffer: crate::osc_cwd::CwdSniffer,
-    /// Latest PTY size, cached off `Msg::Resize`: inline images need the cell
-    /// pixel size to convert image pixels into grid rows.
-    window_size: Option<WindowSize>,
+    stream: StreamProcessor,
 }
 
 impl State {