rss attach: send commands to service stdin

- rss send <name> <message>: one-line command to service stdin
- rss attach <name>: interactive stdin forwarding with log tailing
- rss attach --full <name>: same but Ctrl+C sends SIGINT
- safe mode: Ctrl+C just detaches, doesn't kill the process
- Ctrl+P then D to detach
- daemon stores ChildStdin in Proc, handles streaming stdin IPC

src/cli.rs

@@ -50,6 +50,17 @@ pub enum Commands {
     List,
     /// Start all enabled services
     Daemon,
+    /// Send one line to a service's stdin
+    Send {
+        name: String,
+        message: Vec<String>,
+    },
+    /// Attach to a service's stdin interactively
+    Attach {
+        name: String,
+        #[arg(long, help = "Send SIGINT to service on Ctrl+C before disconnecting")]
+        full: bool,
+    },
     /// Internal: run supervisor for a service
     #[command(hide = true)]
     Supervise {

src/daemon.rs

@@ -7,6 +7,7 @@ use std::os::unix::process::CommandExt;
 use std::process::{Child, Command, Stdio};
 use std::time::{Duration, SystemTime};
 use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::{Arc, Mutex};
 
 use crate::config::{Config, RestartPolicy, ServiceConfig};
 use crate::state;
@@ -22,6 +23,7 @@ struct Proc {
     pid: u32,
     cfg: ServiceConfig,
     restart_count: u32,
+    stdin: Option<Arc<Mutex<std::process::ChildStdin>>>,
 }
 
 pub fn run(config_path: String) {
@@ -174,7 +176,7 @@ fn spawn(name: &str, cfg: &ServiceConfig) -> Option<Proc> {
             .envs(&cfg.env)
             .stdout(Stdio::piped())
             .stderr(Stdio::piped())
-            .stdin(Stdio::null());
+            .stdin(Stdio::piped());
         // SAFETY: pre_exec runs in the child after fork; minimal ops only
         unsafe {
             sh.pre_exec(|| {
@@ -196,6 +198,7 @@ fn spawn(name: &str, cfg: &ServiceConfig) -> Option<Proc> {
     };
 
     let pid = child.id();
+    let child_stdin = child.stdin.take().map(|s| Arc::new(Mutex::new(s)));
     let log_out = log_path.clone();
     let log_err = log_path.clone();
 
@@ -225,6 +228,7 @@ fn spawn(name: &str, cfg: &ServiceConfig) -> Option<Proc> {
         pid,
         cfg: cfg.clone(),
         restart_count: 0,
+        stdin: child_stdin,
     })
 }
 
@@ -360,9 +364,36 @@ fn handle_client(
     if reader.read_line(&mut line).is_err() {
         return;
     }
-    let line = line.trim();
+    let line = line.trim().to_string();
 
-    let response = match line {
+    if line.starts_with("stdin ") {
+        drop(reader);
+        handle_stdin(&line, stream, procs);
+        return;
+    }
+    if line.starts_with("signal ") {
+        let rest = &line[7..];
+        let response = if let Some(space) = rest.find(' ') {
+            let name = &rest[..space];
+            let sig_name = &rest[space + 1..];
+            if let Some(p) = procs.get(name) {
+                let sig = crate::supervisor::parse_signal(sig_name);
+                match nix::sys::signal::kill(nix::unistd::Pid::from_raw(p.pid as i32), sig) {
+                    Ok(_) => "ok".to_string(),
+                    Err(e) => format!("error: signal failed: {}", e),
+                }
+            } else {
+                format!("error: '{}' not running", name)
+            }
+        } else {
+            "error: usage: signal <name> <signal>".to_string()
+        };
+        let _ = writeln!(stream, "{}", response);
+        let _ = stream.flush();
+        return;
+    }
+
+    let response = match line.as_str() {
         l if l.starts_with("start ") => {
             let names: Vec<&str> = l[6..].split_whitespace().collect();
             let mut out = Vec::new();
@@ -499,6 +530,79 @@ fn handle_client(
     let _ = stream.flush();
 }
 
+fn handle_stdin(line: &str, mut stream: &UnixStream, procs: &mut HashMap<String, Proc>) {
+    let rest = &line[6..];
+    if let Some(space) = rest.find(' ') {
+        let name = &rest[..space];
+        let msg = &rest[space + 1..];
+        match procs.get(name) {
+            Some(p) => match &p.stdin {
+                Some(stdin) => {
+                    let mut stdin = stdin.lock().unwrap();
+                    let _ = writeln!(stdin, "{}", msg);
+                    let _ = stdin.flush();
+                    let _ = writeln!(stream, "ok");
+                }
+                None => {
+                    let _ = writeln!(stream, "error: '{}' has no stdin", name);
+                }
+            },
+            None => {
+                let _ = writeln!(stream, "error: '{}' not running", name);
+            }
+        }
+        let _ = stream.flush();
+    } else {
+        let name = rest;
+        let has_stdin = procs.get(name).and_then(|p| p.stdin.as_ref()).is_some();
+        if !procs.contains_key(name) {
+            let _ = writeln!(stream, "error: '{}' not running", name);
+            let _ = stream.flush();
+            return;
+        }
+        if !has_stdin {
+            let _ = writeln!(stream, "error: '{}' has no stdin", name);
+            let _ = stream.flush();
+            return;
+        }
+        let stdin_arc = procs.get(name).and_then(|p| p.stdin.clone()).unwrap();
+        match stream.try_clone() {
+            Ok(clone) => {
+                let _ = writeln!(stream, "ok");
+                let _ = stream.flush();
+                std::thread::spawn(move || {
+                    let mut reader = BufReader::new(clone);
+                    let mut line = String::new();
+                    loop {
+                        line.clear();
+                        match reader.read_line(&mut line) {
+                            Ok(0) => break,
+                            Ok(_) => {
+                                let t = line.trim();
+                                if !t.is_empty() {
+                                    let mut s = match stdin_arc.lock() {
+                                        Ok(s) => s,
+                                        Err(_) => break,
+                                    };
+                                    if writeln!(s, "{}", t).is_err() {
+                                        break;
+                                    }
+                                    let _ = s.flush();
+                                }
+                            }
+                            Err(_) => break,
+                        }
+                    }
+                });
+            }
+            Err(e) => {
+                let _ = writeln!(stream, "error: {}", e);
+                let _ = stream.flush();
+            }
+        }
+    }
+}
+
 fn append_log(path: &str, msg: &str) {
     let ts = chrono::Local::now().format("%Y-%m-%d %H:%M:%S");
     if let Ok(mut f) = std::fs::OpenOptions::new()

src/ipc.rs

@@ -21,3 +21,27 @@ pub fn send_cmd(cmd: &str) -> Result<Vec<String>, String> {
     }
     Ok(lines)
 }
+
+pub fn send_cmd_get_stream(cmd: &str) -> Result<UnixStream, String> {
+    let path = crate::state::socket_path();
+    let mut stream =
+        UnixStream::connect(&path).map_err(|e| format!("daemon not running: {}", e))?;
+
+    writeln!(stream, "{}", cmd).map_err(|e| format!("write: {}", e))?;
+    stream.flush().map_err(|e| format!("flush: {}", e))?;
+
+    let mut reader = BufReader::new(&stream);
+    let mut line = String::new();
+    reader
+        .read_line(&mut line)
+        .map_err(|e| format!("read: {}", e))?;
+    let trimmed = line.trim();
+    if let Some(msg) = trimmed.strip_prefix("error:") {
+        return Err(msg.trim().to_string());
+    }
+    if trimmed != "ok" {
+        return Err(format!("unexpected response: {}", trimmed));
+    }
+    drop(reader);
+    Ok(stream)
+}

src/main.rs

@@ -8,6 +8,8 @@ mod supervisor;
 
 use clap::Parser;
 use cli::{Cli, Commands};
+use std::io::Write;
+use std::sync::atomic::{AtomicBool, Ordering};
 
 
 fn main() {
@@ -26,6 +28,8 @@ fn main() {
         Commands::Logs { ref name, lines, follow } => cmd_logs(name, lines, follow),
         Commands::Enable { ref names, now } => cmd_enable(names, now),
         Commands::Disable { ref names } => cmd_disable(names),
+        Commands::Send { ref name, ref message } => cmd_send(name, message),
+        Commands::Attach { ref name, full } => cmd_attach(name, full),
         Commands::Supervise { .. } => {
             eprintln!("error: supervise is internal, use daemon");
             std::process::exit(1);
@@ -180,3 +184,80 @@ fn cmd_disable(names: &[String]) {
         }
     }
 }
+
+fn cmd_send(name: &str, message: &[String]) {
+    let msg = message.join(" ");
+    let cmd = format!("stdin {} {}", name, msg);
+    match ipc::send_cmd(&cmd) {
+        Ok(_) => println!("ok"),
+        Err(e) => {
+            eprintln!("error: {}", e);
+            std::process::exit(1);
+        }
+    }
+}
+
+static ATTACH_INTERRUPT: AtomicBool = AtomicBool::new(false);
+
+extern "C" fn handle_attach_int(_: i32) {
+    ATTACH_INTERRUPT.store(true, Ordering::Relaxed);
+}
+
+fn cmd_attach(name: &str, full: bool) {
+    let sig_action = nix::sys::signal::SigAction::new(
+        nix::sys::signal::SigHandler::Handler(handle_attach_int),
+        nix::sys::signal::SaFlags::empty(),
+        nix::sys::signal::SigSet::empty(),
+    );
+    unsafe {
+        let _ = nix::sys::signal::sigaction(nix::sys::signal::Signal::SIGINT, &sig_action);
+    }
+
+    let stream = match ipc::send_cmd_get_stream(&format!("stdin {}", name)) {
+        Ok(s) => s,
+        Err(e) => {
+            eprintln!("error: {}", e);
+            std::process::exit(1);
+        }
+    };
+
+    // Tail the log file in a background thread
+    let log_path = crate::state::log_path(name);
+    if log_path.exists() {
+        let log_path_str = log_path.to_string_lossy().to_string();
+        std::thread::spawn(move || {
+            let _ = crate::log::show_logs(&log_path_str, Some(0), true);
+        });
+    }
+
+    // Main loop: read stdin line by line, send to daemon
+    let stdin = std::io::stdin();
+    loop {
+        if ATTACH_INTERRUPT.load(Ordering::Relaxed) {
+            if full {
+                let _ = ipc::send_cmd(&format!("signal {} INT", name));
+            }
+            break;
+        }
+        let mut line = String::new();
+        match stdin.read_line(&mut line) {
+            Ok(0) => break,
+            Ok(_) => {
+                let trimmed = line.trim_end_matches('\n').trim_end_matches('\r');
+                let bytes = trimmed.as_bytes();
+                if bytes.len() >= 2 && bytes[0] == 0x10 && (bytes[1] == b'd' || bytes[1] == b'D')
+                {
+                    break;
+                }
+                if !trimmed.is_empty() {
+                    let _ = writeln!(&stream, "{}", trimmed);
+                    let _ = (&stream).flush();
+                }
+            }
+            Err(e) if e.kind() == std::io::ErrorKind::Interrupted => {
+                continue;
+            }
+            Err(_) => break,
+        }
+    }
+}