fix(supervisor): terminate orphaned non-main processes when ssh connection drops (#4946)

crates/openshell-supervisor-process/src/ssh.rs

@@ -430,18 +430,29 @@ struct SshHandler {
 
 impl Drop for SshHandler {
     fn drop(&mut self) {
-        let Some(main_session) = self.main_session.as_ref() else {
-            return;
-        };
-        for state in self.channels.values_mut() {
-            if state.main_attached {
-                main_session.end_terminal_attachment();
-            }
-            if let Some(owner) = state.main_input_owner.take() {
-                main_session.release_input(owner);
+        if let Some(main_session) = self.main_session.as_ref() {
+            for state in self.channels.values_mut() {
+                if state.main_attached {
+                    main_session.end_terminal_attachment();
+                    if let Some(owner) = state.main_input_owner.take() {
+                        main_session.release_input(owner);
+                    }
+                    if let Some(task) = state.main_output_task.take() {
+                        task.abort();
+                    }
+                }
             }
-            if let Some(task) = state.main_output_task.take() {
-                task.abort();
+        }
+        for state in self.channels.values_mut() {
+            state.input_task.take();
+            if !state.main_attached
+                && let Some(process) = state.process.take()
+            {
+                tokio::spawn(async move {
+                    if let Err(error) = process.terminate().await {
+                        warn!(%error, "failed to terminate abandoned exec on ssh disconnect");
+                    }
+                });
             }
         }
     }

crates/openshell-supervisor-process/src/ssh/exec_input_tests.rs

@@ -546,3 +546,32 @@ async fn overflow_reports_failure_without_output_window_credit() {
     .await
     .expect("overflow waited for output credit before reporting failure");
 }
+
+#[tokio::test]
+async fn dropped_ssh_session_terminates_in_flight_exec() {
+    tokio::time::timeout(DEADLINE, async {
+        let (session, control) = exec_session();
+        let client = test_client(
+            None,
+            rpc_test::exec(Arc::new(TestExec(Mutex::new([session].into())))),
+            russh::client::Config::default(),
+        )
+        .await;
+        let _channel = start_exec(&client).await;
+
+        assert!(!*control.process.terminated.borrow());
+
+        // Simulate abrupt connection drop by dropping the client handle.
+        drop(client);
+
+        control
+            .process
+            .terminated
+            .subscribe()
+            .wait_for(|done| *done)
+            .await
+            .unwrap();
+    })
+    .await
+    .expect("dropping SSH connection must terminate active exec process");
+}