#![expect(clippy::print_stderr, reason = "CLI/TUI output after terminal restore")]
use std::io;
use std::io::Read;
use std::io::Write;
use std::panic;
use std::str::FromStr;
use std::sync::Arc;
use std::sync::Mutex;
use std::sync::atomic::AtomicBool;
use std::sync::atomic::Ordering;
use std::thread;
use std::time::Duration;
use std::time::Instant;
use crate::adapters::RpcValue;
use crate::adapters::rpc::params;
use crate::app::rpc_client::RpcStream;
use crate::app::rpc_client::call_stream_with_params;
use crate::app::rpc_client::call_with_params;
use crate::common::Colors;
use crate::domain::session_types::TerminalSize;
use crate::infra::ipc::ClientError;
use crate::infra::ipc::DaemonClient;
use crate::infra::terminal::key_to_escape_sequence;
use base64::Engine;
use base64::engine::general_purpose::STANDARD;
use crossbeam_channel as channel;
use crossterm::cursor;
use crossterm::event;
use crossterm::event::Event;
use crossterm::event::KeyCode;
use crossterm::event::KeyEventKind;
use crossterm::event::KeyModifiers;
use crossterm::execute;
use crossterm::queue;
use crossterm::style;
use crossterm::terminal;
use crossterm::terminal::disable_raw_mode;
use crossterm::terminal::enable_raw_mode;
pub use crate::app::error::AttachError;
type PanicHook = Box<dyn Fn(&panic::PanicHookInfo<'_>) + Sync + Send + 'static>;
type SharedPanicHook = Arc<Mutex<Option<PanicHook>>>;
#[must_use = "TerminalGuard must be held for the duration of the attach session"]
struct TerminalGuard {
panic_hook_guard: TerminalPanicHookGuard,
}
impl TerminalGuard {
fn new() -> Result<Self, AttachError> {
enable_raw_mode().map_err(AttachError::Terminal)?;
let panic_hook_guard = TerminalPanicHookGuard::install();
let mut stdout = io::stdout();
prepare_terminal_with_rollback(&mut stdout, prepare_terminal)
.map_err(AttachError::Terminal)?;
Ok(Self { panic_hook_guard })
}
}
impl Drop for TerminalGuard {
fn drop(&mut self) {
restore_terminal_silently();
}
}
#[must_use = "TerminalPanicHookGuard must be held for the duration of the attach session"]
struct TerminalPanicHookGuard {
previous_hook: Option<SharedPanicHook>,
}
impl TerminalPanicHookGuard {
fn install() -> Self {
Self {
previous_hook: Some(install_terminal_panic_hook()),
}
}
fn restore(&mut self) {
if thread::panicking() {
return;
}
let Some(previous_hook) = self.previous_hook.take() else {
return;
};
let current_hook = panic::take_hook();
drop(current_hook);
let previous_hook = {
let mut previous_hook_guard = previous_hook
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
previous_hook_guard.take()
};
if let Some(previous_hook) = previous_hook {
panic::set_hook(previous_hook);
}
}
#[cfg(test)]
fn has_previous_hook(&self) -> bool {
self.previous_hook.is_some()
}
}
impl Drop for TerminalPanicHookGuard {
fn drop(&mut self) {
self.restore();
}
}
#[derive(Debug, Clone, Copy)]
pub enum AttachMode {
Tty,
Stream,
}
#[derive(Debug, Clone)]
pub struct DetachKeys {
sequence: Vec<u8>,
display: String,
}
impl DetachKeys {
pub fn is_disabled(&self) -> bool {
self.sequence.is_empty()
}
pub fn bytes(&self) -> &[u8] {
&self.sequence
}
pub fn display(&self) -> &str {
&self.display
}
fn disabled() -> Self {
Self {
sequence: Vec::new(),
display: "disabled".to_string(),
}
}
}
impl Default for DetachKeys {
fn default() -> Self {
Self {
sequence: vec![0x10, 0x02],
display: "Ctrl-P Ctrl-B".to_string(),
}
}
}
impl FromStr for DetachKeys {
type Err = String;
fn from_str(value: &str) -> Result<Self, Self::Err> {
let trimmed = value.trim();
if trimmed.is_empty() {
return Err("detach keys cannot be empty".to_string());
}
if trimmed.eq_ignore_ascii_case("none") {
return Ok(Self::disabled());
}
let tokens: Vec<&str> = trimmed
.split(',')
.map(str::trim)
.filter(|token| !token.is_empty())
.collect();
if tokens.is_empty() {
return Err("detach keys cannot be empty".to_string());
}
let mut sequence = Vec::with_capacity(tokens.len());
let mut display_tokens = Vec::with_capacity(tokens.len());
for token in tokens {
let (byte, display) = parse_detach_key_token(token)?;
sequence.push(byte);
display_tokens.push(display);
}
Ok(Self {
sequence,
display: display_tokens.join(" "),
})
}
}
pub fn attach_ipc<C: DaemonClient>(
client: &mut C,
session_id: &str,
mode: AttachMode,
detach_keys: DetachKeys,
) -> Result<(), AttachError> {
eprintln!(
"{} Attaching to session {}...",
Colors::dim("[attach]"),
Colors::session_id(session_id)
);
match mode {
AttachMode::Tty => {
if detach_keys.is_disabled() {
eprintln!(
"{} Detach keys disabled (use --detach-keys to enable).",
Colors::success("Connected!")
);
} else {
eprintln!(
"{} Press {} to detach.",
Colors::success("Connected!"),
Colors::bold(detach_keys.display())
);
}
eprintln!();
let term_guard = TerminalGuard::new()?;
let stdout = Arc::new(Mutex::new(io::stdout()));
let initial_resize_warning = match terminal::size().map_err(AttachError::Terminal) {
Ok((cols, rows)) => TerminalSize::try_new(cols, rows).ok().and_then(|size| {
sync_attach_resize(client, session_id, size)
.err()
.map(|error| attach_resize_warning(&error))
}),
Err(error) => return Err(error),
};
if let Ok(mut guard) = stdout.lock() {
render_initial_screen(client, session_id, &mut *guard);
if let Some(message) = initial_resize_warning.as_deref() {
render_status_line(&mut *guard, Some(message));
}
}
let result = attach_ipc_loop(client, session_id, &detach_keys, stdout);
drop(term_guard);
result
}
AttachMode::Stream => {
let stdout = Arc::new(Mutex::new(io::stdout()));
attach_stream_loop(client, session_id, stdout)
}
}?;
eprintln!();
eprintln!(
"{} Detached from session {}",
Colors::dim("[attach]"),
Colors::session_id(session_id)
);
Ok(())
}
fn render_initial_screen<C: DaemonClient>(
client: &mut C,
session_id: &str,
stdout: &mut impl Write,
) {
let params = params::SnapshotParams {
session: Some(session_id.to_string()),
include_cursor: true,
include_render: true,
..Default::default()
};
let snapshot = match call_with_params(client, "snapshot", params) {
Ok(snapshot) => snapshot,
Err(_) => return,
};
let rendered = snapshot.get("rendered").and_then(|v| v.as_str());
let screenshot = snapshot.get("screenshot").and_then(|v| v.as_str());
let screen = match rendered.or(screenshot) {
Some(screen) => screen,
None => return,
};
if screen.is_empty() {
return;
}
let _ = queue!(
stdout,
terminal::Clear(terminal::ClearType::All),
cursor::MoveTo(0, 0),
style::SetAttribute(style::Attribute::Reset),
style::ResetColor,
style::Print(screen)
);
if let Some(cursor) = snapshot.get("cursor") {
let row = cursor
.get("row")
.and_then(|v| v.as_u64())
.unwrap_or(0)
.min(u16::MAX as u64) as u16;
let col = cursor
.get("col")
.and_then(|v| v.as_u64())
.unwrap_or(0)
.min(u16::MAX as u64) as u16;
let visible = cursor
.get("visible")
.and_then(|v| v.as_bool())
.unwrap_or(true);
let _ = queue!(stdout, cursor::MoveTo(col, row));
if visible {
let _ = queue!(stdout, cursor::Show);
} else {
let _ = queue!(stdout, cursor::Hide);
}
}
let _ = stdout.flush();
}
fn attach_ipc_loop<C: DaemonClient>(
client: &mut C,
session_id: &str,
detach_keys: &DetachKeys,
stdout: Arc<Mutex<io::Stdout>>,
) -> Result<(), AttachError> {
let mut detach_detector = DetachDetector::new(detach_keys);
let mut paste_burst = PasteBurstState::default();
let mut hint_active = false;
let stream = call_stream_with_params(
client,
"attach_stream",
params::SessionParams {
session: Some(session_id.to_string()),
},
)
.map_err(|e| AttachError::PtyRead(format_client_error(&e)))?;
let mut output_worker = start_attach_stream_output(stream, Arc::clone(&stdout), false)?;
let event_worker = spawn_event_reader();
let paste_flush_tick = channel::tick(ATTACH_PASTE_BURST_CHAR_INTERVAL);
loop {
channel::select! {
recv(output_worker.receiver()) -> result => {
match result {
Ok(Ok(())) => break,
Ok(Err(err)) => return Err(err),
Err(_) => break,
}
}
recv(paste_flush_tick) -> _ => {
if let Some(input) = paste_burst.flush_ready(Instant::now())
&& process_attach_input(
client,
session_id,
detach_keys,
&mut detach_detector,
&mut hint_active,
&stdout,
input,
)?
{
finish_detach(&mut output_worker, &event_worker);
break;
}
}
recv(event_worker.receiver()) -> msg => {
match msg {
Ok(EventMessage::Event(Event::Key(key_event))) => {
let now = Instant::now();
if let Some(character) = plain_char_for_paste_burst(&key_event) {
if let Some(input) = paste_burst.on_plain_char(character, now)
&& process_attach_input(
client,
session_id,
detach_keys,
&mut detach_detector,
&mut hint_active,
&stdout,
input,
)?
{
finish_detach(&mut output_worker, &event_worker);
break;
}
continue;
}
if let Some(input) = paste_burst.flush_all()
&& process_attach_input(
client,
session_id,
detach_keys,
&mut detach_detector,
&mut hint_active,
&stdout,
input,
)?
{
finish_detach(&mut output_worker, &event_worker);
break;
}
if let Some(bytes) = key_event_to_bytes(&key_event)
&& process_attach_input(
client,
session_id,
detach_keys,
&mut detach_detector,
&mut hint_active,
&stdout,
BufferedAttachInput::normal(bytes),
)?
{
finish_detach(&mut output_worker, &event_worker);
break;
}
}
Ok(EventMessage::Event(Event::Paste(data))) => {
if let Some(input) = paste_burst.flush_all()
&& process_attach_input(
client,
session_id,
detach_keys,
&mut detach_detector,
&mut hint_active,
&stdout,
input,
)?
{
finish_detach(&mut output_worker, &event_worker);
break;
}
if !data.is_empty()
&& process_attach_input(
client,
session_id,
detach_keys,
&mut detach_detector,
&mut hint_active,
&stdout,
BufferedAttachInput::bypass(data.into_bytes()),
)?
{
finish_detach(&mut output_worker, &event_worker);
break;
}
}
Ok(EventMessage::Event(Event::Resize(cols, rows))) => {
if let Some(input) = paste_burst.flush_all()
&& process_attach_input(
client,
session_id,
detach_keys,
&mut detach_detector,
&mut hint_active,
&stdout,
input,
)?
{
finish_detach(&mut output_worker, &event_worker);
break;
}
if let Ok(size) = TerminalSize::try_new(cols, rows)
&& let Err(error) = sync_attach_resize(client, session_id, size)
{
announce_attach_warning(&stdout, &attach_resize_warning(&error));
}
}
Ok(EventMessage::Event(_)) => {
if let Some(input) = paste_burst.flush_all()
&& process_attach_input(
client,
session_id,
detach_keys,
&mut detach_detector,
&mut hint_active,
&stdout,
input,
)?
{
finish_detach(&mut output_worker, &event_worker);
break;
}
}
Ok(EventMessage::Error) => return Err(AttachError::EventRead),
Err(_) => return Err(AttachError::EventRead),
}
}
}
}
Ok(())
}
fn attach_stream_loop<C: DaemonClient>(
client: &mut C,
session_id: &str,
stdout: Arc<Mutex<io::Stdout>>,
) -> Result<(), AttachError> {
let stream = call_stream_with_params(
client,
"attach_stream",
params::SessionParams {
session: Some(session_id.to_string()),
},
)
.map_err(|e| AttachError::PtyRead(format_client_error(&e)))?;
let output_worker = start_attach_stream_output(stream, Arc::clone(&stdout), true)?;
let stdin_worker = spawn_stdin_reader();
let mut stdin_active = true;
loop {
if stdin_active {
channel::select! {
recv(output_worker.receiver()) -> result => {
match result {
Ok(Ok(())) => return Ok(()),
Ok(Err(err)) => return Err(err),
Err(_) => return Ok(()),
}
}
recv(stdin_worker.receiver()) -> msg => {
match msg {
Ok(StdinMessage::Data(data)) => {
if !data.is_empty() {
let data_b64 = STANDARD.encode(&data);
let params = params::PtyWriteParams {
session: Some(session_id.to_string()),
data: data_b64,
};
if let Err(e) = call_with_params(client, "pty_write", params) {
return Err(AttachError::PtyWrite(format_client_error(&e)));
}
}
}
Ok(StdinMessage::Eof) | Ok(StdinMessage::Error) => {
stdin_active = false;
stdin_worker.cancel();
}
Err(_) => {
stdin_active = false;
stdin_worker.cancel();
}
}
}
}
} else {
match output_worker.receiver().recv() {
Ok(Ok(())) => return Ok(()),
Ok(Err(err)) => return Err(err),
Err(_) => return Ok(()),
}
}
}
}
fn format_client_error(error: &ClientError) -> String {
let mut msg = error.to_string();
if let Some(suggestion) = error.suggestion() {
msg.push_str(&format!(" ({suggestion})"));
}
msg
}
fn prepare_terminal<W: Write>(stdout: &mut W) -> io::Result<()> {
execute!(
stdout,
terminal::EnterAlternateScreen,
terminal::Clear(terminal::ClearType::All),
cursor::MoveTo(0, 0),
style::SetAttribute(style::Attribute::Reset),
style::ResetColor,
terminal::EnableLineWrap,
cursor::Show,
event::DisableMouseCapture,
event::DisableFocusChange,
event::EnableBracketedPaste
)?;
stdout.flush()
}
fn prepare_terminal_with_rollback<W, F>(stdout: &mut W, prepare: F) -> io::Result<()>
where
W: Write,
F: FnOnce(&mut W) -> io::Result<()>,
{
if let Err(err) = prepare(stdout) {
let _ = disable_raw_mode();
let _ = reset_terminal_modes(stdout);
return Err(err);
}
Ok(())
}
fn restore_terminal_silently() {
let _ = disable_raw_mode();
let mut stdout = io::stdout();
let _ = reset_terminal_modes(&mut stdout);
}
fn install_terminal_panic_hook() -> SharedPanicHook {
let previous_hook: SharedPanicHook = Arc::new(Mutex::new(Some(panic::take_hook())));
let previous_hook_for_panic = Arc::clone(&previous_hook);
panic::set_hook(Box::new(move |panic_info| {
restore_terminal_silently();
let previous_hook_guard = previous_hook_for_panic
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
if let Some(previous_hook) = previous_hook_guard.as_ref() {
previous_hook(panic_info);
}
}));
previous_hook
}
fn reset_terminal_modes(stdout: &mut impl Write) -> io::Result<()> {
execute!(
stdout,
style::SetAttribute(style::Attribute::Reset),
style::ResetColor,
cursor::Show,
terminal::EnableLineWrap,
event::DisableMouseCapture,
event::DisableFocusChange,
event::DisableBracketedPaste,
terminal::LeaveAlternateScreen
)?;
stdout.flush()
}
#[derive(Debug)]
struct DetachDetector {
sequence: Vec<u8>,
matched: usize,
}
impl DetachDetector {
fn new(detach_keys: &DetachKeys) -> Self {
Self {
sequence: detach_keys.bytes().to_vec(),
matched: 0,
}
}
fn is_partial_match(&self) -> bool {
self.matched > 0
}
fn consume(&mut self, bytes: &[u8]) -> (Vec<u8>, bool) {
let mut output = Vec::new();
for &byte in bytes {
if self.consume_byte(byte, &mut output) {
return (output, true);
}
}
(output, false)
}
fn consume_byte(&mut self, byte: u8, output: &mut Vec<u8>) -> bool {
if self.sequence.is_empty() {
output.push(byte);
return false;
}
if byte == self.sequence[self.matched] {
self.matched += 1;
if self.matched == self.sequence.len() {
self.matched = 0;
return true;
}
return false;
}
if self.matched > 0 {
output.extend_from_slice(&self.sequence[..self.matched]);
self.matched = 0;
output.push(byte);
return false;
}
output.push(byte);
false
}
fn cancel_partial_match(&mut self) -> Vec<u8> {
if self.matched == 0 {
return Vec::new();
}
let pending = self.sequence[..self.matched].to_vec();
self.matched = 0;
pending
}
}
#[derive(Debug, PartialEq, Eq)]
struct BufferedAttachInput {
bytes: Vec<u8>,
bypass_detach: bool,
}
impl BufferedAttachInput {
fn normal(bytes: Vec<u8>) -> Self {
Self {
bytes,
bypass_detach: false,
}
}
fn bypass(bytes: Vec<u8>) -> Self {
Self {
bytes,
bypass_detach: true,
}
}
}
#[derive(Debug, Default)]
struct PasteBurstState {
pending_first_char: Option<(char, Instant)>,
buffer: String,
last_plain_char_at: Option<Instant>,
}
impl PasteBurstState {
fn on_plain_char(&mut self, character: char, now: Instant) -> Option<BufferedAttachInput> {
if !self.buffer.is_empty() {
if self
.last_plain_char_at
.is_some_and(|last| now.duration_since(last) <= ATTACH_PASTE_BURST_CHAR_INTERVAL)
{
self.buffer.push(character);
self.last_plain_char_at = Some(now);
return None;
}
let flushed = self.take_buffer();
self.pending_first_char = Some((character, now));
return Some(flushed);
}
if let Some((pending, pending_at)) = self.pending_first_char {
if now.duration_since(pending_at) <= ATTACH_PASTE_BURST_CHAR_INTERVAL {
self.pending_first_char = None;
self.buffer.push(pending);
self.buffer.push(character);
self.last_plain_char_at = Some(now);
return None;
}
self.pending_first_char = Some((character, now));
return Some(Self::single_char_input(pending));
}
self.pending_first_char = Some((character, now));
None
}
fn flush_ready(&mut self, now: Instant) -> Option<BufferedAttachInput> {
if !self.buffer.is_empty() {
if self
.last_plain_char_at
.is_some_and(|last| now.duration_since(last) > ATTACH_PASTE_BURST_CHAR_INTERVAL)
{
return Some(self.take_buffer());
}
return None;
}
if let Some((pending, pending_at)) = self.pending_first_char
&& now.duration_since(pending_at) > ATTACH_PASTE_BURST_CHAR_INTERVAL
{
self.pending_first_char = None;
return Some(Self::single_char_input(pending));
}
None
}
fn flush_all(&mut self) -> Option<BufferedAttachInput> {
if !self.buffer.is_empty() {
return Some(self.take_buffer());
}
self.pending_first_char
.take()
.map(|(pending, _)| Self::single_char_input(pending))
}
fn single_char_input(character: char) -> BufferedAttachInput {
let mut buf = [0u8; 4];
let bytes = character.encode_utf8(&mut buf).as_bytes().to_vec();
BufferedAttachInput::normal(bytes)
}
fn take_buffer(&mut self) -> BufferedAttachInput {
self.last_plain_char_at = None;
let bytes = self.buffer.as_bytes().to_vec();
self.buffer.clear();
BufferedAttachInput::bypass(bytes)
}
}
fn render_status_line(stdout: &mut impl Write, message: Option<&str>) {
let (cols, rows) = match terminal::size() {
Ok(size) => size,
Err(_) => return,
};
let row = rows.saturating_sub(1);
let mut line = message.unwrap_or("").to_string();
let max_len = cols as usize;
if line.len() > max_len {
line.truncate(max_len);
}
if line.len() < max_len {
let pad = max_len - line.len();
line.reserve(pad);
line.extend(std::iter::repeat_n(' ', pad));
}
let _ = queue!(
stdout,
cursor::SavePosition,
cursor::MoveTo(0, row),
style::SetAttribute(style::Attribute::Reset),
style::ResetColor,
style::Print(line),
cursor::RestorePosition
);
let _ = stdout.flush();
}
fn sync_detach_hint(
stdout: &Arc<Mutex<io::Stdout>>,
detach_keys: &DetachKeys,
detach_detector: &DetachDetector,
hint_active: &mut bool,
) {
if detach_keys.is_disabled() {
return;
}
let now_active = detach_detector.is_partial_match();
if now_active == *hint_active {
return;
}
if let Ok(mut guard) = stdout.lock() {
render_status_line(
&mut *guard,
if now_active {
Some("Detach: sequence started, press remaining keys to detach")
} else {
None
},
);
}
*hint_active = now_active;
}
fn finish_detach(
output_worker: &mut AttachOutputWorker,
event_worker: &AttachReaderWorker<EventMessage>,
) {
output_worker.abort();
let _ = output_worker
.receiver()
.recv_timeout(ATTACH_OUTPUT_SHUTDOWN_WAIT);
event_worker.cancel();
}
fn write_session_bytes<C: DaemonClient>(
client: &mut C,
session_id: &str,
data: &[u8],
) -> Result<(), AttachError> {
if data.is_empty() {
return Ok(());
}
let data_b64 = STANDARD.encode(data);
let params = params::PtyWriteParams {
session: Some(session_id.to_string()),
data: data_b64,
};
call_with_params(client, "pty_write", params)
.map(|_| ())
.map_err(|e| AttachError::PtyWrite(format_client_error(&e)))
}
fn sync_attach_resize<C: DaemonClient>(
client: &mut C,
session_id: &str,
size: TerminalSize,
) -> Result<(), ClientError> {
let params = params::ResizeParams {
size,
session: Some(session_id.to_string()),
};
call_with_params(client, "resize", params).map(|_| ())
}
fn attach_resize_warning(error: &ClientError) -> String {
format!("Resize sync failed: {}", format_client_error(error))
}
fn announce_attach_warning(stdout: &Arc<Mutex<io::Stdout>>, message: &str) {
tracing::warn!(warning = %message, "Attach warning");
if let Ok(mut guard) = stdout.lock() {
render_status_line(&mut *guard, Some(message));
}
}
fn process_attach_input<C: DaemonClient>(
client: &mut C,
session_id: &str,
detach_keys: &DetachKeys,
detach_detector: &mut DetachDetector,
hint_active: &mut bool,
stdout: &Arc<Mutex<io::Stdout>>,
input: BufferedAttachInput,
) -> Result<bool, AttachError> {
let (to_send, detach) = if input.bypass_detach {
let mut bytes = detach_detector.cancel_partial_match();
bytes.extend_from_slice(&input.bytes);
(bytes, false)
} else {
detach_detector.consume(&input.bytes)
};
sync_detach_hint(stdout, detach_keys, detach_detector, hint_active);
if detach {
return Ok(true);
}
write_session_bytes(client, session_id, &to_send)?;
Ok(false)
}
enum StdinMessage {
Data(Vec<u8>),
Eof,
Error,
}
const ATTACH_STDIN_CHANNEL_CAPACITY: usize = 64;
const ATTACH_INPUT_POLL_INTERVAL: Duration = Duration::from_millis(50);
const ATTACH_THREAD_JOIN_TIMEOUT: Duration = Duration::from_millis(500);
const ATTACH_OUTPUT_SHUTDOWN_WAIT: Duration = Duration::from_millis(500);
const ATTACH_PASTE_BURST_CHAR_INTERVAL: Duration = Duration::from_millis(8);
struct AttachReaderWorker<T> {
rx: channel::Receiver<T>,
cancelled: Arc<AtomicBool>,
join: Option<thread::JoinHandle<()>>,
name: &'static str,
}
impl<T> AttachReaderWorker<T> {
fn receiver(&self) -> &channel::Receiver<T> {
&self.rx
}
fn cancel(&self) {
self.cancelled.store(true, Ordering::Relaxed);
}
fn shutdown(&mut self, timeout: Duration) {
self.cancel();
join_thread_with_timeout(&mut self.join, timeout, self.name);
}
}
impl<T> Drop for AttachReaderWorker<T> {
fn drop(&mut self) {
self.shutdown(ATTACH_THREAD_JOIN_TIMEOUT);
}
}
struct AttachOutputWorker {
done_rx: channel::Receiver<Result<(), AttachError>>,
join: Option<thread::JoinHandle<()>>,
shutdown_signal: Option<Arc<dyn Fn() + Send + Sync>>,
}
impl AttachOutputWorker {
fn receiver(&self) -> &channel::Receiver<Result<(), AttachError>> {
&self.done_rx
}
fn abort(&mut self) {
if let Some(shutdown_signal) = self.shutdown_signal.take() {
shutdown_signal();
}
}
fn shutdown(&mut self, timeout: Duration) {
self.abort();
join_thread_with_timeout(&mut self.join, timeout, "attach-stream-output");
}
}
impl Drop for AttachOutputWorker {
fn drop(&mut self) {
self.shutdown(ATTACH_THREAD_JOIN_TIMEOUT);
}
}
fn join_thread_with_timeout(
join: &mut Option<thread::JoinHandle<()>>,
timeout: Duration,
name: &'static str,
) {
let Some(handle) = join.take() else {
return;
};
let deadline = Instant::now() + timeout;
while !handle.is_finished() {
if Instant::now() >= deadline {
tracing::warn!(
thread = name,
timeout_ms = timeout.as_millis(),
"Timed out joining attach helper thread"
);
return;
}
thread::park_timeout(Duration::from_millis(10));
}
let _ = handle.join();
}
fn spawn_stdin_reader() -> AttachReaderWorker<StdinMessage> {
let (tx, rx) = channel::bounded(ATTACH_STDIN_CHANNEL_CAPACITY);
let span = tracing::debug_span!("attach_stdin_reader");
let builder = thread::Builder::new().name("attach-stdin".to_string());
let cancelled = Arc::new(AtomicBool::new(false));
let cancelled_for_thread = Arc::clone(&cancelled);
let tx_thread = tx.clone();
let join = match builder.spawn(move || {
let _guard = span.enter();
let mut stdin = io::stdin();
let mut buf = [0u8; 4096];
loop {
if cancelled_for_thread.load(Ordering::Relaxed) {
break;
}
#[cfg(unix)]
{
match stdin_ready(ATTACH_INPUT_POLL_INTERVAL) {
Ok(true) => {}
Ok(false) => continue,
Err(_) => {
let _ = tx_thread.send(StdinMessage::Error);
break;
}
}
}
match stdin.read(&mut buf) {
Ok(0) => {
let _ = tx_thread.send(StdinMessage::Eof);
break;
}
Ok(n) => {
if tx_thread
.send(StdinMessage::Data(buf[..n].to_vec()))
.is_err()
{
break;
}
}
Err(_) => {
let _ = tx_thread.send(StdinMessage::Error);
break;
}
}
}
}) {
Ok(handle) => Some(handle),
Err(err) => {
tracing::warn!(error = %err, "Failed to spawn stdin reader");
let _ = tx.send(StdinMessage::Error);
None
}
};
AttachReaderWorker {
rx,
cancelled,
join,
name: "attach-stdin",
}
}
enum EventMessage {
Event(Event),
Error,
}
const ATTACH_EVENT_CHANNEL_CAPACITY: usize = 256;
#[cfg(unix)]
fn stdin_ready(timeout: Duration) -> io::Result<bool> {
use std::os::fd::AsRawFd;
let mut fds = [libc::pollfd {
fd: io::stdin().as_raw_fd(),
events: libc::POLLIN,
revents: 0,
}];
let timeout_ms = timeout.as_millis().min(i32::MAX as u128) as i32;
loop {
let rc = unsafe { libc::poll(fds.as_mut_ptr(), 1, timeout_ms) };
if rc > 0 {
return Ok(fds[0].revents & libc::POLLIN != 0);
}
if rc == 0 {
return Ok(false);
}
let err = io::Error::last_os_error();
if err.kind() == io::ErrorKind::Interrupted {
continue;
}
return Err(err);
}
}
fn spawn_event_reader() -> AttachReaderWorker<EventMessage> {
let (tx, rx) = channel::bounded(ATTACH_EVENT_CHANNEL_CAPACITY);
let span = tracing::debug_span!("attach_event_reader");
let builder = thread::Builder::new().name("attach-events".to_string());
let cancelled = Arc::new(AtomicBool::new(false));
let cancelled_for_thread = Arc::clone(&cancelled);
let tx_thread = tx.clone();
let join = match builder.spawn(move || {
let _guard = span.enter();
loop {
if cancelled_for_thread.load(Ordering::Relaxed) {
break;
}
match event::poll(ATTACH_INPUT_POLL_INTERVAL) {
Ok(true) => {}
Ok(false) => continue,
Err(_) => {
let _ = tx_thread.send(EventMessage::Error);
break;
}
}
match event::read() {
Ok(ev) => {
if tx_thread.send(EventMessage::Event(ev)).is_err() {
break;
}
}
Err(_) => {
let _ = tx_thread.send(EventMessage::Error);
break;
}
}
}
}) {
Ok(handle) => Some(handle),
Err(err) => {
tracing::warn!(error = %err, "Failed to spawn event reader");
let _ = tx.send(EventMessage::Error);
None
}
};
AttachReaderWorker {
rx,
cancelled,
join,
name: "attach-events",
}
}
enum AttachStreamEvent {
Output { data: Vec<u8>, dropped_bytes: u64 },
Dropped(u64),
Closed,
}
fn parse_stream_event(value: RpcValue) -> Result<Option<AttachStreamEvent>, AttachError> {
let event = value.get("event").and_then(|v| v.as_str());
let Some(event) = event else {
return Ok(None);
};
match event {
"output" => {
let data_b64 = value.get("data").and_then(|v| v.as_str()).unwrap_or("");
if data_b64.is_empty() {
return Ok(None);
}
let data = STANDARD
.decode(data_b64)
.map_err(|e| AttachError::PtyRead(e.to_string()))?;
let dropped_bytes = value
.get("dropped_bytes")
.and_then(|v| v.as_u64())
.unwrap_or(0);
Ok(Some(AttachStreamEvent::Output {
data,
dropped_bytes,
}))
}
"dropped" => {
let dropped_bytes = value
.get("dropped_bytes")
.and_then(|v| v.as_u64())
.unwrap_or(0);
Ok(Some(AttachStreamEvent::Dropped(dropped_bytes)))
}
"closed" => Ok(Some(AttachStreamEvent::Closed)),
"ready" | "heartbeat" => Ok(None),
_ => Ok(None),
}
}
fn start_attach_stream_output(
stream: RpcStream,
stdout: Arc<Mutex<io::Stdout>>,
report_drops: bool,
) -> Result<AttachOutputWorker, AttachError> {
let (tx, rx) = channel::bounded(1);
let shutdown_signal = stream
.abort_handle()
.map(|handle| Arc::new(move || handle.abort()) as Arc<dyn Fn() + Send + Sync>);
let builder = thread::Builder::new().name("attach-stream-output".to_string());
let join = builder
.spawn(move || {
let result = stream_output_loop(stream, stdout, report_drops);
let _ = tx.send(result);
})
.map_err(|err| {
AttachError::PtyRead(format!("Failed to spawn attach output thread: {err}"))
})?;
Ok(AttachOutputWorker {
done_rx: rx,
join: Some(join),
shutdown_signal,
})
}
fn stream_output_loop(
mut stream: RpcStream,
stdout: Arc<Mutex<io::Stdout>>,
report_drops: bool,
) -> Result<(), AttachError> {
loop {
let next = stream
.next_result()
.map_err(|e| AttachError::PtyRead(format_client_error(&e)))?;
let Some(value) = next else {
return Ok(());
};
match parse_stream_event(value)? {
Some(AttachStreamEvent::Output {
data,
dropped_bytes,
}) => {
if let (true, Ok(mut guard)) = (!data.is_empty(), stdout.lock()) {
guard.write_all(&data).map_err(AttachError::Terminal)?;
guard.flush().map_err(AttachError::Terminal)?;
}
if report_drops && dropped_bytes > 0 {
eprintln!(
"{} Dropped {} bytes from stream buffer.",
Colors::warning("[attach]"),
dropped_bytes
);
}
}
Some(AttachStreamEvent::Dropped(dropped_bytes)) => {
if report_drops && dropped_bytes > 0 {
eprintln!(
"{} Dropped {} bytes from stream buffer.",
Colors::warning("[attach]"),
dropped_bytes
);
}
}
Some(AttachStreamEvent::Closed) => return Ok(()),
None => {}
}
}
}
fn plain_char_for_paste_burst(key_event: &event::KeyEvent) -> Option<char> {
if key_event.kind == KeyEventKind::Release || key_event.modifiers != KeyModifiers::NONE {
return None;
}
match key_event.code {
KeyCode::Char(character) => Some(character),
_ => None,
}
}
fn key_event_to_bytes(key_event: &event::KeyEvent) -> Option<Vec<u8>> {
if key_event.kind == KeyEventKind::Release {
return None;
}
match key_event.code {
KeyCode::Char(c) => key_char_to_bytes(c, key_event.modifiers),
KeyCode::F(n) => {
let key = format!("F{n}");
key_with_modifiers_to_bytes(&key, key_event.modifiers)
}
KeyCode::BackTab => key_to_escape_sequence("Shift+Tab"),
_ => {
let base = keycode_to_name(key_event.code)?;
key_with_modifiers_to_bytes(base, key_event.modifiers)
}
}
}
fn key_char_to_bytes(c: char, modifiers: KeyModifiers) -> Option<Vec<u8>> {
if modifiers.contains(KeyModifiers::CONTROL) {
let key = format_modified_char("Ctrl", c);
return key_to_escape_sequence(&key);
}
if modifiers.contains(KeyModifiers::ALT) {
let key = format_modified_char("Alt", c);
return key_to_escape_sequence(&key);
}
let mut buf = [0u8; 4];
let s = c.encode_utf8(&mut buf);
Some(s.as_bytes().to_vec())
}
fn key_with_modifiers_to_bytes(base: &str, modifiers: KeyModifiers) -> Option<Vec<u8>> {
if modifiers.contains(KeyModifiers::SHIFT) && base == "Tab" {
return key_to_escape_sequence("Shift+Tab");
}
if modifiers.contains(KeyModifiers::CONTROL) {
let key = format_modified_key("Ctrl", base);
return key_to_escape_sequence(&key);
}
if modifiers.contains(KeyModifiers::ALT) {
let key = format_modified_key("Alt", base);
return key_to_escape_sequence(&key);
}
key_to_escape_sequence(base)
}
fn format_modified_key(prefix: &str, base: &str) -> String {
let mut key = String::with_capacity(prefix.len() + 1 + base.len());
key.push_str(prefix);
key.push('+');
key.push_str(base);
key
}
fn format_modified_char(prefix: &str, c: char) -> String {
let mut key = String::with_capacity(prefix.len() + 2);
key.push_str(prefix);
key.push('+');
key.push(c);
key
}
fn keycode_to_name(code: KeyCode) -> Option<&'static str> {
match code {
KeyCode::Enter => Some("Enter"),
KeyCode::Tab => Some("Tab"),
KeyCode::Backspace => Some("Backspace"),
KeyCode::Delete => Some("Delete"),
KeyCode::Esc => Some("Escape"),
KeyCode::Up => Some("ArrowUp"),
KeyCode::Down => Some("ArrowDown"),
KeyCode::Right => Some("ArrowRight"),
KeyCode::Left => Some("ArrowLeft"),
KeyCode::Home => Some("Home"),
KeyCode::End => Some("End"),
KeyCode::PageUp => Some("PageUp"),
KeyCode::PageDown => Some("PageDown"),
KeyCode::Insert => Some("Insert"),
_ => None,
}
}
fn parse_detach_key_token(token: &str) -> Result<(u8, String), String> {
let trimmed = token.trim();
if trimmed.is_empty() {
return Err("detach keys cannot be empty".to_string());
}
let lower = trimmed.to_ascii_lowercase();
if lower.starts_with("ctrl-") || lower.starts_with("control-") {
let split_pos = match trimmed.find('-') {
Some(pos) => pos,
None => {
return Err("detach keys: ctrl- requires a key (e.g. ctrl-p)".to_string());
}
};
let rest = trimmed[split_pos + 1..].trim();
if rest.is_empty() {
return Err("detach keys: ctrl- requires a key (e.g. ctrl-p)".to_string());
}
let ch = if rest.eq_ignore_ascii_case("space") {
' '
} else {
let mut chars = rest.chars();
match (chars.next(), chars.next()) {
(Some(ch), None) => ch,
_ => return Err(format!("detach keys: unsupported ctrl key '{rest}'")),
}
};
let byte = ctrl_char_to_byte(ch)
.ok_or_else(|| format!("detach keys: unsupported ctrl key '{rest}'"))?;
let display = format!("Ctrl-{}", display_char(ch));
return Ok((byte, display));
}
if lower == "space" {
return Ok((b' ', "Space".to_string()));
}
let mut chars = trimmed.chars();
if let (Some(ch), None) = (chars.next(), chars.next()) {
if !ch.is_ascii() {
return Err("detach keys must be ASCII".to_string());
}
let display = display_char(ch);
return Ok((ch as u8, display));
}
Err(format!("detach keys: unsupported token '{trimmed}'"))
}
fn ctrl_char_to_byte(ch: char) -> Option<u8> {
if ch.is_ascii_lowercase() {
return Some(ch as u8 - b'a' + 1);
}
if ch.is_ascii_uppercase() {
return Some(ch as u8 - b'A' + 1);
}
match ch {
'[' => Some(0x1b),
'\\' => Some(0x1c),
']' => Some(0x1d),
'^' => Some(0x1e),
'_' => Some(0x1f),
'?' => Some(0x7f),
' ' | '@' => Some(0x00),
_ => None,
}
}
fn display_char(ch: char) -> String {
if ch == ' ' {
return "Space".to_string();
}
if ch.is_ascii_alphabetic() {
return ch.to_ascii_uppercase().to_string();
}
ch.to_string()
}
#[cfg(test)]
#[path = "attach_tests.rs"]
mod tests;