use anyhow::{anyhow, Context, Result};
use portable_pty::{native_pty_system, Child, CommandBuilder, MasterPty, PtySize};
use std::collections::hash_map::DefaultHasher;
use std::hash::{Hash, Hasher};
use std::io::{Read, Write};
use std::path::{Path, PathBuf};
use std::process::Command;
use std::sync::mpsc::{self, TryRecvError};
use std::sync::{Arc, Mutex as StdMutex};
use std::thread;
use std::time::{Duration, Instant};
use tokio::sync::Mutex;
use tracing::{debug, info, warn};
use crate::state::line_ring::LineRing;
use crate::state::live_terminal::{LiveTerminal, ScreenUpdate};
pub const DEFAULT_COLS: u16 = 200;
pub const DEFAULT_ROWS: u16 = 50;
const MAX_OUTPUT_SIZE: usize = 1_000_000;
const SCRATCH_MAX_BYTES: u64 = 50 * 1024 * 1024;
pub const RING_BUFFER_LINES: usize = 2_000;
const MAX_PARTIAL_LINE_BYTES: usize = 64 * 1024;
const WCGW_PROMPT_PATTERN: &str = "â—‰";
const WCGW_PROMPT_END: &str = "──➤";
fn attachable_command(restricted_mode: bool) -> (CommandBuilder, Option<String>, bool) {
let requested = crate::config::env_text("WINX_ATTACH_TERMINAL")
.or_else(|| crate::config::env_text("WINX_USE_SCREEN"))
.unwrap_or_default();
if requested == "tmux"
|| requested == "screen"
|| crate::config::parse_bool(&requested) == Some(true)
{
let session = format!("winx-{}-{}", std::process::id(), timestamp_millis());
if requested == "tmux" && command_available("tmux") {
let mut cmd = CommandBuilder::new("tmux");
cmd.args(["new-session", "-A", "-s", &session, "bash"]);
if restricted_mode {
cmd.arg("-r");
}
return (cmd, Some(format!("tmux attach -t {session}")), false);
}
if command_available("screen") {
ensure_screenrc();
cleanup_orphaned_screens();
let mut cmd = CommandBuilder::new("screen");
cmd.args(["-q", "-S", &session, "bash"]);
if restricted_mode {
cmd.arg("-r");
}
return (cmd, Some(format!("screen -x {session}")), false);
}
}
let shell = preferred_shell(restricted_mode);
let is_zsh = shell == "zsh";
let mut cmd = CommandBuilder::new(&shell);
if restricted_mode && !is_zsh {
cmd.arg("-r");
}
(cmd, None, is_zsh)
}
fn preferred_shell(restricted_mode: bool) -> String {
if !restricted_mode {
if let Ok(requested) = std::env::var("WINX_SHELL") {
if requested == "zsh" && command_available("zsh") {
return "zsh".to_string();
}
}
}
"bash".to_string()
}
fn command_available(command: &str) -> bool {
Command::new("sh")
.args(["-c", &format!("command -v {command}")])
.output()
.is_ok_and(|output| output.status.success())
}
fn ensure_screenrc() {
let Some(home) = home::home_dir() else {
return;
};
let screenrc = home.join(".screenrc");
if screenrc.exists() {
return;
}
let _ = std::fs::write(
&screenrc,
"defscrollback 10000\ntermcapinfo xterm* ti@:te@\nstartup_message off\n",
);
}
fn cleanup_orphaned_screens() {
let Ok(output) = Command::new("screen").arg("-ls").output() else {
return;
};
let listing = String::from_utf8_lossy(&output.stdout);
for line in listing.lines() {
let Some(session) = line.split_whitespace().next() else {
continue;
};
let Some((_, name)) = session.split_once('.') else {
continue;
};
if let Some(creator_pid) = winx_creator_pid(name) {
if !process_exists(creator_pid) {
let _ = Command::new("screen").args(["-S", session, "-X", "quit"]).output();
}
}
}
}
fn winx_creator_pid(name: &str) -> Option<u32> {
name.strip_prefix("winx-")?.split('-').next()?.parse::<u32>().ok()
}
#[cfg(unix)]
fn process_exists(pid: u32) -> bool {
crate::os::unix::process_exists(pid)
}
#[cfg(not(unix))]
fn process_exists(pid: u32) -> bool {
std::path::Path::new("/proc").join(pid.to_string()).exists()
}
fn timestamp_millis() -> u128 {
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map_or(0, |duration| duration.as_millis())
}
pub struct PtyShell {
master: Box<dyn MasterPty + Send>,
child: Box<dyn Child + Send + Sync>,
writer: Box<dyn Write + Send>,
output_rx: mpsc::Receiver<String>,
size: PtySize,
pub last_command: String,
pub output_buffer: String,
pub command_running: bool,
current_cwd: PathBuf,
command_started_at: Option<Instant>,
command_generation: u64,
incarnation: u64,
pub last_exit_code: Option<i32>,
max_output_size: usize,
pub output_truncated: bool,
ring: LineRing,
pub last_returned_hash: Option<u64>,
delivered_output: String,
pub attach_hint: Option<String>,
prompt_end_marker: String,
live: Arc<StdMutex<LiveTerminal>>,
scratch_workspace_root: Option<PathBuf>,
scratch_path: Option<PathBuf>,
scratch_bytes: u64,
}
impl std::fmt::Debug for PtyShell {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("PtyShell")
.field("size", &format!("{}x{}", self.size.cols, self.size.rows))
.field("last_command", &self.last_command)
.field("command_running", &self.command_running)
.field("current_cwd", &self.current_cwd)
.field("output_truncated", &self.output_truncated)
.field("output_buffer_len", &self.output_buffer.len())
.field("attach_hint", &self.attach_hint)
.finish_non_exhaustive()
}
}
#[cfg(unix)]
fn killable_group(pid: u32) -> Option<i32> {
crate::os::unix::owned_process_group(pid)
}
#[cfg(unix)]
fn signal_group(pgid: i32, signal: i32) {
let _ = crate::os::unix::signal_group(pgid, signal);
}
impl Drop for PtyShell {
fn drop(&mut self) {
#[cfg(unix)]
let pgid = if self.attach_hint.is_some() {
None
} else {
self.child.process_id().and_then(killable_group)
};
#[cfg(unix)]
if let Some(pgid) = pgid {
signal_group(pgid, libc::SIGTERM);
}
let _ = self.child.kill();
#[cfg(unix)]
if let Some(pgid) = pgid {
signal_group(pgid, libc::SIGKILL);
}
let _ = self.child.wait();
}
}
impl PtyShell {
pub fn new(initial_dir: &Path, restricted_mode: bool) -> Result<Self> {
Self::new_with_agent_paths(initial_dir, restricted_mode, None, None)
}
pub(crate) fn new_with_agent_paths(
initial_dir: &Path,
restricted_mode: bool,
workspace_root: Option<&Path>,
temporary_artifact_dir: Option<&Path>,
) -> Result<Self> {
info!(
"Creating new PTY shell (restricted: {}) in {}",
restricted_mode,
initial_dir.display()
);
let pty_system = native_pty_system();
let size =
PtySize { rows: DEFAULT_ROWS, cols: DEFAULT_COLS, pixel_width: 0, pixel_height: 0 };
let pair = pty_system.openpty(size).context("Failed to open PTY pair")?;
let (mut cmd, attach_hint, is_zsh) = attachable_command(restricted_mode);
cmd.env("TERM", "xterm-256color");
cmd.env("COLORTERM", "truecolor");
cmd.env("PAGER", "cat");
cmd.env("GIT_PAGER", "cat");
cmd.env("COLUMNS", DEFAULT_COLS.to_string());
cmd.env("ROWS", DEFAULT_ROWS.to_string());
if let Some(workspace_root) = workspace_root {
cmd.env("WINX_WORKSPACE_ROOT", workspace_root);
}
if let Some(temporary_artifact_dir) = temporary_artifact_dir {
cmd.env("WINX_TEMP_DIR", temporary_artifact_dir);
}
let nonce = format!("{:016x}", rand::random::<u64>());
cmd.env(
"PROMPT_COMMAND",
format!(r#"__winx_ec=$?; printf "◉ %s──➤{nonce}:%s " "$PWD" "$__winx_ec""#),
);
cmd.cwd(initial_dir);
let child = pair.slave.spawn_command(cmd).context("Failed to spawn bash in PTY")?;
let mut reader = pair.master.try_clone_reader().context("Failed to clone PTY reader")?;
let writer = pair.master.take_writer().context("Failed to take PTY writer")?;
let (output_tx, output_rx) = mpsc::sync_channel::<String>(1024);
let live = Arc::new(StdMutex::new(LiveTerminal::new(DEFAULT_ROWS, DEFAULT_COLS)));
let live_reader = Arc::clone(&live);
thread::spawn(move || {
let mut buf = [0u8; 4096];
let mut carry: Vec<u8> = Vec::new();
loop {
match reader.read(&mut buf) {
Ok(0) => {
if !carry.is_empty() {
let _ = output_tx.send(String::from_utf8_lossy(&carry).into_owned());
}
break;
}
Ok(n) => {
{
let mut emu = live_reader
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
emu.feed(&buf[..n]);
}
carry.extend_from_slice(&buf[..n]);
let (chunk, rest) = decode_keep_incomplete(&carry);
carry = rest;
if !chunk.is_empty() && output_tx.send(chunk).is_err() {
break;
}
}
Err(e) => {
debug!("PTY reader thread error: {}", e);
break;
}
}
}
debug!("PTY reader thread exiting");
});
let mut shell = Self {
master: pair.master,
child,
writer,
output_rx,
size,
last_command: String::new(),
output_buffer: String::new(),
command_running: false,
current_cwd: initial_dir.to_path_buf(),
command_started_at: None,
command_generation: 0,
incarnation: rand::random::<u64>().max(1),
last_exit_code: None,
max_output_size: MAX_OUTPUT_SIZE,
output_truncated: false,
ring: LineRing::new(RING_BUFFER_LINES, MAX_PARTIAL_LINE_BYTES),
last_returned_hash: None,
delivered_output: String::new(),
attach_hint,
prompt_end_marker: format!("{WCGW_PROMPT_END}{nonce}"),
live,
scratch_workspace_root: None,
scratch_path: None,
scratch_bytes: 0,
};
shell.initialize_prompt(is_zsh, &nonce)?;
debug!("PTY shell created successfully");
Ok(shell)
}
fn initialize_prompt(&mut self, is_zsh: bool, nonce: &str) -> Result<()> {
let prompt_statement = if is_zsh {
format!(
r#"export GIT_PAGER=cat PAGER=cat; precmd_functions=(); preexec_functions=(); PROMPT=''; RPROMPT=''; precmd() {{ local __winx_ec=$?; printf "◉ %s──➤{nonce}:%s " "$PWD" "$__winx_ec" }}"#
)
} else {
format!(
r#"export GIT_PAGER=cat PAGER=cat PROMPT_COMMAND='__winx_ec=$?; printf "◉ %s──➤{nonce}:%s " "$PWD" "$__winx_ec"'; PS1=''"#
)
};
self.write_command(&prompt_statement)?;
std::thread::sleep(Duration::from_millis(100));
let _ = self.drain_output();
Ok(())
}
fn write_command(&mut self, command: &str) -> Result<()> {
let cmd_with_newline = format!("{command}\r");
self.writer.write_all(cmd_with_newline.as_bytes()).context("Failed to write to PTY")?;
self.writer.flush().context("Failed to flush PTY")?;
Ok(())
}
fn drain_output(&mut self) -> String {
let mut output = String::new();
let deadline = Instant::now() + Duration::from_millis(200);
while Instant::now() < deadline {
match self.output_rx.try_recv() {
Ok(chunk) => {
output.push_str(&chunk);
if output.len() > self.max_output_size {
self.output_truncated = true;
break;
}
}
Err(TryRecvError::Empty) => {
thread::sleep(Duration::from_millis(10));
}
Err(TryRecvError::Disconnected) => {
break;
}
}
}
output
}
pub fn send_command(&mut self, command: &str) -> Result<()> {
debug!(bytes = command.len(), "PTY sending command");
self.output_buffer.clear();
self.output_truncated = false;
self.reset_scratch();
self.last_command = command.to_string();
self.command_running = true;
self.mark_command_started();
self.last_exit_code = None;
self.last_returned_hash = None;
self.delivered_output.clear();
self.write_command(command)?;
Ok(())
}
pub fn mark_command_started(&mut self) {
self.command_generation = self.command_generation.saturating_add(1);
self.command_started_at = Some(Instant::now());
}
pub fn command_generation(&self) -> u64 {
self.command_generation
}
pub fn incarnation(&self) -> u64 {
self.incarnation
}
pub fn command_elapsed(&self) -> Option<Duration> {
self.command_started_at.map(|started_at| started_at.elapsed())
}
pub fn current_cwd(&self) -> &Path {
&self.current_cwd
}
pub fn delivered_output(&self) -> String {
self.delivered_output.clone()
}
pub fn mark_output_delivered(&mut self, output: &str) {
self.delivered_output.clear();
self.delivered_output.push_str(output);
}
pub fn collect_scrollback(&self, lines: usize) -> String {
let rendered =
crate::state::terminal::render_terminal_output(&self.ring.raw(lines)).join("\n");
match self.ring.scrollback_notice(lines) {
Some(notice) => format!("{notice}\n{rendered}"),
None => rendered,
}
}
pub fn fingerprint(text: &str) -> u64 {
let mut hasher = DefaultHasher::new();
text.hash(&mut hasher);
hasher.finish()
}
pub fn read_output(&mut self, timeout_secs: f32) -> Result<(String, bool)> {
let timeout = Duration::from_secs_f32(timeout_secs.clamp(0.1, 60.0));
let start = Instant::now();
let mut complete = false;
let mut no_data_count = 0;
let mut prompt_detected_at: Option<Instant> = None;
while start.elapsed() < timeout {
match self.output_rx.try_recv() {
Ok(chunk) => {
self.output_buffer.push_str(&chunk);
self.ring.push_chunk(&chunk);
no_data_count = 0;
if prompt_detected_at.is_none()
&& (Self::check_prompt_complete(&chunk, &self.prompt_end_marker)
|| Self::check_prompt_complete(
&self.output_buffer,
&self.prompt_end_marker,
))
{
prompt_detected_at = Some(Instant::now());
debug!("Prompt detected, draining remaining output...");
}
self.truncate_output_buffer_with_offload();
}
Err(TryRecvError::Empty) => {
thread::sleep(Duration::from_millis(10));
no_data_count += 1;
if let Some(detected_time) = prompt_detected_at {
if detected_time.elapsed() > Duration::from_millis(100) {
complete = true;
debug!("Command completed - prompt detected and drained");
break;
}
} else if no_data_count > 10
&& Self::check_prompt_complete(&self.output_buffer, &self.prompt_end_marker)
{
prompt_detected_at = Some(Instant::now());
debug!("Prompt detected after wait, draining...");
}
}
Err(TryRecvError::Disconnected) => {
warn!("PTY reader disconnected");
complete = true;
break;
}
}
}
if complete || prompt_detected_at.is_some() {
self.command_running = false;
self.last_exit_code =
Self::parse_prompt_exit_code(&self.output_buffer, &self.prompt_end_marker);
self.update_current_cwd_from_prompt();
complete = true;
}
Ok((self.output_buffer.clone(), complete))
}
pub fn poll_output_nonblocking(&mut self) -> bool {
let mut prompt_seen = false;
loop {
match self.output_rx.try_recv() {
Ok(chunk) => {
let chunk_has_prompt =
Self::check_prompt_complete(&chunk, &self.prompt_end_marker);
self.output_buffer.push_str(&chunk);
self.ring.push_chunk(&chunk);
if chunk_has_prompt
|| Self::check_prompt_complete(&self.output_buffer, &self.prompt_end_marker)
{
prompt_seen = true;
}
self.truncate_output_buffer_with_offload();
}
Err(TryRecvError::Empty) => break,
Err(TryRecvError::Disconnected) => {
self.command_running = false;
return true;
}
}
}
if prompt_seen {
self.command_running = false;
self.last_exit_code =
Self::parse_prompt_exit_code(&self.output_buffer, &self.prompt_end_marker);
self.update_current_cwd_from_prompt();
}
!self.command_running
}
fn update_current_cwd_from_prompt(&mut self) {
let Some(last) =
self.output_buffer.lines().rev().find(|line| line.contains(&self.prompt_end_marker))
else {
return;
};
let clean = crate::state::terminal::strip_ansi_codes(last);
let Some((before_marker, _)) = clean.rsplit_once(&self.prompt_end_marker) else {
return;
};
let Some((_, cwd)) = before_marker.rsplit_once("â—‰ ") else {
return;
};
let cwd = cwd.trim();
if !cwd.is_empty() {
self.current_cwd = PathBuf::from(cwd);
}
}
pub fn output_snapshot(&self) -> String {
self.output_buffer.clone()
}
fn check_prompt_complete(text: &str, prompt_end: &str) -> bool {
Self::parse_prompt_exit_code(text, prompt_end).is_some()
}
fn parse_prompt_exit_code(text: &str, prompt_end: &str) -> Option<i32> {
let last = text.lines().rev().find(|line| !line.trim().is_empty())?;
let clean = crate::state::terminal::strip_ansi_codes(last);
let clean = clean.trim_end();
if !clean.contains(WCGW_PROMPT_PATTERN) {
return None;
}
let after = clean.rsplit_once(prompt_end)?.1;
after.strip_prefix(':')?.trim().parse::<i32>().ok()
}
pub fn set_scratch_root(&mut self, root: &Path) {
self.scratch_workspace_root = Some(root.to_path_buf());
}
pub fn scratch_path(&self) -> Option<&Path> {
self.scratch_path.as_deref()
}
pub fn reset_scratch(&mut self) {
self.scratch_path = None;
self.scratch_bytes = 0;
}
fn offload_dropped_head(&mut self, head: &str) {
let Some(root) = self.scratch_workspace_root.clone() else {
return;
};
let remaining = SCRATCH_MAX_BYTES.saturating_sub(self.scratch_bytes);
if remaining == 0 {
return;
}
let head = match usize::try_from(remaining) {
Ok(rem) if rem < head.len() => &head[..crate::utils::floor_char_boundary(head, rem)],
_ => head,
};
if self.scratch_path.is_none() {
self.scratch_path = crate::utils::scratch_file::new_scratch_path(&root);
}
let Some(path) = self.scratch_path.clone() else {
return;
};
match crate::utils::scratch_file::append_scratch(&path, head.as_bytes()) {
Ok(0) if self.scratch_bytes == 0 => self.scratch_path = None,
Ok(written) => {
self.scratch_bytes = self.scratch_bytes.saturating_add(written as u64);
}
Err(e) => debug!("scratch: append to {} failed: {e}", path.display()),
}
}
fn truncate_output_buffer_with_offload(&mut self) {
if self.output_buffer.len() <= self.max_output_size {
return;
}
self.output_truncated = true;
let keep_size = self.max_output_size / 2;
let cut = crate::utils::floor_char_boundary(
&self.output_buffer,
self.output_buffer.len() - keep_size,
);
let head = self.output_buffer[..cut].to_owned();
self.offload_dropped_head(&head);
self.output_buffer.replace_range(..cut, "\n(...output truncated...)\n");
}
pub fn send_interrupt(&mut self) -> Result<()> {
debug!("PTY sending Ctrl+C");
self.writer
.write_all(&[0x03]) .context("Failed to send Ctrl+C")?;
self.writer.flush()?;
Ok(())
}
pub fn send_eof(&mut self) -> Result<()> {
debug!("PTY sending Ctrl+D");
self.writer
.write_all(&[0x04]) .context("Failed to send Ctrl+D")?;
self.writer.flush()?;
Ok(())
}
pub fn send_suspend(&mut self) -> Result<()> {
debug!("PTY sending Ctrl+Z");
self.writer
.write_all(&[0x1A]) .context("Failed to send Ctrl+Z")?;
self.writer.flush()?;
Ok(())
}
pub fn send_text(&mut self, text: &str) -> Result<()> {
debug!(bytes = text.len(), "PTY sending text");
self.send_bytes(text.as_bytes()).context("Failed to send text")?;
Ok(())
}
pub fn send_bytes(&mut self, bytes: &[u8]) -> Result<()> {
self.writer.write_all(bytes).context("Failed to send bytes")?;
self.writer.flush()?;
Ok(())
}
pub fn send_special_key(&mut self, key: &str) -> Result<()> {
let bytes: &[u8] = match key {
"Enter" => b"\r",
"Tab" => b"\t",
"Backspace" => b"\x7F",
"Escape" => b"\x1B",
"Up" | "KeyUp" => b"\x1B[A",
"Down" | "KeyDown" => b"\x1B[B",
"Right" | "KeyRight" => b"\x1B[C",
"Left" | "KeyLeft" => b"\x1B[D",
"Home" => b"\x1B[H",
"End" => b"\x1B[F",
"PageUp" => b"\x1B[5~",
"PageDown" => b"\x1B[6~",
"Delete" => b"\x1B[3~",
"Insert" => b"\x1B[2~",
"CtrlC" | "Ctrl-C" => b"\x03",
"CtrlD" | "Ctrl-D" => b"\x04",
"CtrlZ" | "Ctrl-Z" => b"\x1A",
"CtrlL" | "Ctrl-L" => b"\x0C",
_ => return Err(anyhow!("Unknown special key: {key}")),
};
debug!("PTY sending special key: {} ({:?})", key, bytes);
self.send_bytes(bytes)?;
Ok(())
}
pub fn live_snapshot(&self, max_lines: usize) -> Vec<String> {
self.live.lock().unwrap_or_else(std::sync::PoisonError::into_inner).snapshot(max_lines)
}
pub fn live_in_alt_screen(&self) -> bool {
self.live.lock().unwrap_or_else(std::sync::PoisonError::into_inner).in_alt_screen()
}
pub fn live_cursor_position(&self) -> (u16, u16) {
self.live.lock().unwrap_or_else(std::sync::PoisonError::into_inner).cursor_position()
}
pub fn live_snapshot_diff(&self, max_lines: usize, threshold: usize) -> ScreenUpdate {
self.live
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.snapshot_diff(max_lines, threshold)
}
pub fn resize(&mut self, cols: u16, rows: u16) -> Result<()> {
debug!("PTY resizing to {}x{}", cols, rows);
let new_size = PtySize { rows, cols, pixel_width: 0, pixel_height: 0 };
self.master.resize(new_size).context("Failed to resize PTY")?;
self.live.lock().unwrap_or_else(std::sync::PoisonError::into_inner).resize(rows, cols);
self.size = new_size;
Ok(())
}
pub fn get_size(&self) -> (u16, u16) {
(self.size.cols, self.size.rows)
}
pub fn process_id(&self) -> Option<u32> {
self.child.process_id()
}
pub fn is_alive(&mut self) -> bool {
self.child.try_wait().is_ok_and(|status| status.is_none())
}
}
pub type SharedPtyShell = Arc<Mutex<Option<PtyShell>>>;
pub fn create_shared_pty(initial_dir: &Path, restricted_mode: bool) -> Result<SharedPtyShell> {
let shell = PtyShell::new(initial_dir, restricted_mode)?;
Ok(Arc::new(Mutex::new(Some(shell))))
}
fn decode_keep_incomplete(bytes: &[u8]) -> (String, Vec<u8>) {
let mut decoded = String::new();
let mut rest = bytes;
loop {
match std::str::from_utf8(rest) {
Ok(text) => {
decoded.push_str(text);
return (decoded, Vec::new());
}
Err(error) => {
let valid = error.valid_up_to();
decoded.push_str(&String::from_utf8_lossy(&rest[..valid]));
match error.error_len() {
None => return (decoded, rest[valid..].to_vec()),
Some(bad) => {
decoded.push('\u{FFFD}');
rest = &rest[valid + bad..];
}
}
}
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use tempfile::TempDir;
#[test]
fn decode_passes_clean_utf8_through() {
let (text, carry) = decode_keep_incomplete("hello ✓ café".as_bytes());
assert_eq!(text, "hello ✓ café");
assert!(carry.is_empty());
}
#[test]
fn decode_carries_incomplete_trailing_char_across_reads() {
let (text1, carry1) = decode_keep_incomplete(&[b'a', 0xC3]);
assert_eq!(text1, "a");
assert_eq!(carry1, vec![0xC3]);
let mut next = carry1;
next.push(0xA9); let (text2, carry2) = decode_keep_incomplete(&next);
assert_eq!(text2, "é");
assert!(carry2.is_empty());
}
#[test]
fn decode_replaces_genuinely_invalid_bytes() {
let (text, carry) = decode_keep_incomplete(&[b'x', 0xFF, b'y']);
assert_eq!(text, "x\u{FFFD}y");
assert!(carry.is_empty());
}
#[test]
fn decode_handles_large_invalid_burst_without_overflow() {
let big = vec![0xFFu8; 200_000];
let (text, carry) = decode_keep_incomplete(&big);
assert!(carry.is_empty());
assert_eq!(text.chars().filter(|&c| c == '\u{FFFD}').count(), 200_000);
}
#[test]
fn screen_session_name_parser_is_strict() {
assert_eq!(super::winx_creator_pid("winx-123-456"), Some(123));
assert_eq!(super::winx_creator_pid("winx-nope-456"), None);
assert_eq!(super::winx_creator_pid("other-123-456"), None);
}
#[cfg(unix)]
#[test]
fn process_probe_recognizes_current_process_and_rejects_reserved_pids() {
assert!(super::process_exists(std::process::id()));
assert!(!super::process_exists(0));
assert!(!super::process_exists(1));
}
#[test]
fn prompt_detection_is_suffix_anchored() {
let end = "──➤deadbeefcafe0001";
assert!(PtyShell::check_prompt_complete("out\nmore\n◉ /home/x──➤deadbeefcafe0001:0 ", end));
assert!(PtyShell::check_prompt_complete("◉ /home/x──➤deadbeefcafe0001:0 \u{1b}[K", end));
assert!(!PtyShell::check_prompt_complete(
"menu: ◉ start ──➤deadbeefcafe0001 stop\nstill running",
end
));
assert!(!PtyShell::check_prompt_complete("◉ /home/x──➤deadbeefcafe0001:0 ls -la", end));
assert!(!PtyShell::check_prompt_complete("just some output\n", end));
assert!(!PtyShell::check_prompt_complete("◉ /fake──➤ ", end));
assert!(!PtyShell::check_prompt_complete("◉ /fake──➤wrongnonce:0 ", end));
assert!(!PtyShell::check_prompt_complete("◉ /home/x──➤deadbeefcafe0001 ", end));
}
#[test]
fn parses_exit_code_from_prompt_marker() {
let end = "──➤cafe1234";
assert_eq!(PtyShell::parse_prompt_exit_code("ok\n◉ /x──➤cafe1234:0 ", end), Some(0));
assert_eq!(PtyShell::parse_prompt_exit_code("oops\n◉ /x──➤cafe1234:1 ", end), Some(1));
assert_eq!(PtyShell::parse_prompt_exit_code("◉ /x──➤cafe1234:42 ", end), Some(42));
assert_eq!(PtyShell::parse_prompt_exit_code("building...", end), None);
assert_eq!(PtyShell::parse_prompt_exit_code("◉ /x──➤wrong:0 ", end), None);
}
use proptest::prelude::*;
proptest! {
#[test]
fn parse_prompt_exit_code_never_panics(
text in prop::collection::vec(any::<char>(), 0..256),
nonce in "[0-9a-f]{0,16}",
) {
let s: String = text.into_iter().collect();
let end = format!("{WCGW_PROMPT_END}{nonce}");
let _ = PtyShell::parse_prompt_exit_code(&s, &end);
}
#[test]
fn parse_prompt_exit_code_round_trips(code in any::<i32>(), nonce in "[0-9a-f]{1,16}") {
let end = format!("{WCGW_PROMPT_END}{nonce}");
let line = format!("output line\n{WCGW_PROMPT_PATTERN} /home/x{end}:{code} ");
prop_assert_eq!(PtyShell::parse_prompt_exit_code(&line, &end), Some(code));
}
}
#[test]
fn test_pty_shell_creation() -> Result<()> {
let temp_dir = TempDir::new()?;
let result = PtyShell::new(temp_dir.path(), false);
assert!(result.is_ok(), "Failed to create PTY shell: {:?}", result.err());
Ok(())
}
#[test]
fn test_pty_shell_echo() -> Result<()> {
let temp_dir = TempDir::new()?;
let mut shell = PtyShell::new(temp_dir.path(), false)?;
shell.send_command("echo 'hello pty'")?;
let (output, _complete) = shell.read_output(2.0)?;
assert!(output.contains("hello pty"), "Output should contain 'hello pty': {output}");
Ok(())
}
#[test]
fn agent_paths_are_exported_to_the_pty() -> Result<()> {
let workspace = TempDir::new()?;
let temporary_artifact_dir = workspace.path().join(".winx/tmp/session-test");
let mut shell = PtyShell::new_with_agent_paths(
workspace.path(),
false,
Some(workspace.path()),
Some(&temporary_artifact_dir),
)?;
shell.send_command("printf '%s\\n%s\\n' \"$WINX_WORKSPACE_ROOT\" \"$WINX_TEMP_DIR\"")?;
let (output, _complete) = shell.read_output(2.0)?;
assert!(output.contains(&workspace.path().to_string_lossy().to_string()), "{output}");
assert!(output.contains(&temporary_artifact_dir.to_string_lossy().to_string()), "{output}");
Ok(())
}
#[test]
fn completed_nonblocking_poll_is_level_triggered() -> Result<()> {
let temp_dir = TempDir::new()?;
let mut shell = PtyShell::new(temp_dir.path(), false)?;
shell.send_command("true")?;
let deadline = Instant::now() + Duration::from_secs(2);
while Instant::now() < deadline && !shell.poll_output_nonblocking() {
std::thread::sleep(Duration::from_millis(10));
}
assert!(!shell.command_running, "shell did not complete before the deadline");
assert!(
shell.poll_output_nonblocking(),
"a later consumer must still observe the completed state"
);
Ok(())
}
#[test]
fn test_pty_shell_pwd() -> Result<()> {
let temp_dir = TempDir::new()?;
let mut shell = PtyShell::new(temp_dir.path(), false)?;
shell.send_command("pwd && echo 'pwd_done'")?;
let (output, _complete) = shell.read_output(2.0)?;
assert!(output.contains("pwd_done"), "Output should contain 'pwd_done': {output}");
Ok(())
}
#[test]
fn test_pty_resize() -> Result<()> {
let temp_dir = TempDir::new()?;
let mut shell = PtyShell::new(temp_dir.path(), false)?;
let result = shell.resize(120, 40);
assert!(result.is_ok());
let (cols, rows) = shell.get_size();
assert_eq!(cols, 120);
assert_eq!(rows, 40);
Ok(())
}
#[cfg(unix)]
fn pid_alive(pid: i32) -> bool {
u32::try_from(pid).is_ok_and(crate::os::unix::process_exists)
}
#[cfg(unix)]
#[test]
fn dropping_shell_kills_orphaned_background_child() -> Result<()> {
let dir = TempDir::new()?;
let pidfile = dir.path().join("child.pid");
let mut shell = PtyShell::new(dir.path(), false)?;
shell
.send_command(&format!("sleep 300 & echo $! > {} ; echo spawned", pidfile.display()))?;
let mut child_pid = None;
for _ in 0..150 {
let _ = shell.read_output(0.1);
if let Some(pid) =
std::fs::read_to_string(&pidfile).ok().and_then(|c| c.trim().parse::<i32>().ok())
{
child_pid = Some(pid);
break;
}
thread::sleep(Duration::from_millis(20));
}
let child_pid =
child_pid.ok_or_else(|| anyhow!("background child pid never materialized"))?;
assert!(pid_alive(child_pid), "background child should be running before drop");
drop(shell);
let mut gone = false;
for _ in 0..150 {
if !pid_alive(child_pid) {
gone = true;
break;
}
thread::sleep(Duration::from_millis(20));
}
assert!(gone, "background child {child_pid} survived shell teardown (orphan leak)");
Ok(())
}
}