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
2026-07-16 08:51:55 +07:00
parent 536b50220f
commit 3476892cd1
10 changed files with 408 additions and 115 deletions
@@ -1,4 +1,5 @@
 #!/usr/bin/env bash
+set -euo pipefail

 # sometimes it breaks, run to fix

@@ -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 {
-      setSessions(remaining);
-      setSessionTitles((current) => {
-        const next = { ...current };
-        delete next[closingId];
-        return next;
-      });
-      if (
-        window.sessionStorage.getItem(REMEMBERED_TERMINAL_KEY) === closingId
-      ) {
-        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);
-      }
+      void listTerminals()
+        .then((listed) =>
+          setSessions((current) => reconcileTerminalSessions(current, listed)),
+        )
+        .catch(() => undefined);
+      return;
+    }
+
+    setSessions((current) =>
+      current.filter((item) => item.terminal_id !== closingId),
+    );
+    setSessionTitles((current) => {
+      const next = { ...current };
+      delete next[closingId];
+      return next;
+    });
+    if (
+      window.sessionStorage.getItem(REMEMBERED_TERMINAL_KEY) === closingId
+    ) {
+      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);
    }
  }

@@ -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 {
@@ -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 {
@@ -14,6 +14,8 @@ use tracing::{info, warn};

 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.
@@ -47,7 +49,7 @@ impl TerminalState {
 }

 struct ActiveTerminal {
-    cancel: oneshot::Sender<()>,
+    cancel: Option<oneshot::Sender<()>>,
    relay_credentials: mpsc::UnboundedSender<String>,
    created_at_unix: u64,
 }
@@ -88,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();
@@ -107,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,
                },
@@ -135,30 +138,38 @@ impl TerminalManager {
    }

    pub fn close(&self, terminal_id: &TerminalId) -> bool {
-        self.active
-            .lock()
-            .expect("terminal manager poisoned")
-            .remove(terminal_id.as_str())
-            .is_some_and(|active| active.cancel.send(()).is_ok())
+        let mut active = self.active.lock().expect("terminal manager poisoned");
+        let Some(terminal) = active.get_mut(terminal_id.as_str()) else {
+            return false;
+        };
+        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");
-        active.retain(|_, terminal| !terminal.relay_credentials.is_closed());
+        let active = self.active.lock().expect("terminal manager poisoned");
        active
            .iter()
+            .filter(|(_, terminal)| {
+                terminal.cancel.is_some() && !terminal.relay_credentials.is_closed()
+            })
            .filter_map(|(terminal_id, active)| {
                TerminalId::new(terminal_id.clone())
                    .ok()
@@ -177,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(());
+                }
            }
        }
    }
@@ -214,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;
@@ -222,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;
@@ -229,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() => {
@@ -248,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");
                        }
@@ -290,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) => {
@@ -298,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,
@@ -320,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");
@@ -360,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)]
@@ -385,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;
@@ -398,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")?
@@ -408,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"),
@@ -441,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
-            .send(Message::Binary(chunk.to_vec().into()))
-            .await
-            .is_err()
-        {
+        if tx.try_send(Message::Binary(chunk.to_vec().into())).is_err() {
            *relay = None;
-            return;
+            return false;
        }
    }
+    true
 }

 #[cfg(not(unix))]
@@ -564,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,
                },
@@ -585,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");

@@ -600,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);
@@ -738,7 +832,7 @@ mod tests {
        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 {
@@ -750,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]
+        );
+    }
 }
@@ -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())
-            .await
-            .is_some()
+        if close_registered_terminal(
+            &state,
+            terminal_id.as_str(),
+            TerminalCloseReason::AbsoluteTimeout,
+        )
+        .await
+        .is_some()
        {
            info!(terminal_id = %terminal_id, agent_id, "terminal absolute timeout reached");
        }
@@ -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
-            .inner
-            .lock()
-            .await
-            .get(terminal_id)
-            .and_then(|session| session.agent_tx.clone());
-        if let Some(tx) = tx {
-            return 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);
+        // Select the attached transport or enqueue the frame while holding one
+        // lock. Otherwise an agent can attach between those decisions and
+        // leave input stranded in the pre-attachment queue.
+        let tx = {
+            let mut sessions = self.inner.lock().await;
+            let session = active_session(&mut sessions, terminal_id)?;
+            if let Some(tx) = session.agent_tx.clone() {
+                tx
            } 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 session.expires_at <= Instant::now() {
+    if sessions
+        .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();
@@ -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());
+    }
 }
@@ -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:?}"