(tried to) prioritize terminal input

refresh interactive sessions

Cargo.lock

@@ -3700,6 +3700,7 @@ dependencies = [
  "futures",
  "lda-ipjs",
  "macaddr",
+ "nix 0.31.2",
  "pty-process",
  "serde",
  "serde_json",

docs/adr/0002-remote-terminal-sessions.md

@@ -54,12 +54,13 @@ The terminal control vocabulary is deliberately small:
 
 ```text
 operator -> agent: resize, close
+control plane -> agent: refresh
 agent -> operator: ready, exited, error
 ```
 
 Initial terminal dimensions are part of session attachment. Resize carries rows and columns; pixel dimensions may be added later if a demonstrated program requires them.
 
-Terminal behavior such as Ctrl-C, Ctrl-Z, Ctrl-D, cursor keys, mouse reporting, bracketed paste, colors, and terminal-title changes remains in the byte stream. The control plane does not parse ANSI escape sequences. The browser terminal emulator renders output and owns connected-session scrollback.
+Terminal behavior such as Ctrl-C, Ctrl-Z, Ctrl-D, cursor keys, mouse reporting, bracketed paste, colors, and terminal-title changes remains in the byte stream. The control plane does not parse ANSI escape sequences. The browser terminal emulator renders output and owns connected-session scrollback. Agent PTYs advertise `TERM=xterm-256color` and `COLORTERM=truecolor` because xterm.js is the actual frontend; applications may therefore negotiate xterm mouse reporting without a Wakey-specific mouse protocol.
 
 WebSockets already provide ordered, reliable delivery, so terminal frames do not carry sequence numbers in the initial protocol.
 
@@ -69,6 +70,8 @@ Terminal sessions allow only one attached operator. Their lifetime belongs to th
 
 When the operator socket disconnects, the session becomes detached. Navigation, browser closure, and network loss do not terminate the PTY. A protected operator request may discover the live session and obtain a fresh, single-use attachment credential. The control plane retains only a bounded in-memory output buffer during the gap and replays it before live output on reattachment.
 
+After operator attachment, the control plane sends a `refresh` hint. The agent queries the PTY's current foreground process group and sends it `SIGWINCH`, allowing full-screen applications to redraw without assuming that the original shell remains in the foreground. Refresh failure is logged but never terminates the session.
+
 The agent keeps its terminal manager outside the control-WebSocket reconnect loop. If a dedicated terminal relay disconnects, the PTY continues draining output into a bounded local replay buffer while waiting for replacement relay credentials. On every authenticated control connection, the agent reports its in-memory live terminal IDs and creation times. The control plane reconciles that inventory, adopts sessions missing from its volatile registry, and issues fresh relay credentials. This allows sessions to survive control-plane restart without persisting terminal state in SQLite.
 
 Closing a session must terminate the entire PTY process group, not only the shell process. Cleanup should send a hangup first, then escalate to termination and forced kill after bounded grace periods. Explicit operator close, absolute session timeout, agent process exit, and machine reboot close the session.
@@ -83,6 +86,8 @@ Terminal transport must not use unbounded queues.
 
 Every queue between PTY, agent socket, control-plane relay, and browser socket is bounded. Agent and control-plane replay buffers are bounded as well. The agent keeps draining the PTY while detached so a noisy child cannot stall the agent; output older than the replay bound is discarded.
 
+When input/control and output are ready simultaneously, relay loops prioritize input and control. This keeps interrupt, resize, close, and refresh traffic responsive while commands produce sustained output.
+
 The implementation also enforces:
 
 - maximum terminal frame size;

wakey-agent/src/protocol.rs

@@ -57,6 +57,7 @@ pub struct AgentTerminalSession {
 #[serde(tag = "type", rename_all = "snake_case")]
 pub enum TerminalControl {
     Resize { rows: u16, cols: u16 },
+    Refresh,
     Ready,
     Exited { exit_code: Option<i32> },
     Error { code: String, message: String },
@@ -335,6 +336,8 @@ mod tests {
         };
         let json = serde_json::to_string(&resize).expect("serialize resize");
         assert_eq!(json, r#"{"type":"resize","rows":40,"cols":160}"#);
+        let refresh = serde_json::to_string(&TerminalControl::Refresh).expect("serialize refresh");
+        assert_eq!(refresh, r#"{"type":"refresh"}"#);
 
         let inventory = ClientMessage::TerminalSessions {
             sessions: vec![AgentTerminalSession {

wakey-agent/src/terminal.rs

@@ -178,11 +178,7 @@ async fn run_terminal(
             return Err(err);
         }
     };
-    let wakey::wakey_linux::terminal::TerminalPty {
-        mut reader,
-        mut writer,
-        mut child,
-    } = terminal;
+    let (mut reader, mut writer, mut child) = terminal.into_parts();
     let process_group = child.id();
     info!(terminal_id = %terminal_id, shell = %config.terminal.shell.display(), "terminal PTY ready");
 
@@ -197,6 +193,7 @@ async fn run_terminal(
     let mut observed_status = None;
     loop {
         tokio::select! {
+            biased;
             _ = &mut cancel => {
                 requested_close = true;
                 break;
@@ -205,18 +202,42 @@ async fn run_terminal(
                 observed_status = Some(status.context("failed waiting for terminal child")?);
                 break;
             }
-            read = reader.read(&mut output) => {
-                match read {
-                    Ok(0) => break,
-                    Ok(count) => send_terminal_output(
-                        Message::Binary(output[..count].to_vec().into()),
-                        &mut relay_output,
-                        &mut replay,
-                        &mut replay_bytes,
-                    ).await,
-                    // Linux PTY masters commonly report EIO after the slave closes.
-                    Err(err) if err.raw_os_error() == Some(5) => break,
-                    Err(err) => return Err(err).context("failed to read PTY output"),
+            incoming = relay_input_rx.recv() => {
+                let Some(message) = incoming else { break; };
+                match message {
+                    RelayInput::Binary { generation, bytes } if generation == relay_generation => {
+                        if bytes.len() > MAX_TERMINAL_FRAME_BYTES {
+                            anyhow::bail!("terminal input frame exceeds size limit");
+                        }
+                        writer.write_all(&bytes).await.context("failed to write PTY input")?;
+                    }
+                    RelayInput::Resize { generation, rows, cols } if generation == relay_generation => {
+                        validate_size(rows, cols)?;
+                        writer.resize(rows, cols)?;
+                    }
+                    RelayInput::Refresh { generation } if generation == relay_generation => {
+                        if let Err(err) = writer.refresh() {
+                            warn!(terminal_id = %terminal_id, error = %err, "terminal redraw signal failed");
+                        }
+                    }
+                    RelayInput::Close { generation } if generation == relay_generation => {
+                        requested_close = true;
+                        break;
+                    }
+                    RelayInput::Connected { generation, output } if generation == relay_generation => {
+                        relay_output = Some(output.clone());
+                        while let Some(frame) = replay.pop_front() {
+                            replay_bytes = replay_bytes.saturating_sub(message_size(&frame));
+                            if output.send(frame).await.is_err() {
+                                relay_output = None;
+                                break;
+                            }
+                        }
+                    }
+                    RelayInput::Disconnected { generation } if generation == relay_generation => {
+                        relay_output = None;
+                    }
+                    _ => {}
                 }
             }
             credential = relay_credentials.recv() => {
@@ -249,37 +270,18 @@ async fn run_terminal(
                     let _ = relay_input_tx.send(RelayInput::Disconnected { generation });
                 }));
             }
-            incoming = relay_input_rx.recv() => {
-                let Some(message) = incoming else { break; };
-                match message {
-                    RelayInput::Binary { generation, bytes } if generation == relay_generation => {
-                        if bytes.len() > MAX_TERMINAL_FRAME_BYTES {
-                            anyhow::bail!("terminal input frame exceeds size limit");
-                        }
-                        writer.write_all(&bytes).await.context("failed to write PTY input")?;
-                    }
-                    RelayInput::Resize { generation, rows, cols } if generation == relay_generation => {
-                        validate_size(rows, cols)?;
-                        wakey::wakey_linux::terminal::resize_terminal(&writer, rows, cols)?;
-                    }
-                    RelayInput::Close { generation } if generation == relay_generation => {
-                        requested_close = true;
-                        break;
-                    }
-                    RelayInput::Connected { generation, output } if generation == relay_generation => {
-                        relay_output = Some(output.clone());
-                        while let Some(frame) = replay.pop_front() {
-                            replay_bytes = replay_bytes.saturating_sub(message_size(&frame));
-                            if output.send(frame).await.is_err() {
-                                relay_output = None;
-                                break;
-                            }
-                        }
-                    }
-                    RelayInput::Disconnected { generation } if generation == relay_generation => {
-                        relay_output = None;
-                    }
-                    _ => {}
+            read = reader.read(&mut output) => {
+                match read {
+                    Ok(0) => break,
+                    Ok(count) => send_terminal_output(
+                        Message::Binary(output[..count].to_vec().into()),
+                        &mut relay_output,
+                        &mut replay,
+                        &mut replay_bytes,
+                    ).await,
+                    // Linux PTY masters commonly report EIO after the slave closes.
+                    Err(err) if err.raw_os_error() == Some(5) => break,
+                    Err(err) => return Err(err).context("failed to read PTY output"),
                 }
             }
         }
@@ -315,6 +317,9 @@ enum RelayInput {
         rows: u16,
         cols: u16,
     },
+    Refresh {
+        generation: u64,
+    },
     Close {
         generation: u64,
     },
@@ -367,10 +372,7 @@ async fn run_terminal_relay(relay: RelayConnection) -> Result<()> {
     let mut output = relay.output_rx;
     loop {
         tokio::select! {
-            outgoing = output.recv() => {
-                let Some(message) = outgoing else { break; };
-                sink.send(message).await.context("failed to send terminal relay output")?;
-            }
+            biased;
             incoming = source.next() => {
                 let Some(message) = incoming else { break; };
                 match message.context("terminal relay websocket receive failed")? {
@@ -391,6 +393,11 @@ async fn run_terminal_relay(relay: RelayConnection) -> Result<()> {
                             })
                                 .map_err(|_| anyhow::anyhow!("terminal worker stopped"))?;
                         }
+                        TerminalControl::Refresh => {
+                            relay.input.send(RelayInput::Refresh {
+                                generation: relay.generation,
+                            }).map_err(|_| anyhow::anyhow!("terminal worker stopped"))?;
+                        }
                         TerminalControl::Close => {
                             let _ = relay.input.send(RelayInput::Close {
                                 generation: relay.generation,
@@ -407,6 +414,10 @@ async fn run_terminal_relay(relay: RelayConnection) -> Result<()> {
                     Message::Frame(_) => {}
                 }
             }
+            outgoing = output.recv() => {
+                let Some(message) = outgoing else { break; };
+                sink.send(message).await.context("failed to send terminal relay output")?;
+            }
         }
     }
     Ok(())

wakey-control-plane/src/api/terminals.rs

@@ -274,6 +274,7 @@ async fn handle_agent_terminal_socket(state: AppState, terminal_id: String, mut
     }
     loop {
         tokio::select! {
+            biased;
             outbound_frame = outbound.recv() => {
                 let Some(frame) = outbound_frame else { break; };
                 if send_relay_frame(&mut write, frame).await.is_err() { break; }
@@ -332,6 +333,15 @@ async fn handle_operator_terminal_socket(
         }
     };
     info!(terminal_id, "operator terminal socket attached");
+    let refresh = serde_json::to_string(&TerminalControl::Refresh)
+        .expect("terminal refresh control serializes");
+    if let Err(code) = state
+        .terminals
+        .relay_to_agent(&terminal_id, TerminalRelayFrame::Text(refresh))
+        .await
+    {
+        warn!(terminal_id, code, "failed to request terminal redraw");
+    }
     let summary = state.terminals.summary(&terminal_id).await;
     if let Some((agent_id, _, _, _)) = &summary {
         append_terminal_audit(
@@ -370,10 +380,7 @@ async fn handle_operator_terminal_socket(
     let mut explicit_close = false;
     loop {
         tokio::select! {
-            outbound_frame = outbound.recv() => {
-                let Some(frame) = outbound_frame else { break; };
-                if send_relay_frame(&mut write, frame).await.is_err() { break; }
-            }
+            biased;
             incoming = read.next() => {
                 let Some(Ok(message)) = incoming else { break; };
                 match operator_relay_frame(message) {
@@ -389,6 +396,10 @@ async fn handle_operator_terminal_socket(
                     }
                 }
             }
+            outbound_frame = outbound.recv() => {
+                let Some(frame) = outbound_frame else { break; };
+                if send_relay_frame(&mut write, frame).await.is_err() { break; }
+            }
         }
     }
 
@@ -645,5 +656,8 @@ mod tests {
 
         let ready = serde_json::to_string(&TerminalControl::Ready).expect("ready json");
         assert!(operator_relay_frame(Message::Text(ready.into())).is_err());
+
+        let refresh = serde_json::to_string(&TerminalControl::Refresh).expect("refresh json");
+        assert!(operator_relay_frame(Message::Text(refresh.into())).is_err());
     }
 }

wakey-linux/Cargo.toml

@@ -23,4 +23,9 @@ version = "0"
 features = ["experimental-nl"]
 
 [target.'cfg(unix)'.dependencies]
+nix = { version = "0.31", default-features = false, features = [
+    "signal",
+    "process",
+    "term",
+] }
 pty-process = { version = "0.5", features = ["async"] }

wakey-linux/src/terminal.rs

@@ -1,9 +1,13 @@
 //! Unix PTY ownership for interactive agent terminal sessions.
 
+use std::os::fd::{AsFd, OwnedFd};
 use std::path::Path;
+use std::pin::Pin;
+use std::task::{Context as TaskContext, Poll};
 
 use anyhow::{Context, Result};
 use pty_process::{Command, OwnedReadPty, OwnedWritePty, Size};
+use tokio::io::AsyncWrite;
 use tokio::process::Child;
 
 /// An interactive child process and the independently owned sides of its PTY.
@@ -11,9 +15,15 @@ use tokio::process::Child;
 /// Keeping the write half available is important: `OwnedWritePty` also owns the
 /// resize operation used by terminal WebSocket control frames.
 pub struct TerminalPty {
-    pub reader: OwnedReadPty,
-    pub writer: OwnedWritePty,
-    pub child: Child,
+    reader: OwnedReadPty,
+    writer: TerminalWriter,
+    child: Child,
+}
+
+/// Writable PTY half with the control operations needed by the agent.
+pub struct TerminalWriter {
+    io: OwnedWritePty,
+    control: OwnedFd,
 }
 
 impl TerminalPty {
@@ -22,27 +32,73 @@ impl TerminalPty {
         let (pty, pts) = pty_process::open().context("failed to open PTY")?;
         pty.resize(Size::new(rows, cols))
             .context("failed to set initial PTY size")?;
+        let control = pty
+            .as_fd()
+            .try_clone_to_owned()
+            .context("failed to duplicate PTY control descriptor")?;
 
-        let command = Command::new(program).kill_on_drop(true);
+        // The remote frontend is xterm.js regardless of the daemon's own
+        // environment, so advertise the terminal the child actually receives.
+        let command = Command::new(program)
+            .env("TERM", "xterm-256color")
+            .env("COLORTERM", "truecolor")
+            .kill_on_drop(true);
         let child = command
             .spawn(pts)
             .with_context(|| format!("failed to spawn {} in PTY", program.display()))?;
-        let (reader, writer) = pty.into_split();
+        let (reader, io) = pty.into_split();
 
         Ok(Self {
             reader,
-            writer,
+            writer: TerminalWriter { io, control },
             child,
         })
     }
+
+    /// Consumes the terminal into independently driven async parts.
+    pub fn into_parts(self) -> (OwnedReadPty, TerminalWriter, Child) {
+        (self.reader, self.writer, self.child)
+    }
 }
 
-/// Resizes an owned PTY writer without exposing `pty-process` protocol types
-/// to higher-level crates.
-pub fn resize_terminal(writer: &OwnedWritePty, rows: u16, cols: u16) -> Result<()> {
-    writer
+impl TerminalWriter {
+    pub fn resize(&self, rows: u16, cols: u16) -> Result<()> {
+        self.io
             .resize(Size::new(rows, cols))
             .context("failed to resize PTY")
+    }
+
+    /// Requests a full-screen application to redraw after reattachment.
+    ///
+    /// Job-control shells give foreground programs their own process groups,
+    /// so signaling the original shell group would miss programs like `btop`.
+    pub fn refresh(&self) -> Result<()> {
+        let foreground = nix::unistd::tcgetpgrp(&self.control)
+            .context("failed to get PTY foreground process group")?;
+        nix::sys::signal::killpg(foreground, nix::sys::signal::Signal::SIGWINCH)
+            .context("failed to signal PTY foreground process group")
+    }
+}
+
+impl AsyncWrite for TerminalWriter {
+    fn poll_write(
+        mut self: Pin<&mut Self>,
+        cx: &mut TaskContext<'_>,
+        buf: &[u8],
+    ) -> Poll<std::io::Result<usize>> {
+        Pin::new(&mut self.io).poll_write(cx, buf)
+    }
+
+    fn poll_flush(mut self: Pin<&mut Self>, cx: &mut TaskContext<'_>) -> Poll<std::io::Result<()>> {
+        Pin::new(&mut self.io).poll_flush(cx)
+    }
+
+    fn poll_shutdown(
+        mut self: Pin<&mut Self>,
+        cx: &mut TaskContext<'_>,
+    ) -> Poll<std::io::Result<()>> {
+        Pin::new(&mut self.io).poll_shutdown(cx)
+    }
 }
 
 #[cfg(test)]
@@ -59,30 +115,33 @@ mod tests {
     /// `./scripts/test_remote.ps1 -Package wakey-linux -Filter terminal::tests::pty_round_trip_and_resize -Exact -ShowOutput`
     #[tokio::test]
     async fn pty_round_trip_and_resize() {
-        let TerminalPty {
-            mut reader,
-            mut writer,
-            mut child,
-        } = TerminalPty::spawn(Path::new("/bin/sh"), 24, 80).expect("spawn PTY shell");
+        let (mut reader, mut writer, mut child) = TerminalPty::spawn(Path::new("/bin/sh"), 24, 80)
+            .expect("spawn PTY shell")
+            .into_parts();
 
+        writer.resize(31, 101).expect("resize owned PTY writer");
         writer
-            .resize(Size::new(31, 101))
-            .expect("resize owned PTY writer");
-        writer
-            .write_all(b"read line; printf 'WAKEY_PTY_OK:%s\\n' \"$line\"; exit 0\nprobe\n")
+            .write_all(
+                b"trap 'printf \"WAKEY_WINCH\\n\"' WINCH; \
+                  printf 'WAKEY_ENV:%s:%s\\n' \"$TERM\" \"$COLORTERM\"; \
+                  read line; printf 'WAKEY_PTY_OK:%s\\n' \"$line\"; exit 0\n",
+            )
             .await
             .expect("write shell input");
-        let output = tokio::time::timeout(Duration::from_secs(5), async {
         let mut output = Vec::new();
-            let mut chunk = [0_u8; 4096];
-            while !String::from_utf8_lossy(&output).contains("WAKEY_PTY_OK:probe") {
-                // Use a fresh ReadBuf for each PTY read. pty-process 0.5.3's
-                // AsyncRead implementation cannot safely extend the partially
-                // filled ReadBuf used internally by Tokio's `read_to_end`.
-                let count = reader.read(&mut chunk).await.expect("read PTY output");
-                assert_ne!(count, 0, "PTY closed before emitting probe marker");
-                output.extend_from_slice(&chunk[..count]);
-            }
+        read_until(
+            &mut reader,
+            &mut output,
+            "WAKEY_ENV:xterm-256color:truecolor",
+        )
+        .await;
+        writer.refresh().expect("signal foreground process group");
+        writer
+            .write_all(b"probe\n")
+            .await
+            .expect("write shell probe");
+        let output = tokio::time::timeout(Duration::from_secs(5), async {
+            read_until(&mut reader, &mut output, "WAKEY_PTY_OK:probe").await;
             output
         })
         .await
@@ -98,5 +157,25 @@ mod tests {
             output.contains("WAKEY_PTY_OK:probe"),
             "unexpected PTY output: {output:?}"
         );
+        assert!(
+            output.contains("WAKEY_WINCH"),
+            "foreground process did not receive SIGWINCH: {output:?}"
+        );
+    }
+
+    async fn read_until(reader: &mut OwnedReadPty, output: &mut Vec<u8>, marker: &str) {
+        tokio::time::timeout(Duration::from_secs(5), async {
+            let mut chunk = [0_u8; 4096];
+            while !String::from_utf8_lossy(output).contains(marker) {
+                // Use a fresh ReadBuf for each PTY read. pty-process 0.5.3's
+                // AsyncRead implementation cannot safely extend Tokio's
+                // partially filled buffer used by `read_to_end`.
+                let count = reader.read(&mut chunk).await.expect("read PTY output");
+                assert_ne!(count, 0, "PTY closed before emitting {marker}");
+                output.extend_from_slice(&chunk[..count]);
+            }
+        })
+        .await
+        .unwrap_or_else(|_| panic!("PTY output timed out waiting for {marker}"));
     }
 }