fix(subagent): shut down children concurrently

src/tools/subagent.rs

@@ -1572,6 +1572,7 @@ impl Subagents {
         // retirement transitions reject late startup installation and preserve
         // permits until a child that failed to close actually exits.
         let mut failure = None;
+        let mut cleanups = Vec::new();
         for (id, state) in states {
             // Select the child and retire its state in one transition. Otherwise
             // startup could install between the empty-child check and retirement.
@@ -1629,14 +1630,19 @@ impl Subagents {
                 self.emit_event(event);
             }
             if let Some(cleanup) = cleanup {
-                let result = cleanup.await.unwrap_or_else(|error| {
-                    Err(ChildError::Failed(format!(
-                        "subagent shutdown cleanup task failed: {error}"
-                    )))
-                });
-                if let Err(error) = result {
-                    failure.get_or_insert(error);
-                }
+                cleanups.push(cleanup);
+            }
+        }
+        // All owned cleanup tasks run concurrently before waiting for results.
+        // Dropping these handles detaches, rather than cancels, child cleanup.
+        for cleanup in cleanups {
+            let result = cleanup.await.unwrap_or_else(|error| {
+                Err(ChildError::Failed(format!(
+                    "subagent shutdown cleanup task failed: {error}"
+                )))
+            });
+            if let Err(error) = result {
+                failure.get_or_insert(error);
             }
         }
         failure.map_or(Ok(()), Err)

src/tools/subagent/tests.rs

@@ -2351,6 +2351,51 @@ async fn shutdown_closes_live_children_without_deleting_history() {
     wait_for_available_permits(&scenario.manager, MAX_LIVE_SUBAGENTS).await;
 }
 
+#[tokio::test]
+async fn shutdown_closes_children_concurrently_and_keeps_cleanup_owned() {
+    for cancel in [false, true] {
+        let scenario = MockAcpScenario::new(ScenarioOptions {
+            gate_close: true,
+            ..Default::default()
+        });
+        let first = scenario.create("first").await;
+        let second = scenario.create("second").await;
+        let manager = scenario.manager.clone();
+        let shutdown = tokio::spawn(async move { manager.shutdown().await });
+        // Neither close can finish until both requests have reached the backend.
+        // The timeout is only a deadlock watchdog, not an elapsed-time assertion.
+        tokio::time::timeout(std::time::Duration::from_secs(3), async {
+            loop {
+                let closes = logged_requests(&scenario.requests)
+                    .iter()
+                    .filter(|request| matches!(request, LoggedRequest::Close { .. }))
+                    .count();
+                if closes == 2 {
+                    break;
+                }
+                tokio::time::sleep(std::time::Duration::from_millis(5)).await;
+            }
+        })
+        .await
+        .expect("both children must receive close before either responds");
+        assert!(scenario.manager.lookup(&first).is_err());
+        assert!(scenario.manager.lookup(&second).is_err());
+        assert_eq!(
+            scenario.manager.capacity.available_permits(),
+            MAX_LIVE_SUBAGENTS - 2
+        );
+        if cancel {
+            shutdown.abort();
+            assert!(shutdown.await.unwrap_err().is_cancelled());
+            MockAcpScenario::release(&scenario.close_release);
+        } else {
+            MockAcpScenario::release(&scenario.close_release);
+            shutdown.await.unwrap().unwrap();
+        }
+        wait_for_available_permits(&scenario.manager, MAX_LIVE_SUBAGENTS).await;
+    }
+}
+
 #[tokio::test]
 async fn shutdown_rejects_unsettled_startup_and_prevents_late_installation() {
     let scenario = MockAcpScenario::new(ScenarioOptions {