diff --git a/crates/okena-core/src/client/connection.rs b/crates/okena-core/src/client/connection.rs index 1687c5ef0..00c54dd49 100644 --- a/crates/okena-core/src/client/connection.rs +++ b/crates/okena-core/src/client/connection.rs @@ -25,7 +25,8 @@ pub trait ConnectionHandler: Send + Sync + 'static { ws_sender: async_channel::Sender, ); /// Binary PTY output arrived — route to the terminal's emulator. - fn on_terminal_output(&self, prefixed_id: &str, data: &[u8]); + /// `acked_input_seq` is the server's echo of the latest input sequence (v2 protocol). + fn on_terminal_output(&self, prefixed_id: &str, data: &[u8], acked_input_seq: Option); /// Terminal removed — clean up platform terminal object. fn remove_terminal(&self, prefixed_id: &str); /// Pre-resize a terminal's grid to match the server's dimensions. @@ -614,17 +615,22 @@ impl RemoteClient { let writer_handle = tokio::spawn(async move { while let Ok(msg) = ws_rx_clone.recv().await { // For SendText, prefer binary frame when stream_id is known - if let WsClientMessage::SendText { terminal_id, text } = &msg { + if let WsClientMessage::SendText { terminal_id, text, input_seq } = &msg { let stream_id = stream_map_for_writer .read() .ok() .and_then(|m| m.get(terminal_id).copied()); if let Some(sid) = stream_id { - let frame = crate::ws::build_binary_frame( - crate::ws::FRAME_TYPE_INPUT, - sid, - text.as_bytes(), - ); + // Use v2 frame if we have a non-zero input_seq (prediction active) + let frame = if *input_seq > 0 { + crate::ws::build_input_frame_v2(sid, *input_seq, text.as_bytes()) + } else { + crate::ws::build_binary_frame( + crate::ws::FRAME_TYPE_INPUT, + sid, + text.as_bytes(), + ) + }; if let Err(e) = futures::SinkExt::send( &mut ws_write, tungstenite::Message::Binary(frame.into()), @@ -639,7 +645,7 @@ impl RemoteClient { } let json = match &msg { - WsClientMessage::SendText { terminal_id, text } => { + WsClientMessage::SendText { terminal_id, text, .. } => { serde_json::json!({ "type": "send_text", "terminal_id": terminal_id, @@ -694,16 +700,22 @@ impl RemoteClient { loop { match futures::StreamExt::next(&mut ws_read).await { Some(Ok(tungstenite::Message::Binary(data))) => { - // Generic binary frame: [proto:1][type:1][stream_id:4 BE][payload...] - if let Some((frame_type, stream_id, payload)) = - crate::ws::parse_binary_frame(&data) + // Parse binary frame (v1 or v2) + if let Some((frame_type, stream_id, payload, acked_seq)) = + crate::ws::parse_binary_frame_any(&data) { match frame_type { - crate::ws::FRAME_TYPE_PTY | crate::ws::FRAME_TYPE_SNAPSHOT => { - // Route PTY output or snapshot to the correct terminal + crate::ws::FRAME_TYPE_PTY => { + if let Some(remote_tid) = reverse_stream_map.get(&stream_id) { + let prefixed = make_prefixed_id(&config_id, remote_tid); + handler_clone.on_terminal_output(&prefixed, payload, acked_seq); + } + } + crate::ws::FRAME_TYPE_SNAPSHOT => { + // Snapshots implicitly ack everything if let Some(remote_tid) = reverse_stream_map.get(&stream_id) { let prefixed = make_prefixed_id(&config_id, remote_tid); - handler_clone.on_terminal_output(&prefixed, payload); + handler_clone.on_terminal_output(&prefixed, payload, Some(u64::MAX)); } } _ => { diff --git a/crates/okena-core/src/client/types.rs b/crates/okena-core/src/client/types.rs index 9266d5d65..392e83066 100644 --- a/crates/okena-core/src/client/types.rs +++ b/crates/okena-core/src/client/types.rs @@ -22,8 +22,9 @@ pub enum ConnectionStatus { /// Messages sent from the UI thread to the WebSocket writer task. #[derive(Debug)] pub enum WsClientMessage { - /// Send text input to a remote terminal - SendText { terminal_id: String, text: String }, + /// Send text input to a remote terminal. + /// `input_seq` is the latest prediction sequence (0 if not predicting). + SendText { terminal_id: String, text: String, input_seq: u64 }, /// Resize a remote terminal Resize { terminal_id: String, @@ -111,6 +112,7 @@ mod tests { let msg = WsClientMessage::SendText { terminal_id: "t1".to_string(), text: "hello".to_string(), + input_seq: 0, }; let debug = format!("{:?}", msg); assert!(debug.contains("SendText")); diff --git a/crates/okena-core/src/ws.rs b/crates/okena-core/src/ws.rs index 2afdbb306..e222511bf 100644 --- a/crates/okena-core/src/ws.rs +++ b/crates/okena-core/src/ws.rs @@ -64,6 +64,7 @@ pub enum WsOutbound { // ── Binary frame protocol ────────────────────────────────────────────────── pub const PROTO_VERSION: u8 = 1; +pub const PROTO_VERSION_2: u8 = 2; pub const FRAME_TYPE_PTY: u8 = 1; // server → client: live PTY output pub const FRAME_TYPE_SNAPSHOT: u8 = 2; // server → client: full screen redraw pub const FRAME_TYPE_INPUT: u8 = 3; // client → server: terminal input @@ -105,6 +106,83 @@ pub fn build_pty_frame(stream_id: u32, data: &[u8]) -> Vec { build_binary_frame(FRAME_TYPE_PTY, stream_id, data) } +// ── V2 binary frame protocol (with input sequence numbers) ───────────────── + +/// Build a v2 INPUT frame: [proto=2][type=3][stream_id:4 BE][input_seq:8 BE][payload...] +pub fn build_input_frame_v2(stream_id: u32, input_seq: u64, data: &[u8]) -> Vec { + let mut frame = Vec::with_capacity(14 + data.len()); + frame.push(PROTO_VERSION_2); + frame.push(FRAME_TYPE_INPUT); + frame.extend_from_slice(&stream_id.to_be_bytes()); + frame.extend_from_slice(&input_seq.to_be_bytes()); + frame.extend_from_slice(data); + frame +} + +/// Parse a v2 INPUT frame. +/// Returns (stream_id, input_seq, payload) or None if invalid. +pub fn parse_input_frame_v2(data: &[u8]) -> Option<(u32, u64, &[u8])> { + if data.len() < 14 || data[0] != PROTO_VERSION_2 || data[1] != FRAME_TYPE_INPUT { + return None; + } + let stream_id = u32::from_be_bytes([data[2], data[3], data[4], data[5]]); + let input_seq = u64::from_be_bytes([data[6], data[7], data[8], data[9], data[10], data[11], data[12], data[13]]); + Some((stream_id, input_seq, &data[14..])) +} + +/// Build a v2 PTY frame: [proto=2][type=1][stream_id:4 BE][last_input_seq:8 BE][payload...] +pub fn build_pty_frame_v2(stream_id: u32, last_input_seq: u64, data: &[u8]) -> Vec { + let mut frame = Vec::with_capacity(14 + data.len()); + frame.push(PROTO_VERSION_2); + frame.push(FRAME_TYPE_PTY); + frame.extend_from_slice(&stream_id.to_be_bytes()); + frame.extend_from_slice(&last_input_seq.to_be_bytes()); + frame.extend_from_slice(data); + frame +} + +/// Parse a v2 PTY frame. +/// Returns (stream_id, last_input_seq, payload) or None if invalid. +pub fn parse_pty_frame_v2(data: &[u8]) -> Option<(u32, u64, &[u8])> { + if data.len() < 14 || data[0] != PROTO_VERSION_2 || data[1] != FRAME_TYPE_PTY { + return None; + } + let stream_id = u32::from_be_bytes([data[2], data[3], data[4], data[5]]); + let last_input_seq = u64::from_be_bytes([data[6], data[7], data[8], data[9], data[10], data[11], data[12], data[13]]); + Some((stream_id, last_input_seq, &data[14..])) +} + +/// Parse a binary frame that could be v1 or v2. +/// Returns (frame_type, stream_id, payload, acked_input_seq). +/// For v1 frames, acked_input_seq is None. For v2 PTY/INPUT frames, it's Some(seq). +pub fn parse_binary_frame_any(data: &[u8]) -> Option<(u8, u32, &[u8], Option)> { + if data.len() < 6 { + return None; + } + let proto = data[0]; + let frame_type = data[1]; + + if proto == PROTO_VERSION_2 && (frame_type == FRAME_TYPE_PTY || frame_type == FRAME_TYPE_INPUT) { + // v2 frame with input_seq + if data.len() < 14 { + return None; + } + let stream_id = u32::from_be_bytes([data[2], data[3], data[4], data[5]]); + let seq = u64::from_be_bytes([data[6], data[7], data[8], data[9], data[10], data[11], data[12], data[13]]); + Some((frame_type, stream_id, &data[14..], Some(seq))) + } else if proto == PROTO_VERSION { + // v1 frame + let stream_id = u32::from_be_bytes([data[2], data[3], data[4], data[5]]); + Some((frame_type, stream_id, &data[6..], None)) + } else if proto == PROTO_VERSION_2 && frame_type == FRAME_TYPE_SNAPSHOT { + // v2 snapshot — same format as v1 (no seq) + let stream_id = u32::from_be_bytes([data[2], data[3], data[4], data[5]]); + Some((frame_type, stream_id, &data[6..], None)) + } else { + None + } +} + #[cfg(test)] mod tests { use super::*; @@ -271,4 +349,69 @@ mod tests { assert_eq!(parsed_data, payload.as_bytes()); } } + + // ── V2 frame tests ────────────────────────────────────────────────── + + #[test] + fn v2_input_frame_round_trip() { + let frame = build_input_frame_v2(42, 12345, b"hello"); + let (sid, seq, payload) = parse_input_frame_v2(&frame).unwrap(); + assert_eq!(sid, 42); + assert_eq!(seq, 12345); + assert_eq!(payload, b"hello"); + } + + #[test] + fn v2_pty_frame_round_trip() { + let frame = build_pty_frame_v2(7, 99, b"output"); + let (sid, seq, payload) = parse_pty_frame_v2(&frame).unwrap(); + assert_eq!(sid, 7); + assert_eq!(seq, 99); + assert_eq!(payload, b"output"); + } + + #[test] + fn v1_frames_still_parse() { + let frame = build_binary_frame(FRAME_TYPE_PTY, 1, b"data"); + let (ft, sid, payload) = parse_binary_frame(&frame).unwrap(); + assert_eq!(ft, FRAME_TYPE_PTY); + assert_eq!(sid, 1); + assert_eq!(payload, b"data"); + } + + #[test] + fn v2_parser_rejects_short_frames() { + assert!(parse_input_frame_v2(&[2, 3, 0, 0, 0, 0, 0, 0]).is_none()); + assert!(parse_pty_frame_v2(&[2, 1, 0, 0, 0]).is_none()); + } + + #[test] + fn parse_binary_frame_any_v1() { + let frame = build_binary_frame(FRAME_TYPE_PTY, 5, b"test"); + let (ft, sid, payload, seq) = parse_binary_frame_any(&frame).unwrap(); + assert_eq!(ft, FRAME_TYPE_PTY); + assert_eq!(sid, 5); + assert_eq!(payload, b"test"); + assert!(seq.is_none()); + } + + #[test] + fn parse_binary_frame_any_v2_pty() { + let frame = build_pty_frame_v2(5, 100, b"test"); + let (ft, sid, payload, seq) = parse_binary_frame_any(&frame).unwrap(); + assert_eq!(ft, FRAME_TYPE_PTY); + assert_eq!(sid, 5); + assert_eq!(payload, b"test"); + assert_eq!(seq, Some(100)); + } + + #[test] + fn parse_binary_frame_any_v2_input() { + let frame = build_input_frame_v2(3, 50, b"keys"); + let (ft, sid, payload, seq) = parse_binary_frame_any(&frame).unwrap(); + assert_eq!(ft, FRAME_TYPE_INPUT); + assert_eq!(sid, 3); + assert_eq!(payload, b"keys"); + assert_eq!(seq, Some(50)); + } } diff --git a/mobile/native/src/api/state.rs b/mobile/native/src/api/state.rs index d8d76ad57..a290d5d14 100644 --- a/mobile/native/src/api/state.rs +++ b/mobile/native/src/api/state.rs @@ -81,6 +81,7 @@ pub async fn send_special_key( WsClientMessage::SendText { terminal_id, text, + input_seq: 0, }, ); Ok(()) diff --git a/mobile/native/src/api/terminal.rs b/mobile/native/src/api/terminal.rs index 70d40a199..8f07e821f 100644 --- a/mobile/native/src/api/terminal.rs +++ b/mobile/native/src/api/terminal.rs @@ -169,6 +169,7 @@ pub async fn send_text(conn_id: String, terminal_id: String, text: String) -> an WsClientMessage::SendText { terminal_id, text, + input_seq: 0, }, ); Ok(()) diff --git a/mobile/native/src/client/handler.rs b/mobile/native/src/client/handler.rs index d4402f1bd..d1d4b0354 100644 --- a/mobile/native/src/client/handler.rs +++ b/mobile/native/src/client/handler.rs @@ -47,7 +47,7 @@ impl ConnectionHandler for MobileConnectionHandler { .insert(prefixed_id.to_string(), holder); } - fn on_terminal_output(&self, prefixed_id: &str, data: &[u8]) { + fn on_terminal_output(&self, prefixed_id: &str, data: &[u8], _acked_input_seq: Option) { *self.last_activity.lock() = Instant::now(); if let Some(holder) = self.terminals.read().get(prefixed_id) { holder.process_output(data); @@ -120,7 +120,7 @@ mod tests { let (tx, _rx) = async_channel::bounded(1); handler.create_terminal("conn1", "t1", "remote:conn1:t1", tx); - handler.on_terminal_output("remote:conn1:t1", b"hello"); + handler.on_terminal_output("remote:conn1:t1", b"hello", None); let terminals = handler.terminals().read(); let holder = terminals.get("remote:conn1:t1").unwrap(); diff --git a/src/action_dispatch.rs b/src/action_dispatch.rs index 34f1ed6f0..4adab8059 100644 --- a/src/action_dispatch.rs +++ b/src/action_dispatch.rs @@ -119,14 +119,36 @@ impl ActionDispatcher { }); return; } + ActionRequest::CreateTerminal { project_id } => { + // Optimistically set focus for the expected new terminal path. + // add_terminal() wraps root in Split{[old, new]}, so new terminal + // will be at path [1]. If no layout existed, it's at root []. + let pid = project_id.clone(); + workspace.update(cx, |ws, cx| { + let path = if ws.project(&pid) + .and_then(|p| p.layout.as_ref()) + .is_some() + { + vec![1] + } else { + vec![] + }; + ws.set_focused_terminal(pid, path, cx); + }); + // Don't return — action proceeds to be sent to server below + } _ => {} } + log::info!("[dispatch] sending remote action to server"); let action = strip_remote_ids(action, connection_id); let cid = connection_id.clone(); manager.update(cx, |rm, cx| { + log::info!("[dispatch] inside manager.update, calling send_action"); rm.send_action(&cid, action, cx); + log::info!("[dispatch] send_action returned"); }); + log::info!("[dispatch] manager.update completed"); } } } diff --git a/src/app/headless.rs b/src/app/headless.rs index dbdf87c5b..3faee61a8 100644 --- a/src/app/headless.rs +++ b/src/app/headless.rs @@ -254,14 +254,24 @@ impl HeadlessApp { Err(_) => break, }; + log::info!("[headless] received bridge message: {:?}", msg.command); let result = match msg.command { RemoteCommand::Action(action) => { - cx.update(|cx| { - workspace.update(cx, |ws, cx| { - execute_action(action, ws, &*backend, &terminals, cx) - .into_command_result() - }) - }) + log::info!("[headless] executing action: {:?}", action); + let r = cx.update(|cx| { + log::info!("[headless] inside cx.update"); + let r = workspace.update(cx, |ws, cx| { + log::info!("[headless] inside workspace.update, calling execute_action"); + let r = execute_action(action, ws, &*backend, &terminals, cx) + .into_command_result(); + log::info!("[headless] execute_action returned"); + r + }); + log::info!("[headless] workspace.update returned"); + r + }); + log::info!("[headless] cx.update returned"); + r } RemoteCommand::GetState => { cx.update(|cx| { diff --git a/src/elements/terminal_element.rs b/src/elements/terminal_element.rs index 86c223344..d692d7464 100644 --- a/src/elements/terminal_element.rs +++ b/src/elements/terminal_element.rs @@ -623,6 +623,69 @@ impl Element for TerminalElement { } }); + // Phase 4.5: Paint prediction overlay (remote input echo) + if self.terminal.is_remote() { + let overlay_cells = self.terminal.overlay_cells(); + if !overlay_cells.is_empty() { + let origin = bounds.origin; + for cell in &overlay_cells { + // We need to determine the visual line for this cell + // overlay cells store buffer-coordinate rows; to get visual line + // we need the display_offset which was inside the with_content closure. + // For simplicity, read it again (cheap lock). + let display_offset = self.terminal.display_offset() as i32; + let visual_line = cell.row + display_offset; + let screen_lines = self.terminal.screen_lines() as i32; + + if visual_line < 0 || visual_line >= screen_lines { + continue; + } + + let x = px((f32::from(origin.x) + cell.col as f32 * cell_width_f).floor()); + let y = px((f32::from(origin.y) + visual_line as f32 * line_height_f).floor()); + + // Paint the predicted character with foreground color + let fg_color = Hsla::from(Rgba { + r: 0.7, + g: 0.7, + b: 0.7, + a: 0.9, + }); + + let mut char_buf = [0u8; 4]; + let char_str = cell.character.encode_utf8(&mut char_buf); + let text_run = TextRun { + len: char_str.len(), + font: state.font.clone(), + color: fg_color, + background_color: None, + underline: Some(UnderlineStyle { + color: Some(Hsla::from(Rgba { + r: 0.5, + g: 0.5, + b: 0.5, + a: 0.4, + })), + thickness: px(1.0), + wavy: false, + }), + strikethrough: None, + }; + + let char_string: SharedString = char_str.to_string().into(); + let line = window + .text_system() + .shape_line( + char_string, + font_size, + &[text_run], + None, + ); + line.paint(point(x, y), line_height, TextAlign::Left, None, window, cx).ok(); + } + } + } + // Phase 5: Paint fog overlay for unfocused terminals // Uses the unfocused bg color at partial opacity to wash out text, // creating a subtle "in the fog" effect. Alpha-blending the same color diff --git a/src/elements/terminal_input.rs b/src/elements/terminal_input.rs index 3d7d5472b..2539ee509 100644 --- a/src/elements/terminal_input.rs +++ b/src/elements/terminal_input.rs @@ -35,6 +35,15 @@ impl TerminalInputHandler { return; } + // Predict printable chars for remote terminals + if self.terminal.is_remote() { + for c in filtered.chars() { + if c.is_ascii_graphic() || c == ' ' || (!c.is_control() && !c.is_ascii()) { + self.terminal.predict_char(c); + } + } + } + // Fast path: no control characters, send entire string at once if !filtered.chars().any(|c| matches!(c, '\n' | '\r' | '\u{8}')) { self.terminal.send_input(&filtered); diff --git a/src/remote/routes/stream.rs b/src/remote/routes/stream.rs index 0eb9476f0..6f66d1c0e 100644 --- a/src/remote/routes/stream.rs +++ b/src/remote/routes/stream.rs @@ -1,8 +1,8 @@ use crate::remote::bridge::{BridgeMessage, CommandResult, RemoteCommand}; use crate::remote::routes::AppState; use crate::remote::types::{ - ActionRequest, WsInbound, WsOutbound, build_binary_frame, build_pty_frame, parse_binary_frame, - FRAME_TYPE_INPUT, FRAME_TYPE_SNAPSHOT, + ActionRequest, WsInbound, WsOutbound, build_binary_frame, build_pty_frame, build_pty_frame_v2, + parse_binary_frame_any, FRAME_TYPE_INPUT, FRAME_TYPE_SNAPSHOT, }; use axum::extract::ws::{Message, WebSocket}; use axum::extract::{Query, State, WebSocketUpgrade}; @@ -65,6 +65,8 @@ async fn handle_ws(mut socket: WebSocket, state: AppState, query_token: Option = HashMap::new(); let mut reverse_stream_map: HashMap = HashMap::new(); let mut next_stream_id: u32 = 1; + // Track latest input sequence per stream_id (for v2 protocol echo) + let mut last_input_seq: HashMap = HashMap::new(); // Subscribe to state_version changes (immediate push, no polling) let mut state_rx = state.state_version.subscribe(); @@ -177,8 +179,12 @@ async fn handle_ws(mut socket: WebSocket, state: AppState, query_token: Option { - // Binary input frame from client - if let Some((FRAME_TYPE_INPUT, stream_id, payload)) = parse_binary_frame(&data) { + // Binary input frame from client (v1 or v2) + if let Some((FRAME_TYPE_INPUT, stream_id, payload, seq)) = parse_binary_frame_any(&data) { + // Track input sequence for v2 echo + if let Some(s) = seq { + last_input_seq.insert(stream_id, s); + } if let Some(terminal_id) = reverse_stream_map.get(&stream_id) { let text = String::from_utf8_lossy(payload).to_string(); let (reply_tx, reply_rx) = tokio::sync::oneshot::channel(); @@ -205,7 +211,12 @@ async fn handle_ws(mut socket: WebSocket, state: AppState, query_token: Option { if subscribed_ids.contains(&event.terminal_id) { if let Some(&stream_id) = stream_id_map.get(&event.terminal_id) { - let frame = build_pty_frame(stream_id, &event.data); + // Use v2 frame (with input_seq echo) if client sent v2 input + let frame = if let Some(&seq) = last_input_seq.get(&stream_id) { + build_pty_frame_v2(stream_id, seq, &event.data) + } else { + build_pty_frame(stream_id, &event.data) + }; if socket.send(Message::Binary(frame.into())).await.is_err() { break; } diff --git a/src/remote/types.rs b/src/remote/types.rs index 0779e15f8..cb83836c3 100644 --- a/src/remote/types.rs +++ b/src/remote/types.rs @@ -7,6 +7,8 @@ pub use okena_core::api::{ pub use okena_core::ws::{ WsInbound, WsOutbound, build_binary_frame, build_pty_frame, parse_binary_frame, parse_pty_frame, FRAME_TYPE_INPUT, FRAME_TYPE_PTY, FRAME_TYPE_SNAPSHOT, PROTO_VERSION, + PROTO_VERSION_2, build_input_frame_v2, parse_input_frame_v2, build_pty_frame_v2, + parse_pty_frame_v2, parse_binary_frame_any, }; use crate::workspace::state::LayoutNode; diff --git a/src/remote_client/backend.rs b/src/remote_client/backend.rs index d6feb048f..0256a1b01 100644 --- a/src/remote_client/backend.rs +++ b/src/remote_client/backend.rs @@ -5,6 +5,7 @@ use anyhow::Result; use okena_core::client::{make_prefixed_id, strip_prefix, WsClientMessage}; use std::path::PathBuf; use std::sync::Arc; +use std::sync::atomic::{AtomicU64, Ordering}; /// Transport implementation for remote terminals. /// @@ -14,14 +15,18 @@ use std::sync::Arc; pub struct RemoteTransport { pub(crate) ws_tx: async_channel::Sender, pub(crate) connection_id: String, + /// Latest input prediction sequence number (read by writer task). + pub(crate) latest_input_seq: Arc, } impl TerminalTransport for RemoteTransport { fn send_input(&self, terminal_id: &str, data: &[u8]) { let remote_id = strip_prefix(terminal_id, &self.connection_id); + let seq = self.latest_input_seq.load(Ordering::Relaxed); let _ = self.ws_tx.try_send(WsClientMessage::SendText { terminal_id: remote_id, text: String::from_utf8_lossy(data).to_string(), + input_seq: seq, }); } diff --git a/src/remote_client/connection.rs b/src/remote_client/connection.rs index 13b9ddee2..f17e072e3 100644 --- a/src/remote_client/connection.rs +++ b/src/remote_client/connection.rs @@ -11,6 +11,7 @@ use okena_core::client::{ use std::collections::HashMap; use std::sync::Arc; +use std::sync::atomic::AtomicU64; /// Desktop-specific handler that creates `Terminal` objects and manages the registry. pub struct DesktopConnectionHandler { @@ -31,22 +32,29 @@ impl ConnectionHandler for DesktopConnectionHandler { prefixed_id: &str, ws_sender: async_channel::Sender, ) { + // Create shared atomic so Terminal and RemoteTransport share the same input_seq + let shared_seq = Arc::new(AtomicU64::new(0)); let transport = Arc::new(RemoteTransport { ws_tx: ws_sender, connection_id: connection_id.to_string(), + latest_input_seq: shared_seq.clone(), }); - let terminal = Arc::new(Terminal::new( + let terminal = Arc::new(Terminal::new_remote_with_seq( prefixed_id.to_string(), TerminalSize::default(), transport, String::new(), + shared_seq, )); self.terminals.lock().insert(prefixed_id.to_string(), terminal); } - fn on_terminal_output(&self, prefixed_id: &str, data: &[u8]) { + fn on_terminal_output(&self, prefixed_id: &str, data: &[u8], acked_input_seq: Option) { if let Some(terminal) = self.terminals.lock().get(prefixed_id) { terminal.process_output(data); + if let Some(seq) = acked_input_seq { + terminal.ack_predictions(seq); + } } } @@ -149,6 +157,7 @@ impl RemoteConnection { let transport = Arc::new(RemoteTransport { ws_tx, connection_id: self.config().id.clone(), + latest_input_seq: Arc::new(AtomicU64::new(0)), }); Arc::new(RemoteBackend::new(transport, self.config().id.clone())) } diff --git a/src/remote_client/manager.rs b/src/remote_client/manager.rs index f5a2b7c78..5e712d438 100644 --- a/src/remote_client/manager.rs +++ b/src/remote_client/manager.rs @@ -215,6 +215,7 @@ impl RemoteConnectionManager { self.runtime.spawn(async move { let url = format!("http://{}:{}/v1/actions", host, port); + log::info!("[send_action] POST {} action={:?}", url, action); let client = reqwest::Client::new(); let result = client .post(&url) @@ -223,6 +224,7 @@ impl RemoteConnectionManager { .timeout(std::time::Duration::from_secs(10)) .send() .await; + log::info!("[send_action] POST completed, result={}", result.is_ok()); match result { Ok(resp) if resp.status().is_success() => { diff --git a/src/terminal/input_overlay.rs b/src/terminal/input_overlay.rs new file mode 100644 index 000000000..e35695f47 --- /dev/null +++ b/src/terminal/input_overlay.rs @@ -0,0 +1,428 @@ +//! Optimistic input prediction for remote terminals (Mosh-style local echo). +//! +//! Predicted characters render instantly on the client with a visual hint, +//! then get reconciled when the server confirms or invalidates them. +//! Pure logic — no GPUI dependency. + +use std::collections::VecDeque; +use std::time::{Duration, Instant}; + +/// A single predicted character cell. +#[derive(Clone, Debug)] +pub struct PredictedCell { + pub col: usize, + pub row: i32, + pub character: char, + pub width: u8, + pub input_seq: u64, + pub created_at: Instant, +} + +/// State of the prediction engine. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum PredictionState { + /// Actively predicting characters. + Active, + /// Temporarily paused after a cursor jump (clears after 500ms of stability). + Tentative, + /// Disabled (not used currently, reserved for future use). + Disabled, +} + +/// Prediction state machine for a single terminal. +pub struct InputOverlay { + predictions: VecDeque, + predicted_cursor: Option<(usize, i32)>, + last_server_cursor: (usize, i32), + cursor_row_stable_count: u8, + state: PredictionState, + next_input_seq: u64, + last_acked_seq: u64, + prediction_timeout: Duration, + tentative_since: Option, + epoch: u64, + cols: usize, +} + +/// Duration after which unacked predictions are garbage-collected. +const DEFAULT_PREDICTION_TIMEOUT: Duration = Duration::from_millis(150); + +/// Duration after which Tentative state clears if cursor is stable. +const TENTATIVE_CLEAR_DURATION: Duration = Duration::from_millis(500); + +/// Minimum server frames with stable cursor row before predictions are allowed. +const MIN_STABLE_FRAMES: u8 = 3; + +impl InputOverlay { + pub fn new() -> Self { + Self { + predictions: VecDeque::new(), + predicted_cursor: None, + last_server_cursor: (0, 0), + cursor_row_stable_count: 0, + state: PredictionState::Active, + next_input_seq: 1, + last_acked_seq: 0, + prediction_timeout: DEFAULT_PREDICTION_TIMEOUT, + tentative_since: None, + epoch: 0, + cols: 80, + } + } + + /// Update the column count (call when terminal is resized). + pub fn set_cols(&mut self, cols: usize) { + self.cols = cols; + } + + /// Check whether a character should be predicted. + pub fn should_predict(&self, c: char, is_remote: bool) -> bool { + if !is_remote { + return false; + } + if self.state != PredictionState::Active { + return false; + } + if self.cursor_row_stable_count < MIN_STABLE_FRAMES { + return false; + } + // Only printable ASCII + space + non-control Unicode + c.is_ascii_graphic() || c == ' ' || (!c.is_control() && !c.is_ascii()) + } + + /// Predict a character at the current predicted cursor position. + /// Returns the assigned input sequence number, or None if prediction was skipped. + pub fn predict_char(&mut self, c: char, server_cursor: (usize, i32), cols: usize) -> Option { + self.cols = cols; + + if !self.should_predict(c, true) { + return None; + } + + let (cursor_col, cursor_row) = self.predicted_cursor.unwrap_or(server_cursor); + + // Determine character width (CJK = 2, others = 1) + let width = if is_wide_char(c) { 2u8 } else { 1u8 }; + + // Don't predict if cursor is at or past the end of line + if cursor_col >= cols || cursor_col + width as usize > cols { + return None; + } + + let seq = self.next_input_seq; + self.next_input_seq += 1; + + self.predictions.push_back(PredictedCell { + col: cursor_col, + row: cursor_row, + character: c, + width, + input_seq: seq, + created_at: Instant::now(), + }); + + // Advance predicted cursor + let new_col = cursor_col + width as usize; + if new_col >= cols { + // At end of line — stop predicting further by not advancing cursor + // The predicted_cursor is set to None-equivalent: mark that we're at the edge + self.predicted_cursor = Some((cols, cursor_row)); + } else { + self.predicted_cursor = Some((new_col, cursor_row)); + } + + Some(seq) + } + + /// Called when a server frame arrives with the latest acked sequence and cursor position. + pub fn on_server_frame(&mut self, acked_seq: u64, server_cursor: (usize, i32)) { + // Track cursor row stability + if server_cursor.1 != self.last_server_cursor.1 { + self.cursor_row_stable_count = 0; + } else { + self.cursor_row_stable_count = self.cursor_row_stable_count.saturating_add(1); + } + self.last_server_cursor = server_cursor; + + // Ack predictions up to acked_seq + if acked_seq > self.last_acked_seq { + self.last_acked_seq = acked_seq; + while let Some(front) = self.predictions.front() { + if front.input_seq <= acked_seq { + self.predictions.pop_front(); + } else { + break; + } + } + } + + // Detect cursor row mismatch (server jumped to different row than predicted) + if !self.predictions.is_empty() { + if let Some((_, predicted_row)) = self.predicted_cursor { + if server_cursor.1 != predicted_row { + self.discard_all(); + self.state = PredictionState::Tentative; + self.tentative_since = Some(Instant::now()); + return; + } + } + } + + // If no predictions remain, re-sync predicted cursor with server + if self.predictions.is_empty() { + self.predicted_cursor = None; + } + + // Handle tentative state clearing + if self.state == PredictionState::Tentative { + if let Some(since) = self.tentative_since { + if since.elapsed() >= TENTATIVE_CLEAR_DURATION + && self.cursor_row_stable_count >= MIN_STABLE_FRAMES + { + self.state = PredictionState::Active; + self.tentative_since = None; + } + } + } + } + + /// Discard all predictions and increment epoch. + pub fn discard_all(&mut self) { + self.predictions.clear(); + self.predicted_cursor = None; + self.epoch += 1; + } + + /// Remove predictions that have exceeded the timeout. + pub fn gc_expired(&mut self) { + let now = Instant::now(); + let timeout = self.prediction_timeout; + while let Some(front) = self.predictions.front() { + if now.duration_since(front.created_at) >= timeout { + self.predictions.pop_front(); + } else { + break; + } + } + if self.predictions.is_empty() { + self.predicted_cursor = None; + } + } + + /// Get the current prediction cells for rendering. + pub fn cells(&self) -> &VecDeque { + &self.predictions + } + + /// Get the predicted cursor position (if predictions are active). + pub fn predicted_cursor(&self) -> Option<(usize, i32)> { + self.predicted_cursor + } + + /// Current epoch (incremented on each discard_all). + pub fn epoch(&self) -> u64 { + self.epoch + } + + /// Current prediction state. + pub fn state(&self) -> PredictionState { + self.state + } + + /// The next sequence number that will be assigned. + pub fn next_input_seq(&self) -> u64 { + self.next_input_seq + } +} + +/// Heuristic for CJK wide characters. +fn is_wide_char(c: char) -> bool { + let cp = c as u32; + // CJK Unified Ideographs, CJK Compatibility Ideographs, Hangul Syllables, etc. + matches!(cp, + 0x1100..=0x115F | + 0x2E80..=0x303E | + 0x3041..=0x33BF | + 0x3400..=0x4DBF | + 0x4E00..=0x9FFF | + 0xA000..=0xA4CF | + 0xAC00..=0xD7AF | + 0xF900..=0xFAFF | + 0xFE30..=0xFE6F | + 0xFF01..=0xFF60 | + 0xFFE0..=0xFFE6 | + 0x20000..=0x2FFFF | + 0x30000..=0x3FFFF + ) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn make_overlay() -> InputOverlay { + let mut overlay = InputOverlay::new(); + // Pre-warm stability count + for _ in 0..MIN_STABLE_FRAMES { + overlay.on_server_frame(0, (0, 0)); + } + overlay + } + + #[test] + fn predict_char_advances_cursor() { + let mut overlay = make_overlay(); + let seq = overlay.predict_char('a', (0, 0), 80); + assert_eq!(seq, Some(1)); + assert_eq!(overlay.predicted_cursor(), Some((1, 0))); + + let seq = overlay.predict_char('b', (0, 0), 80); + assert_eq!(seq, Some(2)); + assert_eq!(overlay.predicted_cursor(), Some((2, 0))); + } + + #[test] + fn on_server_frame_acks_predictions() { + let mut overlay = make_overlay(); + overlay.predict_char('a', (0, 0), 80); + overlay.predict_char('b', (0, 0), 80); + assert_eq!(overlay.cells().len(), 2); + + // Ack first prediction + overlay.on_server_frame(1, (1, 0)); + assert_eq!(overlay.cells().len(), 1); + assert_eq!(overlay.cells()[0].character, 'b'); + + // Ack second prediction + overlay.on_server_frame(2, (2, 0)); + assert!(overlay.cells().is_empty()); + } + + #[test] + fn cursor_row_mismatch_discards_and_goes_tentative() { + let mut overlay = make_overlay(); + overlay.predict_char('a', (0, 0), 80); + assert_eq!(overlay.state(), PredictionState::Active); + + // Server cursor jumps to a different row + overlay.on_server_frame(0, (0, 5)); + assert!(overlay.cells().is_empty()); + assert_eq!(overlay.state(), PredictionState::Tentative); + } + + #[test] + fn gc_expired_removes_old_predictions() { + let mut overlay = make_overlay(); + overlay.prediction_timeout = Duration::from_millis(10); + overlay.predict_char('a', (0, 0), 80); + + // Wait for expiry + std::thread::sleep(Duration::from_millis(15)); + overlay.gc_expired(); + assert!(overlay.cells().is_empty()); + } + + #[test] + fn gc_expired_keeps_fresh() { + let mut overlay = make_overlay(); + overlay.prediction_timeout = Duration::from_secs(10); + overlay.predict_char('a', (0, 0), 80); + + overlay.gc_expired(); + assert_eq!(overlay.cells().len(), 1); + } + + #[test] + fn should_predict_rejects_control_chars() { + let overlay = make_overlay(); + assert!(!overlay.should_predict('\x03', true)); // Ctrl-C + assert!(!overlay.should_predict('\x1b', true)); // Escape + assert!(!overlay.should_predict('\n', true)); + assert!(!overlay.should_predict('\r', true)); + } + + #[test] + fn should_predict_accepts_printable() { + let overlay = make_overlay(); + assert!(overlay.should_predict('a', true)); + assert!(overlay.should_predict(' ', true)); + assert!(overlay.should_predict('Z', true)); + assert!(overlay.should_predict('1', true)); + } + + #[test] + fn should_predict_rejects_non_remote() { + let overlay = make_overlay(); + assert!(!overlay.should_predict('a', false)); + } + + #[test] + fn should_predict_rejects_tentative_state() { + let mut overlay = make_overlay(); + overlay.predict_char('a', (0, 0), 80); + // Force tentative + overlay.on_server_frame(0, (0, 5)); + assert!(!overlay.should_predict('b', true)); + } + + #[test] + fn tentative_clears_after_stability() { + let mut overlay = InputOverlay::new(); + overlay.state = PredictionState::Tentative; + overlay.tentative_since = Some(Instant::now() - Duration::from_millis(600)); + + // Pump stable frames + for _ in 0..MIN_STABLE_FRAMES + 1 { + overlay.on_server_frame(0, (0, 0)); + } + assert_eq!(overlay.state(), PredictionState::Active); + } + + #[test] + fn wide_char_prediction_advances_by_two() { + let mut overlay = make_overlay(); + // CJK character + let seq = overlay.predict_char('\u{4E00}', (0, 0), 80); + assert!(seq.is_some()); + assert_eq!(overlay.predicted_cursor(), Some((2, 0))); + assert_eq!(overlay.cells()[0].width, 2); + } + + #[test] + fn prediction_at_end_of_line_stops() { + let mut overlay = make_overlay(); + // Position cursor near end (col 79 in 80-col terminal) + let seq = overlay.predict_char('a', (79, 0), 80); + assert!(seq.is_some()); + // Cursor should advance past the end (col 80, which is >= cols) + assert_eq!(overlay.predicted_cursor(), Some((80, 0))); + + // Next char should fail because cursor is at cols (80 >= 80) + let seq = overlay.predict_char('b', (79, 0), 80); + assert!(seq.is_none()); + } + + #[test] + fn discard_all_increments_epoch() { + let mut overlay = make_overlay(); + let e0 = overlay.epoch(); + overlay.discard_all(); + assert_eq!(overlay.epoch(), e0 + 1); + } + + #[test] + fn should_predict_requires_stable_frames() { + let mut overlay = InputOverlay::new(); + // No frames yet — stability count is 0 + assert!(!overlay.should_predict('a', true)); + + // Only 2 stable frames (need 3) + overlay.on_server_frame(0, (0, 0)); + overlay.on_server_frame(0, (0, 0)); + assert!(!overlay.should_predict('a', true)); + + // Third stable frame + overlay.on_server_frame(0, (0, 0)); + assert!(overlay.should_predict('a', true)); + } +} diff --git a/src/terminal/mod.rs b/src/terminal/mod.rs index e1e8d4625..ad0d1e268 100644 --- a/src/terminal/mod.rs +++ b/src/terminal/mod.rs @@ -1,5 +1,6 @@ pub mod backend; pub mod input; +pub mod input_overlay; pub mod pty_manager; pub mod session_backend; pub mod shell_config; diff --git a/src/terminal/terminal.rs b/src/terminal/terminal.rs index 504648651..d7570e9a2 100644 --- a/src/terminal/terminal.rs +++ b/src/terminal/terminal.rs @@ -6,9 +6,10 @@ use alacritty_terminal::selection::{Selection, SelectionType}; use alacritty_terminal::index::{Point, Line, Column, Side}; use alacritty_terminal::term::cell::Flags; use alacritty_terminal::grid::{Scroll, Dimensions}; +use crate::terminal::input_overlay::{InputOverlay, PredictedCell}; use parking_lot::Mutex; use regex::Regex; -use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; use std::sync::{Arc, OnceLock}; use std::time::Instant; @@ -242,15 +243,56 @@ pub struct Terminal { had_user_input: AtomicBool, /// Timestamp of when the user last viewed this terminal (on blur) last_viewed_time: Arc>, + /// Optimistic input prediction overlay for remote terminals + input_overlay: Mutex, + /// Whether this terminal is connected to a remote server + is_remote: bool, + /// Latest input sequence number (for remote transport to read) + latest_input_seq: Arc, } impl Terminal { - /// Create a new terminal + /// Create a new terminal. + /// Set `is_remote` to true for remote terminals (enables input prediction). pub fn new( terminal_id: String, size: TerminalSize, transport: Arc, initial_cwd: String, + ) -> Self { + Self::new_inner(terminal_id, size, transport, initial_cwd, false) + } + + /// Create a new remote terminal with input prediction enabled. + pub fn new_remote( + terminal_id: String, + size: TerminalSize, + transport: Arc, + initial_cwd: String, + ) -> Self { + Self::new_inner(terminal_id, size, transport, initial_cwd, true) + } + + /// Create a new remote terminal with a shared input sequence atomic. + /// The same `AtomicU64` should be passed to `RemoteTransport` so they stay in sync. + pub fn new_remote_with_seq( + terminal_id: String, + size: TerminalSize, + transport: Arc, + initial_cwd: String, + input_seq: Arc, + ) -> Self { + let mut terminal = Self::new_inner(terminal_id, size, transport, initial_cwd, true); + terminal.latest_input_seq = input_seq; + terminal + } + + fn new_inner( + terminal_id: String, + size: TerminalSize, + transport: Arc, + initial_cwd: String, + is_remote: bool, ) -> Self { let config = TermConfig::default(); let term_size = TermSize::new(size.cols as usize, size.rows as usize); @@ -291,6 +333,9 @@ impl Terminal { waiting_for_input: AtomicBool::new(false), had_user_input: AtomicBool::new(false), last_viewed_time: Arc::new(Mutex::new(Instant::now())), + input_overlay: Mutex::new(InputOverlay::new()), + is_remote, + latest_input_seq: Arc::new(AtomicU64::new(0)), } } @@ -302,6 +347,13 @@ impl Terminal { processor.advance(&mut *term, data); self.dirty.store(true, Ordering::Relaxed); *self.last_output_time.lock() = Instant::now(); + + // Update input overlay: gc expired predictions + if self.is_remote { + drop(processor); + drop(term); + self.input_overlay.lock().gc_expired(); + } } /// Check if terminal has pending changes (and clear the flag) @@ -694,6 +746,53 @@ impl Terminal { *self.last_output_time.lock() > *self.last_viewed_time.lock() } + /// Whether this terminal is connected to a remote server. + pub fn is_remote(&self) -> bool { + self.is_remote + } + + /// Get the latest input sequence atomic (for transport to read). + pub fn latest_input_seq(&self) -> &Arc { + &self.latest_input_seq + } + + /// Predict a character for optimistic display (remote terminals only). + /// Returns the assigned input sequence number if prediction was made. + pub fn predict_char(&self, c: char) -> Option { + if !self.is_remote { + return None; + } + let term = self.term.lock(); + let cursor = term.grid().cursor.point; + let cols = term.grid().columns(); + drop(term); + + let mut overlay = self.input_overlay.lock(); + let seq = overlay.predict_char(c, (cursor.column.0, cursor.line.0), cols)?; + self.latest_input_seq.store(seq, Ordering::Relaxed); + Some(seq) + } + + /// Acknowledge predictions up to the given sequence number. + pub fn ack_predictions(&self, seq: u64) { + let term = self.term.lock(); + let cursor = term.grid().cursor.point; + drop(term); + + let mut overlay = self.input_overlay.lock(); + overlay.on_server_frame(seq, (cursor.column.0, cursor.line.0)); + } + + /// Get overlay cells for rendering (cloned). + pub fn overlay_cells(&self) -> Vec { + self.input_overlay.lock().cells().iter().cloned().collect() + } + + /// Get the predicted cursor position (if predictions are active). + pub fn predicted_cursor(&self) -> Option<(usize, i32)> { + self.input_overlay.lock().predicted_cursor() + } + /// Search the terminal grid for occurrences of a query string /// Returns a list of (line, col, length) for each match /// Supports case-sensitive and regex search, and searches through scrollback buffer diff --git a/src/views/layout/terminal_pane/navigation.rs b/src/views/layout/terminal_pane/navigation.rs index f82d14b36..8e62a3b29 100644 --- a/src/views/layout/terminal_pane/navigation.rs +++ b/src/views/layout/terminal_pane/navigation.rs @@ -94,6 +94,13 @@ impl TerminalPane { terminal.claim_resize_local(); let app_cursor_mode = terminal.is_app_cursor_mode(); if let Some(input) = key_to_bytes(event, app_cursor_mode) { + // Predict printable ASCII chars for remote terminals + if terminal.is_remote() && input.len() == 1 { + let byte = input[0]; + if byte >= 0x20 && byte < 0x7f { + terminal.predict_char(byte as char); + } + } terminal.send_bytes(&input); } } diff --git a/src/views/root/mod.rs b/src/views/root/mod.rs index c042f118b..f01187792 100644 --- a/src/views/root/mod.rs +++ b/src/views/root/mod.rs @@ -180,8 +180,11 @@ impl RootView { // Observe remote manager and sync remote projects into workspace let workspace = self.workspace.clone(); cx.observe(&manager, move |this, rm, cx| { + log::info!("[root] remote manager notified, syncing remote projects"); Self::sync_remote_projects_into_workspace(&workspace, &rm, cx); + log::info!("[root] sync done, syncing project columns"); this.sync_project_columns(cx); + log::info!("[root] project columns synced"); cx.notify(); }).detach(); diff --git a/src/workspace/actions/project.rs b/src/workspace/actions/project.rs index 5d36d0933..88c8c450d 100644 --- a/src/workspace/actions/project.rs +++ b/src/workspace/actions/project.rs @@ -83,6 +83,14 @@ impl Workspace { } self.notify_data(cx); } + + // Focus the newly created terminal (terminal_id: None) + let new_path = self.project(project_id) + .and_then(|p| p.layout.as_ref()) + .and_then(|l| l.find_uninitialized_terminal_path()); + if let Some(path) = new_path { + self.set_focused_terminal(project_id.to_string(), path, cx); + } } /// Add a new terminal running a specific command to a project diff --git a/src/workspace/settings.rs b/src/workspace/settings.rs index 06920e49e..aac07466f 100644 --- a/src/workspace/settings.rs +++ b/src/workspace/settings.rs @@ -260,6 +260,10 @@ pub struct AppSettings { #[serde(default)] pub worktree: WorktreeConfig, + /// Enable optimistic input prediction for remote terminals (default: true) + #[serde(default = "default_prediction_enabled")] + pub prediction_enabled: bool, + /// Saved remote connections for the client feature #[serde(default)] pub remote_connections: Vec, @@ -294,6 +298,7 @@ impl Default for AppSettings { auto_update_enabled: default_auto_update_enabled(), idle_timeout_secs: default_idle_timeout_secs(), worktree: WorktreeConfig::default(), + prediction_enabled: default_prediction_enabled(), remote_connections: Vec::new(), } } @@ -352,6 +357,10 @@ fn default_idle_timeout_secs() -> u32 { 0 } +fn default_prediction_enabled() -> bool { + true +} + fn default_remote_listen_address() -> String { "127.0.0.1".to_string() }