twitchyliquid64 · Jun 18, 2026 · Jun 17, 2026 · coderabbitai · Jun 17, 2026
@@ -211,6 +211,25 @@ impl OneshotSshRpc for RenameSession {
     type Response = Errorable<RenameSessionResponse>;
 }
 
+/// An RPC to destroy an existing session.
+pub struct DestroySession;
+
+/// The request for a [`DestroySession`] RPC.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct DestroySessionRequest {
+    pub id: SessionId,
+}
+
+/// The response for a [`DestroySession`] RPC.
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+pub struct DestroySessionResponse;
+
+impl OneshotSshRpc for DestroySession {
+    const NAME: &'static str = constcat::concat!(RPC_SUBSYSTEM_PREFIX, "DestroySession");
+    type Request<'a> = DestroySessionRequest;
+    type Response = Errorable<DestroySessionResponse>;
+}
+
 // ---------------------------------------------------------------------------
 // Session-creation flow (multi-round contribution composition).
 //

@@ -1,8 +1,8 @@
 use minimald_rpc::{
-    CreateSession, CreateSessionResponse, Errorable, GetSessionRecord, GetSessionRecordRequest,
-    GetSessionRecordResponse, GetVersion, GetVersionResponse, ListSessions, ListSessionsEntry,
-    ListSessionsResponse, OneshotSshRpc, RPC_SUBSYSTEM_PREFIX, RenameSession,
-    RenameSessionResponse,
+    CreateSession, CreateSessionResponse, DestroySession, DestroySessionResponse, Errorable,
+    GetSessionRecord, GetSessionRecordRequest, GetSessionRecordResponse, GetVersion,
+    GetVersionResponse, ListSessions, ListSessionsEntry, ListSessionsResponse, OneshotSshRpc,
+    RPC_SUBSYSTEM_PREFIX, RenameSession, RenameSessionResponse,
 };
 use russh::{
     Channel as RuChannel, ChannelId,
@@ -173,6 +173,23 @@ async fn serve_rename_session(s: ServerStateHandle, c: RuChannel<Msg>) {
     }
 }
 
+async fn serve_destroy_session(s: ServerStateHandle, c: RuChannel<Msg>) {
+    let res = DestroySession
+        .handle_channel(c, async |req| {
+            let res = s.sessions_manager().await.destroy_session(req.id).await;
+            match res {
+                Ok(()) => Ok(Errorable::Ok(DestroySessionResponse)),
+                Err(e) => Ok(Errorable::Err {
+                    error: e.to_string(),
+                }),
+            }
+        })
+        .await;
+    if let Err(e) = res {
+        tracing::warn!("RPC handler for {} failed: {}", DestroySession::NAME, e);
+    }
+}
+
 pub(crate) const STREAM_WORKSPACE_FILES: &str =
     constcat::concat!(RPC_SUBSYSTEM_PREFIX, "WorkspaceFilesTarZst");
 
@@ -246,6 +263,7 @@ pub async fn handle_ssh_rpc(
         | GetSessionRecord::NAME
         | CreateSession::NAME
         | RenameSession::NAME
+        | DestroySession::NAME
         | STREAM_WORKSPACE_FILES => {
             let mut conn_lock = c.lock().await;
             let c_hnd = match conn_lock.take(id) {
@@ -276,6 +294,7 @@ pub async fn handle_ssh_rpc(
         GetSessionRecord::NAME => spawn(serve_get_session_record(s, channel)),
         CreateSession::NAME => spawn(serve_create_session(s, channel)),
         RenameSession::NAME => spawn(serve_rename_session(s, channel)),
+        DestroySession::NAME => spawn(serve_destroy_session(s, channel)),
         STREAM_WORKSPACE_FILES => spawn(serve_stream_workspace_files(s, config, channel)),
         _ => unreachable!(),
     };
@@ -285,7 +304,9 @@ pub async fn handle_ssh_rpc(
 
 #[cfg(test)]
 mod tests {
-    use minimald_rpc::{CreateSession, CreateSessionRequest, RenameSessionRequest};
+    use minimald_rpc::{
+        CreateSession, CreateSessionRequest, DestroySessionRequest, RenameSessionRequest,
+    };
     use paths::HostAbsPath;
     use sessions::SessionId;
     use tokio::io::{AsyncReadExt, AsyncWriteExt};
@@ -612,4 +633,47 @@ mod tests {
             "expected an unknown-id error",
         );
     }
+
+    #[tokio::test]
+    async fn destroy_session_removes_it_from_get_and_list() {
+        let server = TestServer::new().await;
+        let mut client = server.connect().await;
+        let session_id = fresh_session(&mut client).await;
+
+        let resp = client
+            .call::<DestroySession>(&DestroySessionRequest { id: session_id })
+            .await;
+        assert_eq!(resp, Errorable::Ok(DestroySessionResponse));
+
+        // The record is gone: it no longer resolves by id...
+        let get_session = client
+            .call::<GetSessionRecord>(&GetSessionRecordRequest::Id(session_id))
+            .await;
+        assert!(get_session.record.is_none());
+
+        // ...and it's dropped from the listing.
+        let list_sessions = client.call::<ListSessions>(&()).await;
+        assert!(
+            list_sessions.sessions.is_empty(),
+            "destroyed session should not be listed, got {:?}",
+            list_sessions.sessions,
+        );
+    }
+
+    #[tokio::test]
+    async fn destroy_session_errors_for_unknown_id() {
+        let server = TestServer::new().await;
+        let mut client = server.connect().await;
+
+        let resp = client
+            .call::<DestroySession>(&DestroySessionRequest {
+                id: SessionId::nil(),
+            })
+            .await;
+
+        assert!(
+            matches!(resp, Errorable::Err { error } if error.contains("no session with ID")),
+            "expected an unknown-id error",
+        );
+    }
 }
@@ -7,7 +7,9 @@ use paths::DaemonAbsPath;
 use russh::{Channel, server::Msg};
 use sessions::{Record, store::SessionObject};
 use std::fmt::{self};
+use std::ops::ControlFlow;
 use tokio::sync::{mpsc, oneshot};
+use tokio::task::JoinHandle;
 
 /// An error that occurred when attaching to a running session/its-shell.
 #[derive(Debug)]
@@ -49,6 +51,7 @@ enum SessionMessage {
     ),
     GetHostAttrs(oneshot::Sender<Option<HostAttrs>>),
     RefreshRecord(Record),
+    Destroy(oneshot::Sender<()>),
     #[cfg(test)]
     GetRecord(oneshot::Sender<Record>),
 }
@@ -63,7 +66,12 @@ pub struct Session<S: SessionObject> {
     minimal_cache_dir: DaemonAbsPath,
     session: S,
 
-    host: Option<session_host::HostHandle>,
+    /// The running host, if minted, paired with the `JoinHandle` of its runtime
+    /// loop so teardown can be awaited on destroy.
+    host: Option<(
+        session_host::HostHandle,
+        JoinHandle<Result<i32, std::io::Error>>,
+    )>,
 }
 
 impl<S: SessionObject> Session<S> {
@@ -94,12 +102,17 @@ impl<S: SessionObject> Session<S> {
     /// session.
     async fn mainloop(mut self) {
         while let Some(msg) = self.receiver.recv().await {
-                self.handle_message(msg).await;
+            if self.handle_message(msg).await.is_break() {
+                break;
+            }
         }
     }
 
     /// Handles a specific message recieved by the session.
-    async fn handle_message(&mut self, msg: SessionMessage) {
+    ///
+    /// Returns [`ControlFlow::Break`] when the session has been destroyed and
+    /// the actor loop should exit.
+    async fn handle_message(&mut self, msg: SessionMessage) -> ControlFlow<()> {
         match msg {
             SessionMessage::GetPaths(r) => {
                 let _ = r.send(self.paths());
@@ -116,17 +129,35 @@ impl<S: SessionObject> Session<S> {
             SessionMessage::GetHostAttrs(r) => {
                 let _ = r.send(match self.host.as_ref() {
                     None => None,
-                    Some(h) => h.get_attrs().await.ok(),
+                    Some((h, _)) => h.get_attrs().await.ok(),
                 });
             }
             SessionMessage::RefreshRecord(r) => {
                 self.session.refresh_from_record(r);
             }
+            SessionMessage::Destroy(r) => {
+                self.destroy().await;
+                let _ = r.send(());
+                return ControlFlow::Break(());
+            }
             #[cfg(test)]
             SessionMessage::GetRecord(r) => {
                 let _ = r.send(self.session.record().clone());
             }
         }
+        ControlFlow::Continue(())
+    }
+
+    /// Tears down the running host, if any, waiting for the process to be reaped
+    /// and the sandbox guard to be dropped before returning.
+    async fn destroy(&mut self) {
+        if let Some((host, task)) = self.host.take() {
+            // Signal the process to die, then await the runtime loop so the
+            // sandbox files backing its rootfs are released before the caller
+            // removes the session's directory tree.
+            let _ = host.kill().await;
+            let _ = task.await;
+        }
     }
 
     async fn attach(
@@ -146,7 +177,7 @@ impl<S: SessionObject> Session<S> {
                 self.mint_session_host(session_hnd, conn_username, channel, sz)
                     .await
             }
-            Some(h) => {
+            Some((h, _)) => {
                 match h.attach(channel, sz).await {
                     Ok(()) => Ok(()),
                     Err((channel, sz)) => {
@@ -286,6 +317,16 @@ impl SessionHandle {
             .expect("corresponding session is dead");
     }
 
+    /// Tears down the session: kills its host (if any), waits for teardown, and
+    /// stops the actor. The handle is dead once this returns.
+    pub(crate) async fn destroy(&self) {
+        let (send, recv) = oneshot::channel();
+        // Ignore send errors - the recv will also fail.
+        let _ = self.0.send(SessionMessage::Destroy(send)).await;
+        // If the actor died before acking, the session is gone all the same.
+        let _ = recv.await;
+    }
+
     pub async fn attach(
         &self,
         conn_username: String,