use std::fs::OpenOptions;
use std::io::Write;
use std::os::fd::{AsRawFd, RawFd};
use std::time::{Duration, Instant};
use super::color::{Theme, parse_osc_rgb, theme_from_rgb};
const OSC_QUERY: &[u8] = b"\x1b]11;?\x1b\\";
const TERMINATOR: &[u8] = b"\x1b\\";
const MAX_RESPONSE_LEN: usize = 64;
struct TtyQueryGuard {
fd: RawFd,
original: libc::termios,
}
impl TtyQueryGuard {
fn enter(fd: RawFd) -> Option<Self> {
let mut original: libc::termios = unsafe { std::mem::zeroed() };
if unsafe { libc::tcgetattr(fd, &mut original) } != 0 {
return None;
}
let mut raw = original;
raw.c_lflag &= !(libc::ECHO | libc::ICANON);
raw.c_cc[libc::VMIN] = 1;
raw.c_cc[libc::VTIME] = 0;
if unsafe { libc::tcsetattr(fd, libc::TCSANOW, &raw) } != 0 {
return None;
}
Some(Self { fd, original })
}
}
impl Drop for TtyQueryGuard {
fn drop(&mut self) {
unsafe {
let _ = libc::tcsetattr(self.fd, libc::TCSANOW, &self.original);
}
}
}
fn wait_readable(fd: RawFd, timeout: Duration) -> std::io::Result<bool> {
if fd < 0 || (fd as usize) >= libc::FD_SETSIZE {
return Ok(false);
}
let mut read_fds: libc::fd_set = unsafe { std::mem::zeroed() };
unsafe { libc::FD_SET(fd, &mut read_fds) };
let mut tv = libc::timeval {
tv_sec: timeout.as_secs() as libc::time_t,
tv_usec: timeout.subsec_micros() as libc::suseconds_t,
};
let ret = unsafe {
libc::select(
fd + 1,
&mut read_fds,
std::ptr::null_mut(),
std::ptr::null_mut(),
&mut tv,
)
};
if ret < 0 {
return Err(std::io::Error::last_os_error());
}
Ok(ret > 0 && unsafe { libc::FD_ISSET(fd, &read_fds) })
}
pub(super) fn read_until_terminator_or_deadline(fd: RawFd, deadline: Instant) -> Vec<u8> {
let mut response = Vec::with_capacity(32);
loop {
if response.len() >= MAX_RESPONSE_LEN {
break;
}
let Some(remaining) = deadline.checked_duration_since(Instant::now()) else {
break;
};
match wait_readable(fd, remaining) {
Ok(true) => {}
Ok(false) | Err(_) => break,
}
let mut byte = [0u8; 1];
let n = unsafe { libc::read(fd, byte.as_mut_ptr().cast(), 1) };
if n <= 0 {
break; }
response.push(byte[0]);
if response.ends_with(TERMINATOR) {
break;
}
}
response
}
fn parse_osc_response(response: &[u8]) -> Option<Theme> {
let text = std::str::from_utf8(response).ok()?;
let body = text
.strip_prefix("\x1b]")?
.strip_prefix("11;")?
.strip_suffix("\x1b\\")?;
let (r, g, b) = parse_osc_rgb(body)?;
Some(theme_from_rgb(r, g, b))
}
fn is_multiplexer_term() -> bool {
std::env::var("TERM")
.map(|term| term.starts_with("screen") || term.starts_with("tmux"))
.unwrap_or(false)
}
pub fn query_terminal_theme(budget: Duration) -> Option<Theme> {
if is_multiplexer_term() {
return None;
}
let mut tty = OpenOptions::new()
.read(true)
.write(true)
.open("/dev/tty")
.ok()?;
let fd = tty.as_raw_fd();
let _tty_guard = TtyQueryGuard::enter(fd)?;
let deadline = Instant::now() + budget;
tty.write_all(OSC_QUERY).ok()?;
tty.flush().ok()?;
let response = read_until_terminator_or_deadline(fd, deadline);
parse_osc_response(&response)
}
#[cfg(test)]
mod tests {
use super::*;
use std::os::unix::net::UnixStream;
use std::thread;
#[test]
fn parse_osc_response_extracts_theme_from_well_formed_reply() {
let response = b"\x1b]11;rgb:ffff/ffff/ffff\x1b\\";
assert_eq!(parse_osc_response(response), Some(Theme::Light));
}
#[test]
fn parse_osc_response_rejects_missing_terminator_or_prefix() {
assert_eq!(parse_osc_response(b"11;rgb:ffff/ffff/ffff\x1b\\"), None);
assert_eq!(parse_osc_response(b"\x1b]11;rgb:ffff/ffff/ffff"), None);
assert_eq!(parse_osc_response(b""), None);
}
fn run_matrix_case(
chunks: Vec<(&'static [u8], Duration)>,
deadline_budget: Duration,
) -> Vec<u8> {
let (rx, tx) = UnixStream::pair().expect("unix socket pair");
let writer = thread::spawn(move || {
let mut tx = tx;
for (chunk, gap) in chunks {
thread::sleep(gap);
let _ = tx.write_all(chunk);
}
});
let fd = rx.as_raw_fd();
let deadline = Instant::now() + deadline_budget;
let result = read_until_terminator_or_deadline(fd, deadline);
writer.join().unwrap();
result
}
const FULL_REPLY: &[u8] = b"\x1b]11;rgb:ffff/ffff/ffff\x1b\\";
#[test]
fn matrix_whole_packet_reads_complete_response() {
let result = run_matrix_case(
vec![(FULL_REPLY, Duration::ZERO)],
Duration::from_millis(200),
);
assert_eq!(result, FULL_REPLY);
}
#[test]
fn matrix_fragmented_50ms_gap_still_reads_complete_response() {
let result = run_matrix_case(
vec![
(&FULL_REPLY[..2], Duration::ZERO), (&FULL_REPLY[2..], Duration::from_millis(50)),
],
Duration::from_millis(200),
);
assert_eq!(
result, FULL_REPLY,
"a 50ms-fragmented reply must be read in full under a 200ms total deadline"
);
}
#[test]
fn matrix_fragmented_5ms_gap_reads_complete_response() {
let result = run_matrix_case(
vec![
(&FULL_REPLY[..2], Duration::ZERO),
(&FULL_REPLY[2..], Duration::from_millis(5)),
],
Duration::from_millis(200),
);
assert_eq!(result, FULL_REPLY);
}
#[test]
fn matrix_no_reply_returns_empty_at_deadline() {
let start = Instant::now();
let result = run_matrix_case(vec![], Duration::from_millis(80));
assert!(result.is_empty());
assert!(
start.elapsed() < Duration::from_millis(500),
"no-reply path must not hang past its deadline"
);
}
#[test]
fn matrix_reply_fragmented_beyond_total_deadline_is_truncated_not_leaked() {
let result = run_matrix_case(
vec![
(&FULL_REPLY[..2], Duration::ZERO),
(&FULL_REPLY[2..], Duration::from_millis(300)),
],
Duration::from_millis(50),
);
assert_eq!(result, &FULL_REPLY[..2]);
}
#[test]
fn tty_query_guard_fails_closed_on_a_non_tty_fd() {
let (a, _b) = UnixStream::pair().expect("unix socket pair");
let fd = a.as_raw_fd();
assert!(TtyQueryGuard::enter(fd).is_none());
}
#[test]
fn read_loop_reads_newline_free_response_through_pty() {
let mut master: RawFd = -1;
let mut slave: RawFd = -1;
let rc = unsafe {
libc::openpty(
&mut master,
&mut slave,
std::ptr::null_mut(),
std::ptr::null_mut(),
std::ptr::null_mut(),
)
};
assert_eq!(rc, 0, "openpty failed: {}", std::io::Error::last_os_error());
let mut before: libc::termios = unsafe { std::mem::zeroed() };
assert_eq!(unsafe { libc::tcgetattr(slave, &mut before) }, 0);
assert!(
before.c_lflag & libc::ICANON != 0,
"a fresh pty slave must start in canonical mode"
);
let reply = b"\x1b]11;rgb:ffff/ffff/ffff\x1b\\";
let got = {
let _guard = TtyQueryGuard::enter(slave).expect("guard on a real tty");
let mut during: libc::termios = unsafe { std::mem::zeroed() };
assert_eq!(unsafe { libc::tcgetattr(slave, &mut during) }, 0);
assert_eq!(during.c_lflag & libc::ICANON, 0, "ICANON must be cleared");
assert_eq!(during.c_lflag & libc::ECHO, 0, "ECHO must be cleared");
let n = unsafe { libc::write(master, reply.as_ptr().cast(), reply.len()) };
assert_eq!(n, reply.len() as isize, "short write to pty master");
let deadline = Instant::now() + Duration::from_millis(500);
read_until_terminator_or_deadline(slave, deadline)
};
assert_eq!(
got, reply,
"newline-free OSC reply must be read in full once ICANON is cleared"
);
let mut after: libc::termios = unsafe { std::mem::zeroed() };
assert_eq!(unsafe { libc::tcgetattr(slave, &mut after) }, 0);
assert!(
after.c_lflag & libc::ICANON != 0,
"ICANON must be restored after the guard drops"
);
unsafe {
libc::close(master);
libc::close(slave);
}
}
}