mirror of
https://github.com/zellij-org/zellij.git
synced 2026-10-11 22:50:07 +00:00
Nested session handling (#5417)
* fix: theme query and updated pixel dimensions * add no-ui-fullscreen key/action/capability * protocol core * integration test harness * fix integration tests * fix flaky tests * nested ui (dimming) and directional focus initial implementation * integration tests for nested ui dimming and directional focus * initial full screen nesteed implementation * fix: keep hidden floating panes hidden across nested no-ui fullscreen * fix integration tests * no ui fullscreen integration tests * hardening behavior/tests * fix(nested): re-derive descend/dim on host tab switch and new-tab so clicking dimmed chrome ascends * initial modal implementation * initial ui improvements * fix integration tests * improve nested selection ui * nested session modal refactoring and tests * improve integration test speed somewhat * some ui adjustments * simplify full/auto nested session handling * hints in compact-bar * guest stdin leak fix * allow floating panes to be fullscreen * make nested session handling configurable * protocol reannounce * refactor: naming confusion * improve ux somewhat * rustfmt * merge fixes * improve integration test rate limiting (fix ci) * add pr
This commit is contained in:
@@ -1,7 +1,8 @@
|
||||
//! Main input logic.
|
||||
use crate::{
|
||||
os_input_output::ClientOsApi, stdin_ansi_parser::AnsiStdinInstruction, ClientId,
|
||||
ClientInstruction, CommandIsExecuting, InputInstruction,
|
||||
nested_reannounce::NestedReannounce, os_input_output::ClientOsApi,
|
||||
stdin_ansi_parser::AnsiStdinInstruction, ClientId, ClientInstruction, CommandIsExecuting,
|
||||
InputInstruction,
|
||||
};
|
||||
use zellij_utils::{
|
||||
channels::{Receiver, SenderWithContext, OPENCALLS},
|
||||
@@ -15,6 +16,7 @@ use zellij_utils::{
|
||||
options::Options,
|
||||
},
|
||||
ipc::{ClientToServerMsg, ExitReason},
|
||||
nested_session::{self, NestedSessionMessage},
|
||||
position::Position,
|
||||
vendored::termwiz::input::{
|
||||
InputEvent, Modifiers, MouseButtons, MouseEvent as TermwizMouseEvent,
|
||||
@@ -35,6 +37,7 @@ struct InputHandler {
|
||||
receive_input_instructions: Receiver<(InputInstruction, ErrorContext)>,
|
||||
mouse_old_event: MouseEvent,
|
||||
mouse_mode_active: bool,
|
||||
nested_reannounce: Option<NestedReannounce>,
|
||||
}
|
||||
|
||||
fn termwiz_mouse_convert(original_event: &mut MouseEvent, event: &TermwizMouseEvent) {
|
||||
@@ -136,6 +139,7 @@ impl InputHandler {
|
||||
mode: InputMode, // TODO: we can probably get rid of this now that we're tracking it on the
|
||||
// server instead
|
||||
receive_input_instructions: Receiver<(InputInstruction, ErrorContext)>,
|
||||
nested_reannounce: Option<NestedReannounce>,
|
||||
) -> Self {
|
||||
InputHandler {
|
||||
mode,
|
||||
@@ -148,6 +152,7 @@ impl InputHandler {
|
||||
receive_input_instructions,
|
||||
mouse_old_event: MouseEvent::new(),
|
||||
mouse_mode_active: false,
|
||||
nested_reannounce,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -269,6 +274,12 @@ impl InputHandler {
|
||||
reply_bytes,
|
||||
});
|
||||
},
|
||||
Ok((
|
||||
InputInstruction::NestedSessionFrameFromHost(payload_bytes),
|
||||
_error_context,
|
||||
)) => {
|
||||
self.handle_nested_session_frame_from_host(payload_bytes);
|
||||
},
|
||||
Ok((InputInstruction::Exit, _error_context)) => {
|
||||
self.should_exit = true;
|
||||
},
|
||||
@@ -329,6 +340,29 @@ impl InputHandler {
|
||||
},
|
||||
}
|
||||
}
|
||||
fn handle_nested_session_frame_from_host(&mut self, payload_bytes: Vec<u8>) {
|
||||
if let Some(nested_reannounce) = &self.nested_reannounce {
|
||||
nested_reannounce.note_host_contact();
|
||||
}
|
||||
match nested_session::decode_payload(&payload_bytes) {
|
||||
Some(NestedSessionMessage::Ping) => {
|
||||
self.send_client_instructions
|
||||
.send(ClientInstruction::EmitNestedSessionFrame(
|
||||
nested_session::encode_payload(&NestedSessionMessage::Pong),
|
||||
))
|
||||
.unwrap();
|
||||
},
|
||||
Some(_) => {
|
||||
self.os_input
|
||||
.send_to_server(ClientToServerMsg::NestedSessionFrameFromHost {
|
||||
payload_bytes,
|
||||
});
|
||||
},
|
||||
None => {
|
||||
log::debug!("dropping undecodable nested session frame from host");
|
||||
},
|
||||
}
|
||||
}
|
||||
fn handle_mouse_event(&mut self, mouse_event: &MouseEvent) {
|
||||
self.dispatch_action(
|
||||
Action::MouseEvent {
|
||||
@@ -450,6 +484,7 @@ pub(crate) fn input_loop(
|
||||
send_client_instructions: SenderWithContext<ClientInstruction>,
|
||||
default_mode: InputMode,
|
||||
receive_input_instructions: Receiver<(InputInstruction, ErrorContext)>,
|
||||
nested_reannounce: Option<NestedReannounce>,
|
||||
) {
|
||||
let _handler = InputHandler::new(
|
||||
os_input,
|
||||
@@ -459,6 +494,7 @@ pub(crate) fn input_loop(
|
||||
send_client_instructions,
|
||||
default_mode,
|
||||
receive_input_instructions,
|
||||
nested_reannounce,
|
||||
)
|
||||
.handle_input();
|
||||
}
|
||||
|
||||
+146
-3
@@ -11,6 +11,7 @@ pub mod cli_client;
|
||||
mod command_is_executing;
|
||||
mod input_handler;
|
||||
mod keyboard_parser;
|
||||
mod nested_reannounce;
|
||||
pub mod old_config_converter;
|
||||
#[cfg(feature = "web_server_capability")]
|
||||
pub mod remote_attach;
|
||||
@@ -157,6 +158,7 @@ use zellij_utils::{
|
||||
errors::{ClientContext, ContextType, ErrorInstruction},
|
||||
input::{cli_assets::CliAssets, config::Config, options::Options},
|
||||
ipc::{ClientToServerMsg, ExitReason, ResizeCause, ServerToClientMsg},
|
||||
nested_session,
|
||||
pane_size::Size,
|
||||
vendored::termwiz::input::InputEvent,
|
||||
};
|
||||
@@ -186,6 +188,7 @@ pub(crate) enum ClientInstruction {
|
||||
token: u32,
|
||||
query_bytes: Vec<u8>,
|
||||
},
|
||||
EmitNestedSessionFrame(Vec<u8>),
|
||||
}
|
||||
|
||||
impl From<ServerToClientMsg> for ClientInstruction {
|
||||
@@ -211,6 +214,9 @@ impl From<ServerToClientMsg> for ClientInstruction {
|
||||
ServerToClientMsg::ForwardQueryToHost { token, query_bytes } => {
|
||||
ClientInstruction::ForwardQueryToHost { token, query_bytes }
|
||||
},
|
||||
ServerToClientMsg::EmitNestedSessionFrame { payload_bytes } => {
|
||||
ClientInstruction::EmitNestedSessionFrame(payload_bytes)
|
||||
},
|
||||
// Subscribe-only messages — not handled by regular interactive clients
|
||||
ServerToClientMsg::PaneRenderUpdate { .. } => ClientInstruction::UnblockInputThread,
|
||||
ServerToClientMsg::SubscribedPaneClosed { .. } => ClientInstruction::UnblockInputThread,
|
||||
@@ -238,6 +244,7 @@ impl From<&ClientInstruction> for ClientContext {
|
||||
ClientInstruction::RenamedSession(..) => ClientContext::RenamedSession,
|
||||
ClientInstruction::ConfigFileUpdated => ClientContext::ConfigFileUpdated,
|
||||
ClientInstruction::ForwardQueryToHost { .. } => ClientContext::ForwardQueryToHost,
|
||||
ClientInstruction::EmitNestedSessionFrame(..) => ClientContext::EmitNestedSessionFrame,
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -523,6 +530,7 @@ pub(crate) enum InputInstruction {
|
||||
token: u32,
|
||||
reply_bytes: Vec<u8>,
|
||||
},
|
||||
NestedSessionFrameFromHost(Vec<u8>),
|
||||
Exit,
|
||||
}
|
||||
|
||||
@@ -530,6 +538,7 @@ pub(crate) enum InputInstruction {
|
||||
pub async fn run_remote_client_terminal_loop(
|
||||
os_input: Box<dyn ClientOsApi>,
|
||||
mut connections: remote_attach::WebSocketConnections,
|
||||
nested_session_name: Option<String>,
|
||||
) -> Result<Option<ConnectToSession>, RemoteClientError> {
|
||||
use crate::os_input_output::{AsyncSignals, AsyncStdin};
|
||||
|
||||
@@ -563,15 +572,52 @@ pub async fn run_remote_client_terminal_loop(
|
||||
log::error!("Failed to send resize message: {}", e);
|
||||
}
|
||||
|
||||
let mut nested_frame_extractor = nested_session::NestedFrameExtractor::new();
|
||||
let mut last_heard_from_host = std::time::Instant::now();
|
||||
let mut reannounce_check = tokio::time::interval(std::time::Duration::from_millis(
|
||||
nested_session::reannounce_check_interval_ms(),
|
||||
));
|
||||
reannounce_check.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
|
||||
|
||||
loop {
|
||||
tokio::select! {
|
||||
// Handle stdin input
|
||||
result = async_stdin.read() => {
|
||||
match result {
|
||||
Ok(buf) if !buf.is_empty() => {
|
||||
if let Err(e) = connections.terminal_ws.send(Message::Binary(buf)).await {
|
||||
log::error!("Failed to send stdin to terminal WebSocket: {}", e);
|
||||
break;
|
||||
let (cleaned, nested_frames) = nested_frame_extractor.extract(&buf);
|
||||
for payload_bytes in nested_frames {
|
||||
last_heard_from_host = std::time::Instant::now();
|
||||
match nested_session::decode_payload(&payload_bytes) {
|
||||
Some(nested_session::NestedSessionMessage::Ping) => {
|
||||
let mut stdout = os_input.get_stdout_writer();
|
||||
let _ = stdout.write_all(&nested_session::encode_frame(
|
||||
&nested_session::NestedSessionMessage::Pong,
|
||||
));
|
||||
let _ = stdout.flush();
|
||||
},
|
||||
Some(_) => {
|
||||
let control_msg = Message::Text(
|
||||
serde_json::to_string(&WebClientToWebServerControlMessage {
|
||||
web_client_id: connections.web_client_id.clone(),
|
||||
payload: WebClientToWebServerControlMessagePayload::NestedSessionFrameFromHost {
|
||||
payload_bytes,
|
||||
},
|
||||
})
|
||||
.unwrap(),
|
||||
);
|
||||
if let Err(e) = connections.control_ws.send(control_msg).await {
|
||||
log::error!("Failed to forward nested session frame over control WebSocket: {}", e);
|
||||
}
|
||||
},
|
||||
None => {},
|
||||
}
|
||||
}
|
||||
if !cleaned.is_empty() {
|
||||
if let Err(e) = connections.terminal_ws.send(Message::Binary(cleaned)).await {
|
||||
log::error!("Failed to send stdin to terminal WebSocket: {}", e);
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
Ok(_) => {
|
||||
@@ -585,6 +631,26 @@ pub async fn run_remote_client_terminal_loop(
|
||||
}
|
||||
}
|
||||
|
||||
_ = reannounce_check.tick() => {
|
||||
if let Some(session_name) = &nested_session_name {
|
||||
if last_heard_from_host.elapsed()
|
||||
>= std::time::Duration::from_millis(nested_session::reannounce_silence_ms())
|
||||
{
|
||||
let announce = nested_session::NestedSessionMessage::Announce {
|
||||
session_name: session_name.clone(),
|
||||
capabilities: vec![nested_session::NestedSessionCapability::NestedControl],
|
||||
};
|
||||
let mut stdout = os_input.get_stdout_writer();
|
||||
if stdout
|
||||
.write_all(&nested_session::encode_frame(&announce))
|
||||
.is_ok()
|
||||
{
|
||||
let _ = stdout.flush();
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Handle signals
|
||||
Some(signal) = async_signals.recv() => {
|
||||
match signal {
|
||||
@@ -721,6 +787,14 @@ pub fn start_remote_client(
|
||||
) -> Result<Option<ConnectToSession>, RemoteClientError> {
|
||||
info!("Starting Zellij client!");
|
||||
|
||||
let is_nested_inside_zellij_pane = os_input.env_variable("ZELLIJ").is_some();
|
||||
let remote_session_name = remote_session_url
|
||||
.trim_end_matches('/')
|
||||
.rsplit('/')
|
||||
.next()
|
||||
.unwrap_or_default()
|
||||
.to_owned();
|
||||
|
||||
let runtime = crate::async_runtime(async_worker_tasks);
|
||||
|
||||
let connections = remote_attach::attach_to_remote_session(
|
||||
@@ -756,6 +830,16 @@ pub fn start_remote_client(
|
||||
|
||||
os_input.set_raw_mode();
|
||||
stdout.write_all(ENABLE_BRACKETED_PASTE.as_bytes()).unwrap();
|
||||
if is_nested_inside_zellij_pane {
|
||||
let announce = nested_session::NestedSessionMessage::Announce {
|
||||
session_name: remote_session_name.clone(),
|
||||
capabilities: vec![nested_session::NestedSessionCapability::NestedControl],
|
||||
};
|
||||
stdout
|
||||
.write_all(&nested_session::encode_frame(&announce))
|
||||
.unwrap();
|
||||
let _ = stdout.flush();
|
||||
}
|
||||
|
||||
std::panic::set_hook({
|
||||
use zellij_utils::errors::handle_panic;
|
||||
@@ -785,11 +869,25 @@ pub fn start_remote_client(
|
||||
std::process::exit(exit_status);
|
||||
};
|
||||
|
||||
let nested_session_name = if is_nested_inside_zellij_pane {
|
||||
Some(remote_session_name.clone())
|
||||
} else {
|
||||
None
|
||||
};
|
||||
runtime.block_on(run_remote_client_terminal_loop(
|
||||
os_input.clone(),
|
||||
connections,
|
||||
nested_session_name,
|
||||
))?;
|
||||
|
||||
if is_nested_inside_zellij_pane {
|
||||
let mut stdout = os_input.get_stdout_writer();
|
||||
let _ = stdout.write_all(&nested_session::encode_frame(
|
||||
&nested_session::NestedSessionMessage::Bye,
|
||||
));
|
||||
let _ = stdout.flush();
|
||||
}
|
||||
|
||||
let exit_msg = String::from("Bye from Zellij!");
|
||||
|
||||
if reconnect_to_session.is_none() {
|
||||
@@ -822,6 +920,9 @@ pub fn start_client(
|
||||
}
|
||||
info!("Starting Zellij client!");
|
||||
|
||||
let is_nested_inside_zellij_pane = os_input.env_variable("ZELLIJ").is_some();
|
||||
let own_session_name = info.get_session_name().to_owned();
|
||||
|
||||
let explicitly_disable_kitty_keyboard_protocol = config_options
|
||||
.support_kitty_keyboard_protocol
|
||||
.map(|e| !e)
|
||||
@@ -1043,6 +1144,22 @@ pub fn start_client(
|
||||
os_input.set_raw_mode();
|
||||
let mut stdout = os_input.get_stdout_writer();
|
||||
stdout.write_all(ENABLE_BRACKETED_PASTE.as_bytes()).unwrap();
|
||||
let nested_reannounce = if is_nested_inside_zellij_pane {
|
||||
let announce = nested_session::NestedSessionMessage::Announce {
|
||||
session_name: own_session_name.clone(),
|
||||
capabilities: vec![nested_session::NestedSessionCapability::NestedControl],
|
||||
};
|
||||
stdout
|
||||
.write_all(&nested_session::encode_frame(&announce))
|
||||
.unwrap();
|
||||
let _ = stdout.flush();
|
||||
Some(crate::nested_reannounce::NestedReannounce::spawn(
|
||||
os_input.clone(),
|
||||
own_session_name.clone(),
|
||||
))
|
||||
} else {
|
||||
None
|
||||
};
|
||||
|
||||
let (send_client_instructions, receive_client_instructions): ChannelWithContext<
|
||||
ClientInstruction,
|
||||
@@ -1116,6 +1233,7 @@ pub fn start_client(
|
||||
let command_is_executing = command_is_executing.clone();
|
||||
let os_input = os_input.clone();
|
||||
let default_mode = config_options.default_mode.unwrap_or_default();
|
||||
let nested_reannounce = nested_reannounce.clone();
|
||||
move || {
|
||||
input_loop(
|
||||
os_input,
|
||||
@@ -1125,6 +1243,7 @@ pub fn start_client(
|
||||
send_client_instructions,
|
||||
default_mode,
|
||||
receive_input_instructions,
|
||||
nested_reannounce,
|
||||
)
|
||||
}
|
||||
});
|
||||
@@ -1142,6 +1261,10 @@ pub fn start_client(
|
||||
new_size: os_api.get_terminal_size(),
|
||||
cause: ResizeCause::Viewport,
|
||||
});
|
||||
#[cfg(not(windows))]
|
||||
let _ = os_api
|
||||
.get_stdout_writer()
|
||||
.write(crate::stdin_handler::PIXEL_SIZE_QUERY.as_bytes());
|
||||
}
|
||||
}),
|
||||
Box::new({
|
||||
@@ -1340,12 +1463,32 @@ pub fn start_client(
|
||||
let _ = out.write_all(&blob);
|
||||
let _ = out.flush();
|
||||
},
|
||||
ClientInstruction::EmitNestedSessionFrame(payload_bytes) => {
|
||||
if is_nested_inside_zellij_pane {
|
||||
let frame = nested_session::encode_frame_from_payload(&payload_bytes);
|
||||
let mut out = os_input.get_stdout_writer();
|
||||
let _ = out.write_all(&frame);
|
||||
let _ = out.flush();
|
||||
}
|
||||
},
|
||||
_ => {},
|
||||
}
|
||||
}
|
||||
|
||||
if let Some(nested_reannounce) = &nested_reannounce {
|
||||
nested_reannounce.stop();
|
||||
}
|
||||
|
||||
router_thread.join().unwrap();
|
||||
|
||||
if is_nested_inside_zellij_pane {
|
||||
let mut stdout = os_input.get_stdout_writer();
|
||||
let _ = stdout.write_all(&nested_session::encode_frame(
|
||||
&nested_session::NestedSessionMessage::Bye,
|
||||
));
|
||||
let _ = stdout.flush();
|
||||
}
|
||||
|
||||
if reconnect_to_session.is_none() {
|
||||
let goodbye_message = terminal_teardown_message(
|
||||
&exit_msg,
|
||||
|
||||
@@ -0,0 +1,63 @@
|
||||
use crate::os_input_output::ClientOsApi;
|
||||
use std::io::Write;
|
||||
use std::sync::atomic::{AtomicBool, Ordering};
|
||||
use std::sync::{Arc, Mutex};
|
||||
use std::time::{Duration, Instant};
|
||||
use zellij_utils::nested_session::{
|
||||
self, reannounce_check_interval_ms, reannounce_silence_ms, NestedSessionCapability,
|
||||
NestedSessionMessage,
|
||||
};
|
||||
|
||||
#[derive(Clone)]
|
||||
pub struct NestedReannounce {
|
||||
last_heard_from_host: Arc<Mutex<Instant>>,
|
||||
stop: Arc<AtomicBool>,
|
||||
}
|
||||
|
||||
impl NestedReannounce {
|
||||
pub fn spawn(os_input: Box<dyn ClientOsApi>, session_name: String) -> Self {
|
||||
let handle = NestedReannounce {
|
||||
last_heard_from_host: Arc::new(Mutex::new(Instant::now())),
|
||||
stop: Arc::new(AtomicBool::new(false)),
|
||||
};
|
||||
let last_heard_from_host = handle.last_heard_from_host.clone();
|
||||
let stop = handle.stop.clone();
|
||||
let _ = std::thread::Builder::new()
|
||||
.name("nested_reannounce".to_string())
|
||||
.spawn(move || {
|
||||
let check_interval = Duration::from_millis(reannounce_check_interval_ms());
|
||||
loop {
|
||||
std::thread::sleep(check_interval);
|
||||
if stop.load(Ordering::Relaxed) {
|
||||
break;
|
||||
}
|
||||
let silent_for = {
|
||||
let last = last_heard_from_host.lock().unwrap();
|
||||
last.elapsed()
|
||||
};
|
||||
if silent_for >= Duration::from_millis(reannounce_silence_ms()) {
|
||||
let announce = NestedSessionMessage::Announce {
|
||||
session_name: session_name.clone(),
|
||||
capabilities: vec![NestedSessionCapability::NestedControl],
|
||||
};
|
||||
let mut stdout = os_input.get_stdout_writer();
|
||||
if stdout
|
||||
.write_all(&nested_session::encode_frame(&announce))
|
||||
.is_ok()
|
||||
{
|
||||
let _ = stdout.flush();
|
||||
}
|
||||
}
|
||||
}
|
||||
});
|
||||
handle
|
||||
}
|
||||
|
||||
pub fn note_host_contact(&self) {
|
||||
*self.last_heard_from_host.lock().unwrap() = Instant::now();
|
||||
}
|
||||
|
||||
pub fn stop(&self) {
|
||||
self.stop.store(true, Ordering::Relaxed);
|
||||
}
|
||||
}
|
||||
@@ -12,6 +12,7 @@ use serde::{Deserialize, Serialize};
|
||||
use zellij_utils::{
|
||||
data::HostTerminalThemeMode,
|
||||
ipc::PixelDimensions,
|
||||
nested_session,
|
||||
pane_size::SizeInPixels,
|
||||
vendored::termwiz::input::{InputEvent, InputParser},
|
||||
};
|
||||
@@ -187,6 +188,7 @@ pub struct ParseOutput {
|
||||
/// Residue bytes that were not classified as host replies. These are
|
||||
/// the bytes the caller should feed to the keyboard parser.
|
||||
pub residue: Vec<u8>,
|
||||
pub nested_frames: Vec<Vec<u8>>,
|
||||
/// `true` when the parser still holds buffered partial OSC/CSI
|
||||
/// bytes that need an idle-flush to release. The caller uses this
|
||||
/// to schedule a finalize tick even when the current chunk
|
||||
@@ -264,6 +266,7 @@ pub struct StdinAnsiParser {
|
||||
/// Same for CSI device-control reports.
|
||||
partial_csi: Vec<u8>,
|
||||
partial_paste: Vec<u8>,
|
||||
nested_frame_extractor: nested_session::NestedFrameExtractor,
|
||||
in_bracketed_paste: bool,
|
||||
}
|
||||
|
||||
@@ -285,6 +288,7 @@ impl StdinAnsiParser {
|
||||
partial_osc: Vec::new(),
|
||||
partial_csi: Vec::new(),
|
||||
partial_paste: Vec::new(),
|
||||
nested_frame_extractor: nested_session::NestedFrameExtractor::new(),
|
||||
in_bracketed_paste: false,
|
||||
}
|
||||
}
|
||||
@@ -349,6 +353,9 @@ impl StdinAnsiParser {
|
||||
/// are the bytes the caller should feed to the keyboard parser.
|
||||
pub fn feed(&mut self, bytes: &[u8]) -> ParseOutput {
|
||||
let mut out = ParseOutput::default();
|
||||
let (bytes, nested_frames) = self.nested_frame_extractor.extract(bytes);
|
||||
let bytes = &bytes[..];
|
||||
out.nested_frames = nested_frames;
|
||||
// Collect events first (borrow-splits the InputParser across the
|
||||
// callback and the post-processing mutations).
|
||||
let mut events = Vec::new();
|
||||
@@ -431,13 +438,21 @@ impl StdinAnsiParser {
|
||||
out.residue = residue;
|
||||
out.has_partial_state = !self.partial_osc.is_empty()
|
||||
|| !self.partial_csi.is_empty()
|
||||
|| !self.partial_paste.is_empty();
|
||||
|| !self.partial_paste.is_empty()
|
||||
|| !self.nested_frame_extractor.partial_bytes().is_empty();
|
||||
out
|
||||
}
|
||||
|
||||
pub fn pending_partial(&self) -> PendingPartial {
|
||||
if !self.partial_paste.is_empty() {
|
||||
PendingPartial::ReplyInProgress
|
||||
} else if self.partial_csi.is_empty()
|
||||
&& self.partial_osc.is_empty()
|
||||
&& self.nested_frame_extractor.partial_bytes() == [0x1b]
|
||||
{
|
||||
PendingPartial::LoneEsc
|
||||
} else if !self.nested_frame_extractor.partial_bytes().is_empty() {
|
||||
PendingPartial::ReplyInProgress
|
||||
} else if self.partial_csi.is_empty() && self.partial_osc.is_empty() {
|
||||
PendingPartial::None
|
||||
} else if self.partial_csi.is_empty() && self.partial_osc == [0x1b] {
|
||||
@@ -450,6 +465,13 @@ impl StdinAnsiParser {
|
||||
pub fn finalize_lone_esc(&mut self) -> Vec<u8> {
|
||||
if self.partial_csi.is_empty()
|
||||
&& self.partial_paste.is_empty()
|
||||
&& self.partial_osc.is_empty()
|
||||
&& self.nested_frame_extractor.partial_bytes() == [0x1b]
|
||||
{
|
||||
self.nested_frame_extractor.take_partial()
|
||||
} else if self.partial_csi.is_empty()
|
||||
&& self.partial_paste.is_empty()
|
||||
&& self.nested_frame_extractor.partial_bytes().is_empty()
|
||||
&& self.partial_osc == [0x1b]
|
||||
{
|
||||
std::mem::take(&mut self.partial_osc)
|
||||
@@ -460,12 +482,17 @@ impl StdinAnsiParser {
|
||||
|
||||
pub fn finalize_force(&mut self) -> Vec<u8> {
|
||||
self.in_bracketed_paste = false;
|
||||
let mut partial_nested_frame = self.nested_frame_extractor.take_partial();
|
||||
let mut out = Vec::with_capacity(
|
||||
self.partial_osc.len() + self.partial_csi.len() + self.partial_paste.len(),
|
||||
self.partial_osc.len()
|
||||
+ self.partial_csi.len()
|
||||
+ self.partial_paste.len()
|
||||
+ partial_nested_frame.len(),
|
||||
);
|
||||
out.append(&mut self.partial_osc);
|
||||
out.append(&mut self.partial_csi);
|
||||
out.append(&mut self.partial_paste);
|
||||
out.append(&mut partial_nested_frame);
|
||||
out
|
||||
}
|
||||
|
||||
|
||||
@@ -1225,3 +1225,107 @@ fn unsolicited_theme_notification_classifies_without_outstanding_query() {
|
||||
other => panic!("expected HostTerminalThemeChanged, got {:?}", other),
|
||||
}
|
||||
}
|
||||
|
||||
fn announce_frame_and_payload() -> (Vec<u8>, Vec<u8>) {
|
||||
use zellij_utils::nested_session::{
|
||||
encode_frame, encode_payload, NestedSessionCapability, NestedSessionMessage,
|
||||
};
|
||||
let message = NestedSessionMessage::Announce {
|
||||
session_name: "guest-session".to_owned(),
|
||||
capabilities: vec![NestedSessionCapability::NestedControl],
|
||||
};
|
||||
(encode_frame(&message), encode_payload(&message))
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn nested_frame_in_one_chunk_is_extracted_with_no_residue() {
|
||||
let (frame, payload) = announce_frame_and_payload();
|
||||
let mut p = StdinAnsiParser::new();
|
||||
let out = p.feed(&frame);
|
||||
assert_eq!(out.nested_frames, vec![payload]);
|
||||
assert!(out.residue.is_empty(), "residue: {:?}", out.residue);
|
||||
assert_eq!(p.pending_partial(), PendingPartial::None);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn nested_frame_split_at_every_boundary_is_extracted() {
|
||||
let (frame, payload) = announce_frame_and_payload();
|
||||
for split_at in 1..frame.len() {
|
||||
let mut p = StdinAnsiParser::new();
|
||||
let first = p.feed(&frame[..split_at]);
|
||||
let second = p.feed(&frame[split_at..]);
|
||||
let mut extracted = first.nested_frames.clone();
|
||||
extracted.extend(second.nested_frames.clone());
|
||||
assert_eq!(extracted, vec![payload.clone()], "split at {}", split_at);
|
||||
assert!(
|
||||
first.residue.is_empty() && second.residue.is_empty(),
|
||||
"split at {}: residue {:?} / {:?}",
|
||||
split_at,
|
||||
first.residue,
|
||||
second.residue
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn keystrokes_around_nested_frame_survive_as_residue() {
|
||||
let (frame, payload) = announce_frame_and_payload();
|
||||
let mut chunk = b"abc".to_vec();
|
||||
chunk.extend_from_slice(&frame);
|
||||
chunk.extend_from_slice(b"def");
|
||||
let mut p = StdinAnsiParser::new();
|
||||
let out = p.feed(&chunk);
|
||||
assert_eq!(out.nested_frames, vec![payload]);
|
||||
assert_eq!(out.residue, b"abcdef".to_vec());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn nested_frame_with_garbage_base64_is_stripped_without_payload() {
|
||||
let mut p = StdinAnsiParser::new();
|
||||
let out = p.feed(b"\x1bP26661n!!!not-base64!!!\x1b\\");
|
||||
assert!(out.nested_frames.is_empty());
|
||||
assert!(out.residue.is_empty(), "residue: {:?}", out.residue);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn foreign_dcs_passes_through_untouched() {
|
||||
let foreign_dcs = b"\x1bP1$qm\x1b\\q".to_vec();
|
||||
let mut p = StdinAnsiParser::new();
|
||||
let out = p.feed(&foreign_dcs);
|
||||
assert!(out.nested_frames.is_empty());
|
||||
assert_eq!(out.residue, foreign_dcs);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn lone_esc_claimed_by_frame_prepass_reroutes_on_next_chunk() {
|
||||
let mut p = StdinAnsiParser::new();
|
||||
let first = p.feed(b"\x1b");
|
||||
assert!(first.residue.is_empty());
|
||||
assert!(first.has_partial_state);
|
||||
assert_eq!(p.pending_partial(), PendingPartial::LoneEsc);
|
||||
let second = p.feed(b"abc");
|
||||
assert_eq!(second.residue, b"\x1babc".to_vec());
|
||||
assert_eq!(p.pending_partial(), PendingPartial::None);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn unterminated_frame_header_prefix_is_released_by_force_drain() {
|
||||
let mut p = StdinAnsiParser::new();
|
||||
let out = p.feed(b"\x1bP26");
|
||||
assert!(out.residue.is_empty());
|
||||
assert!(out.has_partial_state);
|
||||
assert_eq!(p.pending_partial(), PendingPartial::ReplyInProgress);
|
||||
assert_eq!(p.finalize_force(), b"\x1bP26".to_vec());
|
||||
assert_eq!(p.pending_partial(), PendingPartial::None);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn two_nested_frames_in_one_chunk_are_both_extracted() {
|
||||
let (frame, payload) = announce_frame_and_payload();
|
||||
let mut chunk = frame.clone();
|
||||
chunk.extend_from_slice(&frame);
|
||||
let mut p = StdinAnsiParser::new();
|
||||
let out = p.feed(&chunk);
|
||||
assert_eq!(out.nested_frames, vec![payload.clone(), payload]);
|
||||
assert!(out.residue.is_empty());
|
||||
}
|
||||
|
||||
@@ -97,7 +97,7 @@ pub(crate) fn stdin_loop(
|
||||
});
|
||||
let mut needs_finalization = false;
|
||||
let mut reply_in_progress_since: Option<Instant> = None;
|
||||
loop {
|
||||
'stdin: loop {
|
||||
match if needs_finalization {
|
||||
stdin_rx.recv_timeout(LONE_ESC_FLUSH_INTERVAL)
|
||||
} else {
|
||||
@@ -132,6 +132,10 @@ pub(crate) fn stdin_loop(
|
||||
let _ = send_input_instructions
|
||||
.send(InputInstruction::DesktopNotificationResponse(payload));
|
||||
}
|
||||
for payload_bytes in parse_output.nested_frames {
|
||||
let _ = send_input_instructions
|
||||
.send(InputInstruction::NestedSessionFrameFromHost(payload_bytes));
|
||||
}
|
||||
let residue = parse_output.residue;
|
||||
if residue.is_empty() {
|
||||
schedule_finalization(
|
||||
@@ -153,13 +157,16 @@ pub(crate) fn stdin_loop(
|
||||
// continuation completes the sequence.
|
||||
match kitty_parser.feed(&residue) {
|
||||
KittyParseOutcome::Complete(key_with_modifier) => {
|
||||
send_input_instructions
|
||||
if send_input_instructions
|
||||
.send(InputInstruction::KeyWithModifierEvent(
|
||||
key_with_modifier,
|
||||
current_buffer.drain(..).collect(),
|
||||
true,
|
||||
))
|
||||
.unwrap();
|
||||
.is_err()
|
||||
{
|
||||
break 'stdin;
|
||||
}
|
||||
schedule_finalization(
|
||||
&stdin_ansi_parser,
|
||||
false,
|
||||
@@ -196,9 +203,12 @@ pub(crate) fn stdin_loop(
|
||||
for (input_event, consumed) in events.into_iter() {
|
||||
let take = consumed.min(current_buffer.len());
|
||||
let raw_bytes: Vec<u8> = current_buffer.drain(..take).collect();
|
||||
send_input_instructions
|
||||
if send_input_instructions
|
||||
.send(InputInstruction::KeyEvent(input_event, raw_bytes))
|
||||
.unwrap();
|
||||
.is_err()
|
||||
{
|
||||
break 'stdin;
|
||||
}
|
||||
}
|
||||
|
||||
schedule_finalization(
|
||||
@@ -317,12 +327,22 @@ fn build_startup_query_string() -> String {
|
||||
// <ESC>]11;?<ESC>\ => get background color
|
||||
// <ESC>]10;?<ESC>\ => get foreground color
|
||||
// <ESC>[?2026$p => get synchronised output mode
|
||||
String::from("\u{1b}[14t\u{1b}[16t\u{1b}]11;?\u{1b}\u{5c}\u{1b}]10;?\u{1b}\u{5c}\u{1b}[?2026$p")
|
||||
format!(
|
||||
"{}\u{1b}]11;?\u{1b}\u{5c}\u{1b}]10;?\u{1b}\u{5c}\u{1b}[?2026$p",
|
||||
PIXEL_SIZE_QUERY
|
||||
)
|
||||
}
|
||||
|
||||
pub(crate) const PIXEL_SIZE_QUERY: &str = "\u{1b}[14t\u{1b}[16t";
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::build_startup_query_string;
|
||||
use super::{build_startup_query_string, PIXEL_SIZE_QUERY};
|
||||
|
||||
#[test]
|
||||
fn pixel_size_query_probes_text_area_and_character_cell() {
|
||||
assert_eq!(PIXEL_SIZE_QUERY, "\u{1b}[14t\u{1b}[16t");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn startup_query_has_no_palette_register_loop() {
|
||||
|
||||
@@ -351,7 +351,9 @@ async fn test_stdin_forwarded_to_terminal_websocket() {
|
||||
let os_input = Box::new(TestClientOsApi::new(stdin_rx, signal_rx));
|
||||
|
||||
let loop_handle =
|
||||
tokio::spawn(async move { run_remote_client_terminal_loop(os_input, connections).await });
|
||||
tokio::spawn(
|
||||
async move { run_remote_client_terminal_loop(os_input, connections, None).await },
|
||||
);
|
||||
|
||||
let test_data = b"hello from stdin\n".to_vec();
|
||||
stdin_tx.send(test_data.clone()).unwrap();
|
||||
@@ -420,7 +422,9 @@ async fn test_terminal_output_written_to_stdout() {
|
||||
let os_input = Box::new(os_input);
|
||||
|
||||
let loop_handle =
|
||||
tokio::spawn(async move { run_remote_client_terminal_loop(os_input, connections).await });
|
||||
tokio::spawn(
|
||||
async move { run_remote_client_terminal_loop(os_input, connections, None).await },
|
||||
);
|
||||
|
||||
let test_output = "Hello from terminal";
|
||||
server
|
||||
@@ -490,7 +494,9 @@ async fn test_resize_signal_sends_control_message() {
|
||||
let os_input = Box::new(TestClientOsApi::new(stdin_rx, signal_rx));
|
||||
|
||||
let loop_handle =
|
||||
tokio::spawn(async move { run_remote_client_terminal_loop(os_input, connections).await });
|
||||
tokio::spawn(
|
||||
async move { run_remote_client_terminal_loop(os_input, connections, None).await },
|
||||
);
|
||||
|
||||
tokio::time::sleep(Duration::from_millis(200)).await;
|
||||
|
||||
@@ -576,7 +582,9 @@ async fn test_quit_signal_exits_loop() {
|
||||
let os_input = Box::new(TestClientOsApi::new(stdin_rx, signal_rx));
|
||||
|
||||
let loop_handle =
|
||||
tokio::spawn(async move { run_remote_client_terminal_loop(os_input, connections).await });
|
||||
tokio::spawn(
|
||||
async move { run_remote_client_terminal_loop(os_input, connections, None).await },
|
||||
);
|
||||
|
||||
signal_tx.send(SignalEvent::Quit).unwrap();
|
||||
|
||||
@@ -629,7 +637,9 @@ async fn test_websocket_close_exits_loop() {
|
||||
let os_input = Box::new(TestClientOsApi::new(stdin_rx, signal_rx));
|
||||
|
||||
let loop_handle =
|
||||
tokio::spawn(async move { run_remote_client_terminal_loop(os_input, connections).await });
|
||||
tokio::spawn(
|
||||
async move { run_remote_client_terminal_loop(os_input, connections, None).await },
|
||||
);
|
||||
|
||||
server
|
||||
.terminal_to_client_tx
|
||||
@@ -687,7 +697,9 @@ async fn test_control_message_handling() {
|
||||
let os_input = Box::new(os_input);
|
||||
|
||||
let loop_handle =
|
||||
tokio::spawn(async move { run_remote_client_terminal_loop(os_input, connections).await });
|
||||
tokio::spawn(
|
||||
async move { run_remote_client_terminal_loop(os_input, connections, None).await },
|
||||
);
|
||||
|
||||
tokio::time::sleep(Duration::from_millis(200)).await;
|
||||
|
||||
@@ -746,3 +758,148 @@ async fn test_control_message_handling() {
|
||||
.expect("Loop didn't exit")
|
||||
.unwrap();
|
||||
}
|
||||
|
||||
async fn connect_mock_ws(port: u16) -> WebSocketConnections {
|
||||
let terminal_url = format!("ws://127.0.0.1:{}/ws/terminal", port);
|
||||
let control_url = format!("ws://127.0.0.1:{}/ws/control", port);
|
||||
let terminal_tcp = TcpStream::connect(format!("127.0.0.1:{}", port))
|
||||
.await
|
||||
.unwrap();
|
||||
let (terminal_ws, _) = tokio_tungstenite::client_async_with_config(
|
||||
&terminal_url,
|
||||
MaybeTls::Plain(terminal_tcp),
|
||||
None,
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
let control_tcp = TcpStream::connect(format!("127.0.0.1:{}", port))
|
||||
.await
|
||||
.unwrap();
|
||||
let (control_ws, _) = tokio_tungstenite::client_async_with_config(
|
||||
&control_url,
|
||||
MaybeTls::Plain(control_tcp),
|
||||
None,
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
WebSocketConnections {
|
||||
terminal_ws,
|
||||
control_ws,
|
||||
web_client_id: "test-nested".to_string(),
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[serial]
|
||||
async fn test_nested_ping_frame_is_answered_locally_and_stripped_from_forwarded_stdin() {
|
||||
let (port, server, _server_handle) = mock_ws_server::MockWsServer::start().await;
|
||||
let connections = connect_mock_ws(port).await;
|
||||
|
||||
let (stdin_tx, stdin_rx) = mpsc::unbounded_channel();
|
||||
let (_signal_tx, signal_rx) = mpsc::unbounded_channel();
|
||||
let os_input = TestClientOsApi::new(stdin_rx, signal_rx);
|
||||
let stdout_buffer = os_input.stdout_buffer.clone();
|
||||
let os_input = Box::new(os_input);
|
||||
|
||||
let loop_handle =
|
||||
tokio::spawn(
|
||||
async move { run_remote_client_terminal_loop(os_input, connections, None).await },
|
||||
);
|
||||
|
||||
let mut chunk = b"abc".to_vec();
|
||||
chunk.extend_from_slice(&zellij_utils::nested_session::encode_frame(
|
||||
&zellij_utils::nested_session::NestedSessionMessage::Ping,
|
||||
));
|
||||
chunk.extend_from_slice(b"def");
|
||||
stdin_tx.send(chunk).unwrap();
|
||||
|
||||
tokio::time::sleep(Duration::from_millis(200)).await;
|
||||
let received = tokio::time::timeout(
|
||||
Duration::from_secs(1),
|
||||
server.client_to_terminal_rx.lock().unwrap().recv(),
|
||||
)
|
||||
.await
|
||||
.expect("Timeout")
|
||||
.expect("No message");
|
||||
match received {
|
||||
Message::Binary(data) => assert_eq!(data, b"abcdef".to_vec()),
|
||||
_ => panic!("Expected Binary message, got: {:?}", received),
|
||||
}
|
||||
|
||||
let pong_frame = zellij_utils::nested_session::encode_frame(
|
||||
&zellij_utils::nested_session::NestedSessionMessage::Pong,
|
||||
);
|
||||
let stdout_bytes = stdout_buffer.lock().unwrap().clone();
|
||||
assert!(
|
||||
stdout_bytes
|
||||
.windows(pong_frame.len())
|
||||
.any(|window| window == &pong_frame[..]),
|
||||
"expected a pong frame on stdout, got: {:?}",
|
||||
stdout_bytes
|
||||
);
|
||||
|
||||
drop(stdin_tx);
|
||||
let _ = tokio::time::timeout(Duration::from_secs(2), loop_handle)
|
||||
.await
|
||||
.expect("Loop didn't exit")
|
||||
.unwrap();
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[serial]
|
||||
async fn test_nested_announce_ack_is_relayed_over_control_websocket() {
|
||||
let (port, server, _server_handle) = mock_ws_server::MockWsServer::start().await;
|
||||
let connections = connect_mock_ws(port).await;
|
||||
|
||||
let (stdin_tx, stdin_rx) = mpsc::unbounded_channel();
|
||||
let (_signal_tx, signal_rx) = mpsc::unbounded_channel();
|
||||
let os_input = Box::new(TestClientOsApi::new(stdin_rx, signal_rx));
|
||||
|
||||
let loop_handle =
|
||||
tokio::spawn(
|
||||
async move { run_remote_client_terminal_loop(os_input, connections, None).await },
|
||||
);
|
||||
|
||||
let announce_ack = zellij_utils::nested_session::NestedSessionMessage::AnnounceAck {
|
||||
ancestry: vec!["outer-host".to_owned()],
|
||||
capabilities: vec![zellij_utils::nested_session::NestedSessionCapability::NestedControl],
|
||||
descend_keys: vec![],
|
||||
};
|
||||
stdin_tx
|
||||
.send(zellij_utils::nested_session::encode_frame(&announce_ack))
|
||||
.unwrap();
|
||||
|
||||
tokio::time::sleep(Duration::from_millis(200)).await;
|
||||
let mut relayed_payload = None;
|
||||
loop {
|
||||
let received = tokio::time::timeout(
|
||||
Duration::from_secs(1),
|
||||
server.client_to_control_rx.lock().unwrap().recv(),
|
||||
)
|
||||
.await
|
||||
.expect("Timeout")
|
||||
.expect("No message");
|
||||
if let Message::Text(text) = received {
|
||||
let control_msg: WebClientToWebServerControlMessage =
|
||||
serde_json::from_str(&text).unwrap();
|
||||
if let WebClientToWebServerControlMessagePayload::NestedSessionFrameFromHost {
|
||||
payload_bytes,
|
||||
} = control_msg.payload
|
||||
{
|
||||
assert_eq!(control_msg.web_client_id, "test-nested");
|
||||
relayed_payload = Some(payload_bytes);
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
assert_eq!(
|
||||
zellij_utils::nested_session::decode_payload(&relayed_payload.unwrap()),
|
||||
Some(announce_ack)
|
||||
);
|
||||
|
||||
drop(stdin_tx);
|
||||
let _ = tokio::time::timeout(Duration::from_secs(2), loop_handle)
|
||||
.await
|
||||
.expect("Loop didn't exit")
|
||||
.unwrap();
|
||||
}
|
||||
|
||||
@@ -16,6 +16,7 @@ pub enum WebClientToWebServerControlMessagePayload {
|
||||
TerminalSizeSettled(Size),
|
||||
TerminalMetrics(TerminalMetricsPayload),
|
||||
SoftKeyboardVisibilityChanged { visible: bool },
|
||||
NestedSessionFrameFromHost { payload_bytes: Vec<u8> },
|
||||
}
|
||||
|
||||
#[derive(Serialize, Deserialize, Debug, Clone)]
|
||||
|
||||
@@ -255,6 +255,7 @@ pub fn zellij_server_listener(
|
||||
// Subscribe-only messages — not relevant for web clients
|
||||
Some(ServerToClientMsg::PaneRenderUpdate { .. }) => {},
|
||||
Some(ServerToClientMsg::SubscribedPaneClosed { .. }) => {},
|
||||
Some(ServerToClientMsg::EmitNestedSessionFrame { .. }) => {},
|
||||
Some(ServerToClientMsg::ForwardQueryToHost { token, .. }) => {
|
||||
// Reply immediately with empty reply_bytes.
|
||||
// This is the existing convention that signals
|
||||
|
||||
@@ -99,6 +99,9 @@ async fn handle_ws_control(
|
||||
WebClientToWebServerControlMessagePayload::SoftKeyboardVisibilityChanged {
|
||||
visible,
|
||||
} => ClientToServerMsg::SoftKeyboardVisibilityChanged { visible },
|
||||
WebClientToWebServerControlMessagePayload::NestedSessionFrameFromHost {
|
||||
payload_bytes,
|
||||
} => ClientToServerMsg::NestedSessionFrameFromHost { payload_bytes },
|
||||
};
|
||||
|
||||
let _ = client_connection.send_to_server(client_msg);
|
||||
|
||||
Reference in New Issue
Block a user