fix(terminal): harden relay lifecycle and session cleanup

- bound relay input and detach saturated output transports
- drain PTY output after child exit
- make terminal close lifecycle race-safe
- prune expired CC sessions and audit closure reasons
- require matching agent hello/auth identities
- validate terminal IDs and PTY dimensions

.gitmodules

@@ -0,0 +1,3 @@
+[submodule "vendor/vt100"]
+	path = vendor/vt100
+	url = https://github.com/ldlda/vt100-rust-1

Cargo.lock

@@ -3615,8 +3615,6 @@ checksum = "0b928f33d975fc6ad9f86c8f283853ad26bdd5b10b7f1542aa2fa15e2289105a"
 [[package]]
 name = "vt100"
 version = "0.16.2"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "054ff75fb8fa83e609e685106df4faeffdf3a735d3c74ebce97ec557d5d36fd9"
 dependencies = [
  "itoa",
  "unicode-width",
@@ -3673,7 +3671,6 @@ dependencies = [
  "toml",
  "tracing",
  "tracing-subscriber",
- "unicode-width",
  "url",
  "vt100",
  "wakey",

Cargo.toml

@@ -31,6 +31,13 @@ strip = true
 
 [workspace]
 members = ["ipjs", "wakey-agent", "wakey-control-plane", "wakey-core", "wakey-linux"]
+exclude = ["vendor/vt100"]
+
+[patch.crates-io]
+# vt100 exposes only its active grid. Wakey's reconnect snapshot needs the
+# retained primary grid while a full-screen application owns the alternate
+# grid, so keep the small read-only extension local and auditable.
+vt100 = { path = "vendor/vt100" }
 
 [workspace.dependencies]
 anyhow = "1"

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

@@ -74,6 +74,8 @@ The operator UI lists live sessions as tabs and stores only the active terminal
 
 After operator attachment, the control plane sends a `snapshot` request. The agent serializes up to 5,000 retained physical rows followed by its parsed current screen and input modes as terminal escape bytes, sends them before subsequent live output, and then signals the PTY's current foreground process group with `SIGWINCH`. Historical rows retain their terminal attributes and are replayed as ordinary terminal output so xterm builds its native scrollback; the final formatted state restores the exact live screen. The signal lets full-screen applications redraw without assuming that the original shell remains in the foreground. Signal failure is logged but never terminates the session.
 
+Wakey carries a small, read-only extension to `vt100` because its public API exposes only the active grid. A snapshot streams bounded history from the primary grid, redraws that grid with its saved cursor and attributes, and then reconstructs the alternate grid when a full-screen application is active. The parser is never resized or switched while taking a snapshot. Consequently, reconnecting during `btop` restores the application while preserving the hidden shell history and state that reappear when it exits.
+
 The agent keeps its terminal manager outside the control-WebSocket reconnect loop. If a dedicated terminal relay disconnects, the PTY continues draining output into the agent-owned terminal parser 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.

remote/wslfix.sh

@@ -1,4 +1,5 @@
 #!/usr/bin/env bash
+set -euo pipefail
 
 # sometimes it breaks, run to fix
 

ui/src/pages/TerminalPage.tsx

@@ -570,34 +570,44 @@ export function TerminalPage({
     const fallback = remaining.find((item) => !item.operator_attached);
 
     if (closesActiveSession) {
-      socketRef.current?.send(JSON.stringify({ type: "close" }));
+      if (socketRef.current?.readyState === WebSocket.OPEN) {
+        socketRef.current.send(JSON.stringify({ type: "close" }));
+      }
       detachTransport();
     }
     try {
       await closeTerminal(closingId);
     } catch (error) {
       toast.error("Terminal cleanup failed", { description: String(error) });
-    } finally {
+      void listTerminals()
-      setSessions(remaining);
+        .then((listed) =>
-      setSessionTitles((current) => {
+          setSessions((current) => reconcileTerminalSessions(current, listed)),
-        const next = { ...current };
+        )
-        delete next[closingId];
+        .catch(() => undefined);
-        return next;
+      return;
-      });
+    }
-      if (
+
-        window.sessionStorage.getItem(REMEMBERED_TERMINAL_KEY) === closingId
+    setSessions((current) =>
-      ) {
+      current.filter((item) => item.terminal_id !== closingId),
-        window.sessionStorage.removeItem(REMEMBERED_TERMINAL_KEY);
+    );
-      }
+    setSessionTitles((current) => {
-      if (closesActiveSession) {
+      const next = { ...current };
-        setSession(null);
+      delete next[closingId];
-        activeSessionRef.current = null;
+      return next;
-        setConnection("idle");
+    });
-        // xterm.clear() deliberately preserves the active cursor line. Closing
+    if (
-        // a session should discard its complete screen and terminal modes.
+      window.sessionStorage.getItem(REMEMBERED_TERMINAL_KEY) === closingId
-        terminalRef.current?.reset();
+    ) {
-        if (fallback) void activateSession(fallback);
+      window.sessionStorage.removeItem(REMEMBERED_TERMINAL_KEY);
-      }
+    }
+    if (closesActiveSession) {
+      setSession(null);
+      activeSessionRef.current = null;
+      setConnection("idle");
+      // xterm.clear() deliberately preserves the active cursor line. Closing
+      // a session should discard its complete screen and terminal modes.
+      terminalRef.current?.reset();
+      if (fallback) void activateSession(fallback);
     }
   }
 

wakey-agent/Cargo.toml

@@ -39,7 +39,6 @@ tracing-subscriber = { version = "0.3", features = [
 ] }
 time = { version = "0.3", features = ["formatting", "local-offset"] }
 url = "2"
-unicode-width = "0.2"
 vt100 = "0.16.2"
 wakey = { path = "..", registry = "gitea", version = "0" }
 wakey-core = { path = "../wakey-core", registry = "gitea", version = "0" }

wakey-agent/src/config.rs

@@ -132,7 +132,7 @@ fn default_observation_store_path() -> PathBuf {
 }
 
 fn default_terminal_shell() -> PathBuf {
-    "/bin/ash".into()
+    "/bin/sh".into()
 }
 
 const fn default_terminal_max_sessions() -> usize {

wakey-agent/src/protocol.rs

@@ -17,7 +17,7 @@ impl RequestId {
     }
 }
 
-#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
+#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize)]
 #[serde(transparent)]
 pub struct TerminalId(String);
 
@@ -41,6 +41,24 @@ impl fmt::Display for TerminalId {
     }
 }
 
+impl TryFrom<String> for TerminalId {
+    type Error = String;
+
+    fn try_from(value: String) -> Result<Self, Self::Error> {
+        Self::new(value)
+    }
+}
+
+impl<'de> Deserialize<'de> for TerminalId {
+    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+    where
+        D: serde::Deserializer<'de>,
+    {
+        let raw = String::deserialize(deserializer)?;
+        Self::try_from(raw).map_err(serde::de::Error::custom)
+    }
+}
+
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
 #[serde(rename_all = "snake_case")]
 pub enum AgentCapability {
@@ -297,6 +315,16 @@ mod tests {
         assert!(err.contains("must not be empty"));
     }
 
+    #[test]
+    fn terminal_id_deserialization_enforces_validation() {
+        let valid: TerminalId = serde_json::from_str(r#""term-1""#).expect("valid terminal id");
+        assert_eq!(valid.as_str(), "term-1");
+
+        let error = serde_json::from_str::<TerminalId>(r#""   ""#)
+            .expect_err("empty terminal id must fail");
+        assert!(error.to_string().contains("must not be empty"));
+    }
+
     #[test]
     fn client_result_with_devs_serializes() {
         let msg = ClientMessage::Result {

wakey-agent/src/terminal.rs

@@ -11,10 +11,11 @@ use tokio::io::{AsyncReadExt, AsyncWriteExt};
 use tokio::sync::{mpsc, oneshot};
 use tokio_tungstenite::tungstenite::Message;
 use tracing::{info, warn};
-use unicode_width::UnicodeWidthStr;
 
 const MAX_TERMINAL_FRAME_BYTES: usize = 64 * 1024;
 const TERMINAL_SCROLLBACK_ROWS: usize = 5_000;
+const RELAY_INPUT_QUEUE: usize = 32;
+const PTY_EXIT_DRAIN_TIMEOUT: Duration = Duration::from_secs(1);
 const PROCESS_SIGNAL_GRACE: Duration = Duration::from_secs(1);
 
 /// Tracks the terminal's current rendered state while the browser is detached.
@@ -40,68 +41,15 @@ impl TerminalState {
         self.parser.screen_mut().set_size(rows, cols);
     }
 
-    fn snapshot(&mut self) -> Vec<u8> {
+    fn snapshot(&self) -> Vec<u8> {
-        let screen = self.parser.screen_mut();
+        self.parser
-        let (rows, cols) = screen.size();
+            .screen()
-        let alternate_screen = screen.alternate_screen();
+            .snapshot_formatted(TERMINAL_SCROLLBACK_ROWS)
-        let mut physical_rows = Vec::new();
-
-        // set_scrollback changes the viewport exposed by Screen. Walking its
-        // top row from the maximum offset down to zero yields every retained
-        // physical row exactly once, oldest first.
-        screen.set_scrollback(usize::MAX);
-        let retained_rows = screen.scrollback();
-        for offset in (1..=retained_rows).rev() {
-            screen.set_scrollback(offset);
-            let contents = screen.rows(0, cols).next().unwrap_or_default();
-            physical_rows.push((
-                screen.rows_formatted(0, cols).next().unwrap_or_default(),
-                UnicodeWidthStr::width_cjk(contents.as_str()).min(usize::from(cols)),
-                screen.row_wrapped(0),
-            ));
-        }
-        screen.set_scrollback(0);
-        physical_rows.extend((0..rows).map(|row| {
-            let contents = screen
-                .rows(0, cols)
-                .nth(usize::from(row))
-                .unwrap_or_default();
-            (
-                screen
-                    .rows_formatted(0, cols)
-                    .nth(usize::from(row))
-                    .unwrap_or_default(),
-                UnicodeWidthStr::width_cjk(contents.as_str()).min(usize::from(cols)),
-                screen.row_wrapped(row),
-            )
-        }));
-
-        // Replace browser history, then stream physical rows so xterm builds
-        // its own scrollback. The final formatted state restores colors,
-        // cursor position, and input modes for the live screen.
-        let mut snapshot = b"\x1b[?1049l\x1b[3J\x1b[2J\x1b[H".to_vec();
-        for (index, (contents, display_width, wrapped)) in physical_rows.iter().enumerate() {
-            // rows_formatted encodes each row relative to default attributes.
-            // Reset between rows so attributes cannot leak across boundaries.
-            snapshot.extend_from_slice(b"\x1b[0m");
-            snapshot.extend_from_slice(contents);
-            snapshot.extend_from_slice(b"\x1b[0m");
-            if *wrapped {
-                snapshot.resize(snapshot.len() + usize::from(cols) - display_width, b' ');
-            } else if index + 1 < physical_rows.len() {
-                snapshot.extend_from_slice(b"\r\n");
-            }
-        }
-        if alternate_screen {
-            snapshot.extend_from_slice(b"\x1b[?1049h");
-        }
-        snapshot.extend_from_slice(&screen.state_formatted());
-        snapshot
     }
 }
 
 struct ActiveTerminal {
-    cancel: oneshot::Sender<()>,
+    cancel: Option<oneshot::Sender<()>>,
     relay_credentials: mpsc::UnboundedSender<String>,
     created_at_unix: u64,
 }
@@ -142,6 +90,7 @@ impl TerminalManager {
         if !config.terminal.enabled {
             anyhow::bail!("terminal capability is disabled");
         }
+        validate_size(rows, cols)?;
 
         let terminal_key = terminal_id.to_string();
         let (cancel_tx, cancel_rx) = oneshot::channel();
@@ -161,7 +110,7 @@ impl TerminalManager {
             active.insert(
                 terminal_key.clone(),
                 ActiveTerminal {
-                    cancel: cancel_tx,
+                    cancel: Some(cancel_tx),
                     relay_credentials: relay_tx.clone(),
                     created_at_unix,
                 },
@@ -189,30 +138,38 @@ impl TerminalManager {
     }
 
     pub fn close(&self, terminal_id: &TerminalId) -> bool {
-        self.active
+        let mut active = self.active.lock().expect("terminal manager poisoned");
-            .lock()
+        let Some(terminal) = active.get_mut(terminal_id.as_str()) else {
-            .expect("terminal manager poisoned")
+            return false;
-            .remove(terminal_id.as_str())
+        };
-            .is_some_and(|active| active.cancel.send(()).is_ok())
+        let Some(cancel) = terminal.cancel.take() else {
+            return true;
+        };
+        // A failed send means the worker has already dropped its receiver.
+        // Its wrapper still owns final registry removal, so this close request
+        // is successful either way.
+        let _ = cancel.send(());
+        true
     }
 
     pub fn resume(&self, terminal_id: &TerminalId, relay_token: String) -> Result<()> {
-        let mut active = self.active.lock().expect("terminal manager poisoned");
+        let active = self.active.lock().expect("terminal manager poisoned");
         let session = active
             .get(terminal_id.as_str())
             .with_context(|| format!("terminal session {terminal_id} is not active"))?;
         if session.relay_credentials.send(relay_token).is_err() {
-            active.remove(terminal_id.as_str());
             anyhow::bail!("terminal session {terminal_id} has stopped");
         }
         Ok(())
     }
 
     pub fn sessions(&self) -> Vec<AgentTerminalSession> {
-        let mut active = self.active.lock().expect("terminal manager poisoned");
+        let active = self.active.lock().expect("terminal manager poisoned");
-        active.retain(|_, terminal| !terminal.relay_credentials.is_closed());
         active
             .iter()
+            .filter(|(_, terminal)| {
+                terminal.cancel.is_some() && !terminal.relay_credentials.is_closed()
+            })
             .filter_map(|(terminal_id, active)| {
                 TerminalId::new(terminal_id.clone())
                     .ok()
@@ -231,7 +188,9 @@ impl Drop for TerminalManager {
             && let Ok(mut active) = self.active.lock()
         {
             for (_, active) in active.drain() {
-                let _ = active.cancel.send(());
+                if let Some(cancel) = active.cancel {
+                    let _ = cancel.send(());
+                }
             }
         }
     }
@@ -268,7 +227,7 @@ async fn run_terminal(
     let process_group = child.id();
     info!(terminal_id = %terminal_id, shell = %config.terminal.shell.display(), "terminal PTY ready");
 
-    let (relay_input_tx, mut relay_input_rx) = mpsc::unbounded_channel();
+    let (relay_input_tx, mut relay_input_rx) = mpsc::channel(RELAY_INPUT_QUEUE);
     let mut relay_output: Option<mpsc::Sender<Message>> = None;
     let mut relay_task: Option<tokio::task::JoinHandle<()>> = None;
     let mut relay_generation = 0_u64;
@@ -276,6 +235,7 @@ async fn run_terminal(
     let mut output = [0_u8; 16 * 1024];
     let mut requested_close = false;
     let mut observed_status = None;
+    let mut drain_deadline: Option<std::pin::Pin<Box<tokio::time::Sleep>>> = None;
     loop {
         tokio::select! {
             biased;
@@ -283,8 +243,16 @@ async fn run_terminal(
                 requested_close = true;
                 break;
             }
-            status = child.wait() => {
+            status = child.wait(), if observed_status.is_none() => {
                 observed_status = Some(status.context("failed waiting for terminal child")?);
+                drain_deadline = Some(Box::pin(tokio::time::sleep(PTY_EXIT_DRAIN_TIMEOUT)));
+            }
+            _ = async {
+                match drain_deadline.as_mut() {
+                    Some(deadline) => deadline.as_mut().await,
+                    None => std::future::pending().await,
+                }
+            } => {
                 break;
             }
             incoming = relay_input_rx.recv() => {
@@ -302,7 +270,12 @@ async fn run_terminal(
                         terminal_state.resize(rows, cols);
                     }
                     RelayInput::Snapshot { generation } if generation == relay_generation => {
-                        send_terminal_snapshot(terminal_state.snapshot(), &mut relay_output).await;
+                        if !send_terminal_snapshot(terminal_state.snapshot(), &mut relay_output) {
+                            if let Some(task) = relay_task.take() {
+                                task.abort();
+                            }
+                            warn!(terminal_id = %terminal_id, "terminal relay output queue saturated during snapshot");
+                        }
                         if let Err(err) = writer.refresh() {
                             warn!(terminal_id = %terminal_id, error = %err, "terminal redraw signal failed");
                         }
@@ -344,7 +317,7 @@ async fn run_terminal(
                     }).await {
                         warn!(terminal_id = %terminal_id, error = %err, "terminal relay disconnected");
                     }
-                    let _ = relay_input_tx.send(RelayInput::Disconnected { generation });
+                    let _ = relay_input_tx.send(RelayInput::Disconnected { generation }).await;
                 }));
             }
             read = reader.read(&mut output) => {
@@ -352,10 +325,15 @@ async fn run_terminal(
                     Ok(0) => break,
                     Ok(count) => {
                         terminal_state.process(&output[..count]);
-                        send_terminal_output(
+                        if !send_terminal_output(
                             Message::Binary(output[..count].to_vec().into()),
                             &mut relay_output,
-                        ).await;
+                        ) {
+                            if let Some(task) = relay_task.take() {
+                                task.abort();
+                            }
+                            warn!(terminal_id = %terminal_id, "terminal relay output queue saturated");
+                        }
                     }
                     // Linux PTY masters commonly report EIO after the slave closes.
                     Err(err) if err.raw_os_error() == Some(5) => break,
@@ -374,7 +352,7 @@ async fn run_terminal(
             exit_code: status.code(),
         };
         if let Ok(text) = serde_json::to_string(&control) {
-            let _ = output.send(Message::Text(text.into())).await;
+            let _ = output.try_send(Message::Text(text.into()));
         }
     }
     info!(terminal_id = %terminal_id, exit_code = ?status.code(), requested_close, "terminal worker exited");
@@ -414,7 +392,7 @@ struct RelayConnection {
     generation: u64,
     output_tx: mpsc::Sender<Message>,
     output_rx: mpsc::Receiver<Message>,
-    input: mpsc::UnboundedSender<RelayInput>,
+    input: mpsc::Sender<RelayInput>,
 }
 
 #[cfg(unix)]
@@ -439,6 +417,7 @@ async fn run_terminal_relay(relay: RelayConnection) -> Result<()> {
             generation: relay.generation,
             output: relay.output_tx,
         })
+        .await
         .map_err(|_| anyhow::anyhow!("terminal worker stopped"))?;
 
     let mut output = relay.output_rx;
@@ -452,7 +431,7 @@ async fn run_terminal_relay(relay: RelayConnection) -> Result<()> {
                         relay.input.send(RelayInput::Binary {
                             generation: relay.generation,
                             bytes: bytes.to_vec(),
-                        }).map_err(|_| anyhow::anyhow!("terminal worker stopped"))?;
+                        }).await.map_err(|_| anyhow::anyhow!("terminal worker stopped"))?;
                     }
                     Message::Text(text) => match serde_json::from_str::<TerminalControl>(&text)
                         .context("invalid terminal control frame")?
@@ -462,18 +441,18 @@ async fn run_terminal_relay(relay: RelayConnection) -> Result<()> {
                                 generation: relay.generation,
                                 rows,
                                 cols,
-                            })
+                            }).await
                                 .map_err(|_| anyhow::anyhow!("terminal worker stopped"))?;
                         }
                         TerminalControl::Snapshot => {
                             relay.input.send(RelayInput::Snapshot {
                                 generation: relay.generation,
-                            }).map_err(|_| anyhow::anyhow!("terminal worker stopped"))?;
+                            }).await.map_err(|_| anyhow::anyhow!("terminal worker stopped"))?;
                         }
                         TerminalControl::Close => {
                             let _ = relay.input.send(RelayInput::Close {
                                 generation: relay.generation,
-                            });
+                            }).await;
                             break;
                         }
                         _ => anyhow::bail!("terminal control frame has invalid direction"),
@@ -495,28 +474,27 @@ async fn run_terminal_relay(relay: RelayConnection) -> Result<()> {
     Ok(())
 }
 
-async fn send_terminal_output(frame: Message, relay: &mut Option<mpsc::Sender<Message>>) {
+fn send_terminal_output(frame: Message, relay: &mut Option<mpsc::Sender<Message>>) -> bool {
     if let Some(tx) = relay.as_ref()
-        && tx.send(frame).await.is_err()
+        && tx.try_send(frame).is_err()
     {
         *relay = None;
+        return false;
     }
+    true
 }
 
-async fn send_terminal_snapshot(snapshot: Vec<u8>, relay: &mut Option<mpsc::Sender<Message>>) {
+fn send_terminal_snapshot(snapshot: Vec<u8>, relay: &mut Option<mpsc::Sender<Message>>) -> bool {
     let Some(tx) = relay.as_ref() else {
-        return;
+        return true;
     };
     for chunk in snapshot.chunks(MAX_TERMINAL_FRAME_BYTES) {
-        if tx
+        if tx.try_send(Message::Binary(chunk.to_vec().into())).is_err() {
-            .send(Message::Binary(chunk.to_vec().into()))
-            .await
-            .is_err()
-        {
             *relay = None;
-            return;
+            return false;
         }
     }
+    true
 }
 
 #[cfg(not(unix))]
@@ -618,7 +596,7 @@ mod tests {
             active: Arc::new(Mutex::new(HashMap::from([(
                 terminal_id.to_string(),
                 ActiveTerminal {
-                    cancel,
+                    cancel: Some(cancel),
                     relay_credentials,
                     created_at_unix: 42,
                 },
@@ -639,7 +617,7 @@ mod tests {
     }
 
     #[test]
-    fn failed_resume_removes_stopped_terminal_from_inventory() {
+    fn failed_resume_hides_stopped_terminal_from_inventory() {
         let manager = manager_with_stopped_terminal("stopped-terminal");
         let terminal_id = TerminalId::new("stopped-terminal").expect("terminal id");
 
@@ -654,6 +632,68 @@ mod tests {
         assert!(manager.sessions().is_empty());
     }
 
+    #[test]
+    fn close_succeeds_when_worker_already_dropped_cancel_receiver() {
+        let manager = manager_with_stopped_terminal("stopped-terminal");
+        let terminal_id = TerminalId::new("stopped-terminal").expect("terminal id");
+
+        assert!(manager.close(&terminal_id));
+        assert!(manager.sessions().is_empty());
+    }
+
+    #[test]
+    fn close_keeps_session_registered_until_worker_completion() {
+        let (cancel, mut cancel_rx) = oneshot::channel();
+        let (relay_credentials, _relay_rx) = mpsc::unbounded_channel();
+        let (events, _event_rx) = mpsc::unbounded_channel();
+        let active = Arc::new(Mutex::new(HashMap::from([(
+            "closing-terminal".to_string(),
+            ActiveTerminal {
+                cancel: Some(cancel),
+                relay_credentials,
+                created_at_unix: 42,
+            },
+        )])));
+        let manager = TerminalManager {
+            active: active.clone(),
+            max_sessions: 2,
+            events,
+        };
+        let terminal_id = TerminalId::new("closing-terminal").expect("terminal id");
+
+        assert!(manager.close(&terminal_id));
+        assert!(cancel_rx.try_recv().is_ok());
+        assert!(
+            active
+                .lock()
+                .expect("terminal manager")
+                .contains_key("closing-terminal")
+        );
+        assert!(manager.sessions().is_empty());
+
+        remove_completed(&Arc::downgrade(&active), terminal_id.as_str());
+        assert!(
+            !active
+                .lock()
+                .expect("terminal manager")
+                .contains_key("closing-terminal")
+        );
+    }
+
+    #[test]
+    fn open_rejects_invalid_initial_size_before_spawning() {
+        let mut config = crate::config::DEFAULT_CONFIG.clone();
+        config.terminal.enabled = true;
+        let (manager, _events) = TerminalManager::new(&config);
+        let terminal_id = TerminalId::new("invalid-size").expect("terminal id");
+
+        let error = manager
+            .open(&config, terminal_id, "relay".into(), 0, 80)
+            .expect_err("invalid rows must fail");
+        assert!(error.to_string().contains("outside supported bounds"));
+        assert!(manager.active.lock().expect("terminal manager").is_empty());
+    }
+
     #[test]
     fn snapshot_reconstructs_screen_content_and_cursor() {
         let mut state = TerminalState::new(24, 80);
@@ -737,13 +777,62 @@ mod tests {
         assert_eq!(restored.screen().contents(), "full-screen");
     }
 
+    #[test]
+    fn alternate_screen_snapshot_restores_hidden_primary_on_exit() {
+        let mut state = TerminalState::new(3, 20);
+        state.process(b"\x1b[31mshell history\r\n$ btop\x1b[?1049h\x1b[34mbtop dashboard");
+
+        let mut restored = vt100::Parser::new(3, 20, TERMINAL_SCROLLBACK_ROWS);
+        restored.process(&state.snapshot());
+
+        assert!(restored.screen().alternate_screen());
+        assert_eq!(restored.screen().contents(), "btop dashboard");
+        assert_eq!(restored.screen().fgcolor(), vt100::Color::Idx(4));
+
+        // Compare the reattached terminal with the original parser after both
+        // receive the application's real alternate-screen exit sequence.
+        state.process(b"\x1b[?1049l");
+        restored.process(b"\x1b[?1049l");
+        assert_eq!(
+            restored.screen().contents(),
+            state.parser.screen().contents()
+        );
+        assert_eq!(
+            restored.screen().cursor_position(),
+            state.parser.screen().cursor_position()
+        );
+        assert_eq!(restored.screen().fgcolor(), vt100::Color::Idx(1));
+
+        restored.screen_mut().set_scrollback(usize::MAX);
+        assert!(restored.screen().contents().contains("shell history"));
+        assert!(!restored.screen().contents().contains("btop dashboard"));
+    }
+
+    #[test]
+    fn snapshot_does_not_change_the_source_viewport_or_state() {
+        let mut state = TerminalState::new(3, 20);
+        for line in 0..10 {
+            state.process(format!("history {line}\r\n").as_bytes());
+        }
+        state.parser.screen_mut().set_scrollback(4);
+
+        let before_contents = state.parser.screen().contents();
+        let before_state = state.parser.screen().state_formatted();
+        let before_scrollback = state.parser.screen().scrollback();
+        let _ = state.snapshot();
+
+        assert_eq!(state.parser.screen().contents(), before_contents);
+        assert_eq!(state.parser.screen().state_formatted(), before_state);
+        assert_eq!(state.parser.screen().scrollback(), before_scrollback);
+    }
+
     #[tokio::test]
     async fn snapshot_is_split_at_the_terminal_frame_limit() {
         let snapshot = vec![7; MAX_TERMINAL_FRAME_BYTES * 2 + 1];
         let (tx, mut rx) = mpsc::channel(3);
         let mut relay = Some(tx);
 
-        send_terminal_snapshot(snapshot.clone(), &mut relay).await;
+        assert!(send_terminal_snapshot(snapshot.clone(), &mut relay));
 
         let mut restored = Vec::new();
         for _ in 0..3 {
@@ -755,4 +844,22 @@ mod tests {
         }
         assert_eq!(restored, snapshot);
     }
+
+    #[test]
+    fn saturated_relay_output_detaches_without_waiting() {
+        let (tx, mut rx) = mpsc::channel(1);
+        tx.try_send(Message::Binary(vec![1].into()))
+            .expect("prefill relay queue");
+        let mut relay = Some(tx);
+
+        assert!(!send_terminal_output(
+            Message::Binary(vec![2].into()),
+            &mut relay,
+        ));
+        assert!(relay.is_none());
+        assert_eq!(
+            rx.try_recv().expect("original queued frame").into_data(),
+            vec![1]
+        );
+    }
 }

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

@@ -191,21 +191,10 @@ pub async fn close_terminal(
     State(state): State<AppState>,
     Path(terminal_id): Path<String>,
 ) -> Result<StatusCode, ApiError> {
-    if let Some(agent_id) = close_registered_terminal(&state, &terminal_id).await {
+    if let Some(agent_id) =
+        close_registered_terminal(&state, &terminal_id, TerminalCloseReason::HttpDelete).await
+    {
         info!(terminal_id, agent_id, "terminal session closed by operator");
-        append_terminal_audit(
-            &state,
-            &agent_id,
-            &terminal_id,
-            TerminalAudit {
-                actor_type: "admin_api",
-                event_type: "terminal_close",
-                outcome: "ok",
-                message: "terminal session closed by operator",
-                metadata: serde_json::json!({}),
-            },
-        )
-        .await;
     } else {
         if !state.terminals.was_closed(&terminal_id).await {
             return Err(terminal_not_found(&terminal_id));
@@ -409,7 +398,7 @@ async fn handle_operator_terminal_socket(
     }
 
     if explicit_close {
-        close_registered_terminal(&state, &terminal_id).await;
+        close_registered_terminal(&state, &terminal_id, TerminalCloseReason::BrowserClose).await;
     } else {
         state.terminals.detach_operator(&terminal_id, &auth.0).await;
     }
@@ -510,7 +499,31 @@ where
     write.send(message).await
 }
 
-async fn close_registered_terminal(state: &AppState, terminal_id: &str) -> Option<String> {
+#[derive(Clone, Copy)]
+enum TerminalCloseReason {
+    HttpDelete,
+    BrowserClose,
+    AbsoluteTimeout,
+}
+
+impl TerminalCloseReason {
+    fn as_str(self) -> &'static str {
+        match self {
+            Self::HttpDelete => "http_delete",
+            Self::BrowserClose => "browser_close",
+            Self::AbsoluteTimeout => "absolute_timeout",
+        }
+    }
+}
+
+/// Removes and audits a terminal as one logical operation. Only the caller
+/// that actually removes the session records the close, keeping retries and
+/// simultaneous timeout/operator closes from producing duplicate events.
+async fn close_registered_terminal(
+    state: &AppState,
+    terminal_id: &str,
+    reason: TerminalCloseReason,
+) -> Option<String> {
     let agent_id = state.terminals.remove(terminal_id).await?;
     if let Some(session) = state.sessions.read().await.get(&agent_id) {
         let terminal_id = TerminalId::new(terminal_id.to_string()).ok()?;
@@ -520,15 +533,32 @@ async fn close_registered_terminal(state: &AppState, terminal_id: &str) -> Optio
                 terminal_id,
             }));
     }
+    append_terminal_audit(
+        state,
+        &agent_id,
+        terminal_id,
+        TerminalAudit {
+            actor_type: "admin_api",
+            event_type: "terminal_close",
+            outcome: "ok",
+            message: "terminal session closed",
+            metadata: serde_json::json!({ "reason": reason.as_str() }),
+        },
+    )
+    .await;
     Some(agent_id)
 }
 
 fn spawn_absolute_timeout(state: AppState, terminal_id: TerminalId, agent_id: String) {
     tokio::spawn(async move {
         tokio::time::sleep(TERMINAL_ABSOLUTE_TIMEOUT).await;
-        if close_registered_terminal(&state, terminal_id.as_str())
+        if close_registered_terminal(
-            .await
+            &state,
-            .is_some()
+            terminal_id.as_str(),
+            TerminalCloseReason::AbsoluteTimeout,
+        )
+        .await
+        .is_some()
         {
             info!(terminal_id = %terminal_id, agent_id, "terminal absolute timeout reached");
         }

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

@@ -77,6 +77,7 @@ impl TerminalRegistry {
 
     pub async fn create(&self, agent_id: String) -> Result<CreatedTerminal, &'static str> {
         let mut sessions = self.inner.lock().await;
+        prune_expired_sessions(&mut sessions);
         if sessions
             .values()
             .filter(|session| session.agent_id == agent_id)
@@ -168,10 +169,12 @@ impl TerminalRegistry {
     ) -> Vec<(TerminalId, String)> {
         let reported_ids = reported
             .iter()
+            .filter(|session| expires_at_for_created_at(session.created_at_unix).is_some())
             .map(|session| session.terminal_id.as_str())
             .collect::<std::collections::HashSet<_>>();
         let stale_ids = {
-            let sessions = self.inner.lock().await;
+            let mut sessions = self.inner.lock().await;
+            prune_expired_sessions(&mut sessions);
             sessions
                 .iter()
                 .filter(|(terminal_id, session)| {
@@ -189,13 +192,17 @@ impl TerminalRegistry {
         let mut credentials = Vec::new();
         let mut sessions = self.inner.lock().await;
         for reported_session in reported {
+            let Some(expires_at) = expires_at_for_created_at(reported_session.created_at_unix)
+            else {
+                continue;
+            };
             let terminal_id = reported_session.terminal_id.as_str().to_string();
             let session = sessions
                 .entry(terminal_id)
                 .or_insert_with(|| TerminalSession {
                     agent_id: agent_id.to_string(),
                     created_at_unix: reported_session.created_at_unix,
-                    expires_at: Instant::now() + TERMINAL_ABSOLUTE_TIMEOUT,
+                    expires_at,
                     agent_confirmed: true,
                     relay_token: None,
                     attachment_token: None,
@@ -319,31 +326,30 @@ impl TerminalRegistry {
         terminal_id: &str,
         frame: TerminalRelayFrame,
     ) -> Result<(), &'static str> {
-        let tx = self
+        // Select the attached transport or enqueue the frame while holding one
-            .inner
+        // lock. Otherwise an agent can attach between those decisions and
-            .lock()
+        // leave input stranded in the pre-attachment queue.
-            .await
+        let tx = {
-            .get(terminal_id)
+            let mut sessions = self.inner.lock().await;
-            .and_then(|session| session.agent_tx.clone());
+            let session = active_session(&mut sessions, terminal_id)?;
-        if let Some(tx) = tx {
+            if let Some(tx) = session.agent_tx.clone() {
-            return tx
+                tx
-                .send(frame)
-                .await
-                .map_err(|_| "terminal_agent_disconnected");
-        }
-
-        let mut sessions = self.inner.lock().await;
-        let session = active_session(&mut sessions, terminal_id)?;
-        session.pending_agent_bytes += relay_frame_size(&frame);
-        session.pending_agent.push_back(frame);
-        while session.pending_agent_bytes > TERMINAL_PENDING_AGENT_BYTES {
-            if let Some(dropped) = session.pending_agent.pop_front() {
-                session.pending_agent_bytes -= relay_frame_size(&dropped);
             } else {
-                break;
+                session.pending_agent_bytes += relay_frame_size(&frame);
+                session.pending_agent.push_back(frame);
+                while session.pending_agent_bytes > TERMINAL_PENDING_AGENT_BYTES {
+                    if let Some(dropped) = session.pending_agent.pop_front() {
+                        session.pending_agent_bytes -= relay_frame_size(&dropped);
+                    } else {
+                        break;
+                    }
+                }
+                return Ok(());
             }
-        }
+        };
-        Ok(())
+        tx.send(frame)
+            .await
+            .map_err(|_| "terminal_agent_disconnected")
     }
 
     pub async fn relay_from_agent(
@@ -417,7 +423,9 @@ impl TerminalRegistry {
     }
 
     pub async fn summary(&self, terminal_id: &str) -> Option<(String, u64, bool, bool)> {
-        self.inner.lock().await.get(terminal_id).map(|session| {
+        let mut sessions = self.inner.lock().await;
+        prune_expired_sessions(&mut sessions);
+        sessions.get(terminal_id).map(|session| {
             (
                 session.agent_id.clone(),
                 session.created_at_unix,
@@ -428,7 +436,8 @@ impl TerminalRegistry {
     }
 
     pub async fn summaries(&self) -> Vec<TerminalSummary> {
-        let sessions = self.inner.lock().await;
+        let mut sessions = self.inner.lock().await;
+        prune_expired_sessions(&mut sessions);
         let mut summaries = sessions
             .iter()
             .map(|(terminal_id, session)| TerminalSummary {
@@ -448,11 +457,14 @@ fn active_session<'a>(
     sessions: &'a mut HashMap<String, TerminalSession>,
     terminal_id: &str,
 ) -> Result<&'a mut TerminalSession, &'static str> {
-    let session = sessions.get_mut(terminal_id).ok_or("terminal_not_found")?;
+    if sessions
-    if session.expires_at <= Instant::now() {
+        .get(terminal_id)
+        .is_some_and(|session| session.expires_at <= Instant::now())
+    {
+        sessions.remove(terminal_id);
         return Err("terminal_expired");
     }
-    Ok(session)
+    sessions.get_mut(terminal_id).ok_or("terminal_not_found")
 }
 
 fn validate_operator_id(operator_id: &str) -> Result<(), &'static str> {
@@ -485,6 +497,26 @@ fn prune_tombstones(closed: &mut HashMap<String, Instant>) {
     closed.retain(|_, closed_at| closed_at.elapsed() < TERMINAL_TOMBSTONE_TTL);
 }
 
+fn prune_expired_sessions(sessions: &mut HashMap<String, TerminalSession>) {
+    let now = Instant::now();
+    sessions.retain(|_, session| session.expires_at > now);
+}
+
+/// Converts the agent's durable creation timestamp into CC's monotonic
+/// deadline. Reconciliation must not grant an old PTY a fresh twelve hours.
+fn expires_at_for_created_at(created_at_unix: u64) -> Option<Instant> {
+    let now_unix = SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .unwrap_or_default()
+        .as_secs();
+    let age = Duration::from_secs(now_unix.saturating_sub(created_at_unix));
+    if age >= TERMINAL_ABSOLUTE_TIMEOUT {
+        return None;
+    }
+    let remaining = TERMINAL_ABSOLUTE_TIMEOUT - age;
+    Some(Instant::now() + remaining)
+}
+
 fn relay_frame_size(frame: &TerminalRelayFrame) -> usize {
     match frame {
         TerminalRelayFrame::Binary(bytes) => bytes.len(),
@@ -627,9 +659,10 @@ mod tests {
     async fn agent_inventory_adopts_and_reconnects_live_session() {
         let registry = TerminalRegistry::new();
         let terminal_id = TerminalId::new("survived-cc").expect("terminal id");
+        let created_at_unix = u64::MAX;
         let reported = AgentTerminalSession {
             terminal_id: terminal_id.clone(),
-            created_at_unix: 42,
+            created_at_unix,
         };
 
         let credentials = registry
@@ -647,10 +680,48 @@ mod tests {
             .await
             .expect("adopted summary");
         assert_eq!(summary.0, "router");
-        assert_eq!(summary.1, 42);
+        assert_eq!(summary.1, created_at_unix);
         assert!(summary.2);
     }
 
+    #[tokio::test]
+    async fn expired_sessions_release_agent_quota() {
+        let registry = TerminalRegistry::new();
+        let first = registry.create("router".into()).await.expect("first");
+        registry.create("router".into()).await.expect("second");
+        registry
+            .inner
+            .lock()
+            .await
+            .get_mut(&first.terminal_id)
+            .expect("first session")
+            .expires_at = Instant::now();
+
+        registry
+            .create("router".into())
+            .await
+            .expect("expired session no longer consumes quota");
+        assert!(registry.summary(&first.terminal_id).await.is_none());
+    }
+
+    #[tokio::test]
+    async fn reconciliation_does_not_revive_expired_agent_session() {
+        let registry = TerminalRegistry::new();
+        let terminal_id = TerminalId::new("expired-agent-session").expect("terminal id");
+        let reported = AgentTerminalSession {
+            terminal_id: terminal_id.clone(),
+            created_at_unix: 0,
+        };
+
+        assert!(
+            registry
+                .reconcile_agent_sessions("router", &[reported])
+                .await
+                .is_empty()
+        );
+        assert!(registry.summary(terminal_id.as_str()).await.is_none());
+    }
+
     #[tokio::test]
     async fn operator_detach_does_not_remove_session() {
         let registry = TerminalRegistry::new();

wakey-control-plane/src/ws.rs

@@ -56,6 +56,7 @@ enum IncomingClientMessage {
 #[derive(Default)]
 struct AgentConnectionState {
     authed_agent_id: Option<String>,
+    hello_agent_id: Option<String>,
     hello_at: Option<Instant>,
     capabilities: Vec<AgentCapability>,
 }
@@ -193,18 +194,27 @@ async fn process_agent_text(
             agent_id,
             capabilities,
         } => {
+            if connection
+                .hello_agent_id
+                .as_deref()
+                .is_some_and(|hello_agent_id| hello_agent_id != agent_id)
+            {
+                anyhow::bail!("hello changed agent identity");
+            }
             let now = Instant::now();
             if connection.hello_at.is_none() {
                 connection.hello_at = Some(now);
             }
             let connect_to_hello_ms = connected_at.elapsed().as_millis() as u64;
             info!(agent_id = %agent_id, connect_to_hello_ms, "agent hello received");
+            connection.hello_agent_id = Some(agent_id);
             connection.capabilities = capabilities;
         }
         IncomingClientMessage::Auth {
             agent_id,
             agent_token,
         } => {
+            validate_auth_identity(connection, &agent_id)?;
             let connect_to_auth_ms = connected_at.elapsed().as_millis() as u64;
             let hello_to_auth_ms = connection
                 .hello_at
@@ -372,6 +382,17 @@ fn now_duration_ms(duration: std::time::Duration) -> u64 {
     duration.as_millis() as u64
 }
 
+fn validate_auth_identity(connection: &AgentConnectionState, agent_id: &str) -> Result<()> {
+    let hello_agent_id = connection
+        .hello_agent_id
+        .as_deref()
+        .ok_or_else(|| anyhow::anyhow!("auth before hello"))?;
+    if hello_agent_id != agent_id {
+        anyhow::bail!("auth agent does not match hello agent");
+    }
+    Ok(())
+}
+
 async fn ensure_current_session(
     state: &AppState,
     agent_id: &str,
@@ -406,7 +427,7 @@ mod tests {
 
     use crate::runtime::AgentSession;
 
-    use super::is_current_session;
+    use super::{AgentConnectionState, is_current_session, validate_auth_identity};
 
     #[test]
     fn current_session_check_rejects_stale_connection_ids() {
@@ -425,4 +446,14 @@ mod tests {
         assert!(!is_current_session(&sessions, "agent-a", "conn-old"));
         assert!(!is_current_session(&sessions, "agent-b", "conn-new"));
     }
+
+    #[test]
+    fn auth_requires_a_matching_hello_identity() {
+        let mut connection = AgentConnectionState::default();
+        assert!(validate_auth_identity(&connection, "agent-a").is_err());
+
+        connection.hello_agent_id = Some("agent-a".into());
+        assert!(validate_auth_identity(&connection, "agent-a").is_ok());
+        assert!(validate_auth_identity(&connection, "agent-b").is_err());
+    }
 }

wakey-linux/src/terminal.rs

@@ -120,11 +120,16 @@ mod tests {
             .into_parts();
 
         writer.resize(31, 101).expect("resize owned PTY writer");
+        // Print the size from the parent shell after `stty` exits, so observing
+        // the marker also means foreground control has returned to the shell.
+        // Then wait for the literal probe; WINCH can interrupt a plain `read`.
         writer
             .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",
+                  size=$(stty size); printf 'WAKEY_SIZE:%s\\n' \"$size\"; \
+                  line=; while [ \"$line\" != probe ]; do read line || line=; done; \
+                  printf 'WAKEY_PTY_OK:%s\\n' \"$line\"; exit 0\n",
             )
             .await
             .expect("write shell input");
@@ -135,6 +140,7 @@ mod tests {
             "WAKEY_ENV:xterm-256color:truecolor",
         )
         .await;
+        read_until(&mut reader, &mut output, "WAKEY_SIZE:31 101").await;
         writer.refresh().expect("signal foreground process group");
         writer
             .write_all(b"probe\n")
@@ -157,6 +163,10 @@ mod tests {
             output.contains("WAKEY_PTY_OK:probe"),
             "unexpected PTY output: {output:?}"
         );
+        assert!(
+            output.contains("WAKEY_SIZE:31 101"),
+            "PTY did not report resized dimensions: {output:?}"
+        );
         assert!(
             output.contains("WAKEY_WINCH"),
             "foreground process did not receive SIGWINCH: {output:?}"