goose-agent: run foreground subagents in the GDK (#12792)

crates/goose-agent/src/lib.rs

@@ -2,4 +2,5 @@ pub mod events;
 pub mod inference;
 pub mod machine;
 pub mod operation;
+pub mod subagent;
 pub mod tool;

crates/goose-agent/src/operation.rs

@@ -1,5 +1,6 @@
 use anyhow::{anyhow, Result};
 use async_trait::async_trait;
+use futures::Stream;
 use std::future::Future;
 use std::pin::Pin;
 use tokio::sync::mpsc;
@@ -17,6 +18,12 @@ pub type OperationFuture<'a, T> = Pin<Box<dyn Future<Output = T> + Send + 'a>>;
 #[cfg(target_arch = "wasm32")]
 pub type OperationFuture<'a, T> = Pin<Box<dyn Future<Output = T> + 'a>>;
 
+#[cfg(not(target_arch = "wasm32"))]
+pub type OperationStream<'a, T> = Pin<Box<dyn Stream<Item = T> + Send + 'a>>;
+
+#[cfg(target_arch = "wasm32")]
+pub type OperationStream<'a, T> = Pin<Box<dyn Stream<Item = T> + 'a>>;
+
 pub struct SlashCommand<'a> {
     pub command: &'a str,
     pub params_str: &'a str,

crates/goose-agent/src/subagent.rs

@@ -0,0 +1,290 @@
+use std::collections::HashSet;
+use std::sync::Arc;
+
+use anyhow::Result;
+use async_trait::async_trait;
+use futures::stream::{self, SelectAll, StreamExt};
+use futures::FutureExt;
+use goose_provider_types::conversation::message::Message;
+use goose_provider_types::conversation::Conversation;
+use goose_provider_types::maybe_send::{MaybeSend, MaybeSync};
+use tokio::sync::Mutex;
+use tokio_util::sync::CancellationToken;
+
+use crate::operation::{
+    applied, not_applicable, Emitter, Operation, OperationResult, OperationStream,
+};
+
+const OPERATION_NAME: &str = "foreground_subagent";
+const DELIVERED: &str = "delivered";
+const CANCELLED: &str = "cancelled";
+const ENDED_WITHOUT_OUTCOME: &str = "it ended without an outcome";
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum SubagentOutcome {
+    Completed(String),
+    Failed(String),
+    Cancelled,
+}
+
+#[non_exhaustive]
+#[derive(Debug, Clone)]
+pub enum SubagentUpdate {
+    ToParent(Message),
+    Finished(SubagentOutcome),
+}
+
+#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
+#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
+pub trait SubagentRunner<S>: MaybeSend + MaybeSync {
+    fn started_subagent_session_ids(
+        &self,
+        parent_session: &S,
+        conversation: &Conversation,
+    ) -> Vec<String>;
+
+    async fn start(
+        &self,
+        parent_session: &S,
+        subagent_session_id: &str,
+        cancel: CancellationToken,
+    ) -> OperationStream<'static, SubagentUpdate>;
+
+    fn on_subagent_finished(
+        &self,
+        _subagent_session_id: &str,
+        _outcome: &SubagentOutcome,
+        _remaining: usize,
+        _emit: &Emitter,
+    ) {
+    }
+}
+
+fn finished_subagent_session_ids(message: &Message) -> impl Iterator<Item = &str> {
+    let delivered = message
+        .metadata
+        .operation_note(OPERATION_NAME, DELIVERED)
+        .and_then(serde_json::Value::as_str);
+    let cancelled = message
+        .metadata
+        .operation_note(OPERATION_NAME, CANCELLED)
+        .and_then(serde_json::Value::as_array)
+        .into_iter()
+        .flatten()
+        .filter_map(serde_json::Value::as_str);
+    delivered.into_iter().chain(cancelled)
+}
+
+fn pending_subagents(conversation: &Conversation, started: Vec<String>) -> Vec<String> {
+    let mut seen: HashSet<String> = conversation
+        .messages()
+        .iter()
+        .flat_map(finished_subagent_session_ids)
+        .map(str::to_owned)
+        .collect();
+    started
+        .into_iter()
+        .filter(|subagent_session_id| seen.insert(subagent_session_id.clone()))
+        .collect()
+}
+
+fn subagent_label(subagent_session_id: &str) -> String {
+    format!("Subagent {subagent_session_id}")
+}
+
+fn subagent_result_message(
+    subagent_session_id: &str,
+    outcome: &SubagentOutcome,
+) -> Option<Message> {
+    let label = subagent_label(subagent_session_id);
+    let text = match outcome {
+        SubagentOutcome::Completed(output) => format!("{label} completed: {output}"),
+        SubagentOutcome::Failed(reason) => format!("{label} failed: {reason}"),
+        SubagentOutcome::Cancelled => {
+            return subagent_cancelled_message(&[subagent_session_id.to_string()])
+        }
+    };
+    let mut message = Message::user().with_text(text).with_visibility(false, true);
+    message.metadata.set_operation_note(
+        OPERATION_NAME,
+        DELIVERED,
+        serde_json::json!(subagent_session_id),
+    );
+    Some(message)
+}
+
+fn subagent_cancelled_message(subagent_session_ids: &[String]) -> Option<Message> {
+    if subagent_session_ids.is_empty() {
+        return None;
+    }
+    let text = subagent_session_ids
+        .iter()
+        .map(|subagent_session_id| {
+            format!(
+                "{} was cancelled before it finished and will not run again.",
+                subagent_label(subagent_session_id)
+            )
+        })
+        .collect::<Vec<_>>()
+        .join("\n");
+    let mut message = Message::user().with_text(text).with_visibility(false, true);
+    message.metadata.set_operation_note(
+        OPERATION_NAME,
+        CANCELLED,
+        serde_json::json!(subagent_session_ids),
+    );
+    Some(message)
+}
+
+fn tracked_updates(
+    subagent_session_id: String,
+    updates: OperationStream<'static, SubagentUpdate>,
+) -> OperationStream<'static, (String, SubagentUpdate)> {
+    Box::pin(stream::unfold(Some(updates), move |updates| {
+        let subagent_session_id = subagent_session_id.clone();
+        async move {
+            let mut updates = updates?;
+            let update = updates.next().await.unwrap_or_else(|| {
+                SubagentUpdate::Finished(SubagentOutcome::Failed(ENDED_WITHOUT_OUTCOME.to_string()))
+            });
+            let rest = match update {
+                SubagentUpdate::Finished(_) => None,
+                SubagentUpdate::ToParent(_) => Some(updates),
+            };
+            Some(((subagent_session_id, update), rest))
+        }
+    }))
+}
+
+#[derive(Default)]
+struct Running {
+    subagent_session_ids: HashSet<String>,
+    subagent_runs: SelectAll<OperationStream<'static, (String, SubagentUpdate)>>,
+}
+
+pub struct SubagentOperation<S> {
+    runner: Arc<dyn SubagentRunner<S>>,
+    running: Mutex<Running>,
+}
+
+impl<S> SubagentOperation<S> {
+    pub fn new(runner: Arc<dyn SubagentRunner<S>>) -> Self {
+        Self {
+            runner,
+            running: Mutex::default(),
+        }
+    }
+
+    fn deliver<E: From<Message>>(
+        &self,
+        subagent_session_id: &str,
+        outcome: &SubagentOutcome,
+        remaining: usize,
+        emit: &Emitter,
+    ) -> Option<E> {
+        let message = subagent_result_message(subagent_session_id, outcome)?;
+        self.runner
+            .on_subagent_finished(subagent_session_id, outcome, remaining, emit);
+        Some(E::from(message))
+    }
+}
+
+#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
+#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
+impl<S, E> Operation<S, E> for SubagentOperation<S>
+where
+    S: MaybeSend + MaybeSync + 'static,
+    E: From<Message> + MaybeSend + 'static,
+{
+    fn name(&self) -> &'static str {
+        OPERATION_NAME
+    }
+
+    async fn finalize_cancellation(
+        &self,
+        session: &S,
+        conversation: &Conversation,
+        emit: &Emitter,
+    ) -> Vec<E> {
+        let mut stopped = std::mem::take(&mut *self.running.lock().await);
+        let pending = pending_subagents(
+            conversation,
+            self.runner
+                .started_subagent_session_ids(session, conversation),
+        );
+        let mut effects = Vec::new();
+        let mut delivered = HashSet::new();
+        let finished = std::iter::from_fn(|| loop {
+            match stopped.subagent_runs.next().now_or_never().flatten()? {
+                (subagent_session_id, SubagentUpdate::Finished(outcome)) => {
+                    return Some((subagent_session_id, outcome))
+                }
+                (_, SubagentUpdate::ToParent(_)) => {}
+            }
+        });
+        for (subagent_session_id, outcome) in finished {
+            if outcome == SubagentOutcome::Cancelled {
+                continue;
+            }
+            let remaining = pending.len().saturating_sub(delivered.len() + 1);
+            if let Some(effect) = self.deliver(&subagent_session_id, &outcome, remaining, emit) {
+                delivered.insert(subagent_session_id);
+                effects.push(effect);
+            }
+        }
+        drop(stopped);
+
+        let cancelled: Vec<String> = pending
+            .into_iter()
+            .filter(|subagent_session_id| !delivered.contains(subagent_session_id))
+            .collect();
+        effects.extend(subagent_cancelled_message(&cancelled).map(E::from));
+        effects
+    }
+
+    async fn run(
+        &self,
+        session: &S,
+        conversation: &Conversation,
+        emit: &Emitter,
+    ) -> Result<OperationResult<E>> {
+        let pending = pending_subagents(
+            conversation,
+            self.runner
+                .started_subagent_session_ids(session, conversation),
+        );
+        let remaining = pending.len().saturating_sub(1);
+
+        let mut running = self.running.lock().await;
+        for subagent_session_id in &pending {
+            if running.subagent_session_ids.contains(subagent_session_id) {
+                continue;
+            }
+            let updates = self
+                .runner
+                .start(session, subagent_session_id, emit.cancel_token().clone())
+                .await;
+            running
+                .subagent_session_ids
+                .insert(subagent_session_id.clone());
+            running
+                .subagent_runs
+                .push(tracked_updates(subagent_session_id.clone(), updates));
+        }
+
+        while let Some((subagent_session_id, update)) = running.subagent_runs.next().await {
+            let outcome = match update {
+                SubagentUpdate::ToParent(message) => {
+                    emit.message(message);
+                    continue;
+                }
+                SubagentUpdate::Finished(outcome) => outcome,
+            };
+            running.subagent_session_ids.remove(&subagent_session_id);
+            if let Some(effect) = self.deliver(&subagent_session_id, &outcome, remaining, emit) {
+                return applied([effect]);
+            }
+        }
+        not_applicable()
+    }
+}

crates/goose-agent/tests/subagent.rs

@@ -0,0 +1,450 @@
+use std::collections::HashMap;
+use std::sync::{Arc, Mutex};
+
+use async_trait::async_trait;
+use futures::future;
+use futures::stream;
+use futures::FutureExt;
+use goose_agent::events::AgentEvent;
+use goose_agent::operation::{
+    ConversationEffect, Emitter, Operation, OperationResult, OperationStream,
+};
+use goose_agent::subagent::{SubagentOperation, SubagentOutcome, SubagentRunner, SubagentUpdate};
+use goose_provider_types::conversation::{message::Message, Conversation};
+use serde_json::json;
+use tokio::sync::mpsc;
+use tokio_util::sync::CancellationToken;
+
+struct Parent;
+
+enum Child {
+    Finished(SubagentOutcome),
+    Running(mpsc::UnboundedReceiver<SubagentUpdate>),
+}
+
+struct RunningChild(mpsc::UnboundedSender<SubagentUpdate>);
+
+impl RunningChild {
+    fn to_parent(&self, text: &str) {
+        self.0
+            .send(SubagentUpdate::ToParent(
+                Message::assistant().with_text(text),
+            ))
+            .unwrap();
+    }
+
+    fn finish(&self, outcome: SubagentOutcome) {
+        self.0.send(SubagentUpdate::Finished(outcome)).unwrap();
+    }
+
+    fn is_closed(&self) -> bool {
+        self.0.is_closed()
+    }
+}
+
+#[derive(Default)]
+struct FakeRunner {
+    children: Mutex<HashMap<String, Child>>,
+    started: Mutex<Vec<String>>,
+    events: Mutex<Vec<String>>,
+    started_subagent_session_ids: Vec<String>,
+}
+
+impl FakeRunner {
+    fn finished(&self, subagent_session_id: &str, outcome: SubagentOutcome) {
+        self.children
+            .lock()
+            .unwrap()
+            .insert(subagent_session_id.to_string(), Child::Finished(outcome));
+    }
+
+    fn running(&self, subagent_session_id: &str) -> RunningChild {
+        let (tx, rx) = mpsc::unbounded_channel();
+        self.children
+            .lock()
+            .unwrap()
+            .insert(subagent_session_id.to_string(), Child::Running(rx));
+        RunningChild(tx)
+    }
+
+    fn started(&self) -> Vec<String> {
+        self.started.lock().unwrap().clone()
+    }
+
+    fn events(&self) -> Vec<String> {
+        self.events.lock().unwrap().clone()
+    }
+}
+
+#[async_trait]
+impl SubagentRunner<Parent> for FakeRunner {
+    fn started_subagent_session_ids(
+        &self,
+        _parent_session: &Parent,
+        _conversation: &Conversation,
+    ) -> Vec<String> {
+        self.started_subagent_session_ids.clone()
+    }
+
+    async fn start(
+        &self,
+        _parent_session: &Parent,
+        subagent_session_id: &str,
+        _cancel: CancellationToken,
+    ) -> OperationStream<'static, SubagentUpdate> {
+        self.started
+            .lock()
+            .unwrap()
+            .push(subagent_session_id.to_string());
+        let child = self
+            .children
+            .lock()
+            .unwrap()
+            .remove(subagent_session_id)
+            .expect("unknown subagent");
+        match child {
+            Child::Finished(outcome) => Box::pin(stream::once(future::ready(
+                SubagentUpdate::Finished(outcome),
+            ))),
+            Child::Running(rx) => {
+                self.events
+                    .lock()
+                    .unwrap()
+                    .push(format!("started {subagent_session_id}"));
+                Box::pin(stream::unfold(rx, |mut rx| async move {
+                    rx.recv().await.map(|update| (update, rx))
+                }))
+            }
+        }
+    }
+
+    fn on_subagent_finished(
+        &self,
+        subagent_session_id: &str,
+        _outcome: &SubagentOutcome,
+        remaining: usize,
+        _emit: &Emitter,
+    ) {
+        self.events.lock().unwrap().push(format!(
+            "finished {subagent_session_id}, {remaining} remaining"
+        ));
+    }
+}
+
+fn emitter() -> Emitter {
+    let (tx, _rx) = mpsc::unbounded_channel();
+    Emitter::new(tx, CancellationToken::new())
+}
+
+fn watched_emitter() -> (Emitter, mpsc::UnboundedReceiver<AgentEvent>) {
+    let (tx, rx) = mpsc::unbounded_channel();
+    (Emitter::new(tx, CancellationToken::new()), rx)
+}
+
+fn shown(rx: &mut mpsc::UnboundedReceiver<AgentEvent>) -> Vec<String> {
+    std::iter::from_fn(|| rx.try_recv().ok())
+        .filter_map(|event| match event {
+            AgentEvent::Message(message) => Some(message.as_concat_text()),
+            _ => None,
+        })
+        .collect()
+}
+
+fn kickoff() -> Message {
+    Message::user().with_text("go")
+}
+
+fn fake(subagent_session_ids: &[&str]) -> FakeRunner {
+    FakeRunner {
+        started_subagent_session_ids: subagent_session_ids
+            .iter()
+            .map(|id| id.to_string())
+            .collect(),
+        ..FakeRunner::default()
+    }
+}
+
+fn appended(effects: Vec<ConversationEffect>) -> Vec<Message> {
+    effects
+        .into_iter()
+        .map(|effect| match effect {
+            ConversationEffect::AppendMessage(message) => message,
+            _ => panic!("expected an appended message"),
+        })
+        .collect()
+}
+
+fn texts(messages: &[Message]) -> Vec<String> {
+    messages.iter().map(Message::as_concat_text).collect()
+}
+
+fn operation(runner: &Arc<FakeRunner>) -> SubagentOperation<Parent> {
+    SubagentOperation::new(runner.clone())
+}
+
+async fn run_once(operation: &SubagentOperation<Parent>, messages: &[Message]) -> Vec<Message> {
+    run_once_with(operation, messages, &emitter()).await
+}
+
+async fn run_once_with(
+    operation: &SubagentOperation<Parent>,
+    messages: &[Message],
+    emit: &Emitter,
+) -> Vec<Message> {
+    let conversation = Conversation::new_unvalidated(messages.to_vec());
+    let result =
+        Operation::<Parent, ConversationEffect>::run(operation, &Parent, &conversation, emit)
+            .await
+            .unwrap();
+    match result {
+        OperationResult::NotApplicable => Vec::new(),
+        OperationResult::Applied(result) => appended(result.effects),
+    }
+}
+
+fn interrupt(operation: &SubagentOperation<Parent>, messages: &[Message]) {
+    let conversation = Conversation::new_unvalidated(messages.to_vec());
+    let run =
+        Operation::<Parent, ConversationEffect>::run(operation, &Parent, &conversation, &emitter())
+            .now_or_never();
+    assert!(run.is_none());
+}
+
+async fn stop(operation: &SubagentOperation<Parent>, messages: &[Message]) -> Vec<Message> {
+    stop_with(operation, messages, &emitter()).await
+}
+
+async fn stop_with(
+    operation: &SubagentOperation<Parent>,
+    messages: &[Message],
+    emit: &Emitter,
+) -> Vec<Message> {
+    let conversation = Conversation::new_unvalidated(messages.to_vec());
+    appended(
+        Operation::<Parent, ConversationEffect>::finalize_cancellation(
+            operation,
+            &Parent,
+            &conversation,
+            emit,
+        )
+        .await,
+    )
+}
+
+#[tokio::test]
+async fn a_finished_subagent_is_delivered_without_running() {
+    let runner = Arc::new(fake(&["a"]));
+    runner.finished("a", SubagentOutcome::Completed("done".into()));
+    let operation = operation(&runner);
+    let mut messages = vec![kickoff()];
+
+    let delivered = run_once(&operation, &messages).await;
+
+    assert_eq!(texts(&delivered), ["Subagent a completed: done"]);
+    assert!(!delivered[0].is_user_visible());
+    assert!(delivered[0].is_agent_visible());
+    assert_eq!(runner.events(), ["finished a, 0 remaining"]);
+
+    messages.extend(delivered);
+    assert!(run_once(&operation, &messages).await.is_empty());
+    assert_eq!(runner.started(), ["a"]);
+}
+
+#[tokio::test]
+async fn running_subagents_are_delivered_one_per_pass_and_started_once() {
+    let runner = Arc::new(fake(&["a", "b"]));
+    let a = runner.running("a");
+    let b = runner.running("b");
+    let operation = operation(&runner);
+    let mut messages = vec![kickoff()];
+
+    b.finish(SubagentOutcome::Completed("b done".into()));
+    let delivered = run_once(&operation, &messages).await;
+    assert_eq!(texts(&delivered), ["Subagent b completed: b done"]);
+    messages.extend(delivered);
+
+    a.finish(SubagentOutcome::Failed("boom".into()));
+    let delivered = run_once(&operation, &messages).await;
+    assert_eq!(texts(&delivered), ["Subagent a failed: boom"]);
+    messages.extend(delivered);
+
+    assert!(run_once(&operation, &messages).await.is_empty());
+    assert_eq!(runner.started(), ["a", "b"]);
+    assert_eq!(
+        runner.events(),
+        [
+            "started a",
+            "started b",
+            "finished b, 1 remaining",
+            "finished a, 0 remaining",
+        ]
+    );
+}
+
+#[tokio::test]
+async fn an_already_finished_subagent_does_not_hold_back_the_others() {
+    let runner = Arc::new(fake(&["a", "b"]));
+    runner.finished("a", SubagentOutcome::Completed("a done".into()));
+    let _b = runner.running("b");
+
+    let delivered = run_once(&operation(&runner), &[kickoff()]).await;
+
+    assert_eq!(texts(&delivered), ["Subagent a completed: a done"]);
+    assert_eq!(runner.started(), ["a", "b"]);
+    assert_eq!(runner.events(), ["started b", "finished a, 1 remaining"]);
+}
+
+#[tokio::test]
+async fn a_subagent_cancelled_elsewhere_is_reported_and_not_restarted() {
+    let runner = Arc::new(fake(&["a"]));
+    let a = runner.running("a");
+    let operation = operation(&runner);
+    let mut messages = vec![kickoff()];
+
+    a.finish(SubagentOutcome::Cancelled);
+    let delivered = run_once(&operation, &messages).await;
+
+    assert_eq!(
+        texts(&delivered),
+        ["Subagent a was cancelled before it finished and will not run again."]
+    );
+    messages.extend(delivered);
+    assert!(run_once(&operation, &messages).await.is_empty());
+    assert_eq!(runner.started(), ["a"]);
+}
+
+#[tokio::test]
+async fn stop_delivers_finished_subagents_and_cancels_the_rest() {
+    let runner = Arc::new(fake(&["a", "b", "c"]));
+    let a = runner.running("a");
+    let b = runner.running("b");
+    let c = runner.running("c");
+    let operation = operation(&runner);
+    let mut messages = vec![kickoff()];
+
+    interrupt(&operation, &messages);
+    a.finish(SubagentOutcome::Completed("a done".into()));
+    b.finish(SubagentOutcome::Cancelled);
+    let saved = stop(&operation, &messages).await;
+
+    assert_eq!(
+        texts(&saved),
+        [
+            "Subagent a completed: a done",
+            "Subagent b was cancelled before it finished and will not run again.\n\
+             Subagent c was cancelled before it finished and will not run again.",
+        ]
+    );
+    assert!(c.is_closed());
+    assert_eq!(
+        runner.events(),
+        [
+            "started a",
+            "started b",
+            "started c",
+            "finished a, 2 remaining"
+        ]
+    );
+
+    messages.extend(saved);
+    assert!(run_once(&operation, &messages).await.is_empty());
+    assert_eq!(runner.started(), ["a", "b", "c"]);
+    assert!(stop(&operation, &messages).await.is_empty());
+}
+
+#[tokio::test]
+async fn a_repeated_id_starts_once() {
+    let runner = Arc::new(fake(&["x", "x"]));
+    runner.finished("x", SubagentOutcome::Failed("boom\ndetails".into()));
+
+    let delivered = run_once(&operation(&runner), &[kickoff()]).await;
+
+    assert_eq!(texts(&delivered), ["Subagent x failed: boom\ndetails"]);
+    assert_eq!(runner.started(), ["x"]);
+}
+
+#[tokio::test]
+async fn reads_the_notes_goose_saves_today() {
+    let mut delivered = Message::user()
+        .with_text("Subagent a completed: done")
+        .with_visibility(false, true);
+    delivered
+        .metadata
+        .set_operation_note("foreground_subagent", "delivered", json!("a"));
+    let mut cancelled = Message::user()
+        .with_text("Subagent b was cancelled")
+        .with_visibility(false, true);
+    cancelled
+        .metadata
+        .set_operation_note("foreground_subagent", "cancelled", json!(["b"]));
+    let messages = vec![kickoff(), delivered, cancelled];
+
+    let runner = Arc::new(fake(&["a", "b", "c"]));
+    runner.finished("c", SubagentOutcome::Completed("c done".into()));
+    let delivered = run_once(&operation(&runner), &messages).await;
+
+    assert_eq!(texts(&delivered), ["Subagent c completed: c done"]);
+    assert_eq!(runner.started(), ["c"]);
+}
+
+#[tokio::test]
+async fn messages_from_a_running_subagent_reach_the_parent_before_its_result() {
+    let runner = Arc::new(fake(&["a"]));
+    let a = runner.running("a");
+    let operation = operation(&runner);
+    let (emit, mut rx) = watched_emitter();
+
+    a.to_parent("may I run cargo test?");
+    a.finish(SubagentOutcome::Completed("done".into()));
+    let delivered = run_once_with(&operation, &[kickoff()], &emit).await;
+
+    assert_eq!(shown(&mut rx), ["may I run cargo test?"]);
+    assert_eq!(texts(&delivered), ["Subagent a completed: done"]);
+}
+
+#[tokio::test]
+async fn a_subagent_whose_updates_end_without_an_outcome_has_failed() {
+    let runner = Arc::new(fake(&["a"]));
+    drop(runner.running("a"));
+
+    let delivered = run_once(&operation(&runner), &[kickoff()]).await;
+
+    assert_eq!(
+        texts(&delivered),
+        ["Subagent a failed: it ended without an outcome"]
+    );
+}
+
+#[tokio::test]
+async fn updates_after_the_outcome_are_ignored() {
+    let runner = Arc::new(fake(&["a"]));
+    let a = runner.running("a");
+    let operation = operation(&runner);
+    let (emit, mut rx) = watched_emitter();
+
+    a.finish(SubagentOutcome::Completed("done".into()));
+    a.to_parent("too late");
+    a.finish(SubagentOutcome::Failed("again".into()));
+    let delivered = run_once_with(&operation, &[kickoff()], &emit).await;
+
+    assert_eq!(texts(&delivered), ["Subagent a completed: done"]);
+    assert!(shown(&mut rx).is_empty());
+    assert!(a.is_closed());
+}
+
+#[tokio::test]
+async fn stop_skips_messages_queued_ahead_of_a_finished_result() {
+    let runner = Arc::new(fake(&["a"]));
+    let a = runner.running("a");
+    let operation = operation(&runner);
+    let messages = [kickoff()];
+    let (emit, mut rx) = watched_emitter();
+
+    interrupt(&operation, &messages);
+    a.to_parent("may I run cargo test?");
+    a.finish(SubagentOutcome::Completed("done".into()));
+    let saved = stop_with(&operation, &messages, &emit).await;
+
+    assert_eq!(texts(&saved), ["Subagent a completed: done"]);
+    assert!(shown(&mut rx).is_empty());
+}

crates/goose/Cargo.toml

@@ -202,7 +202,7 @@ blake3 = { version = "1", default-features = false, features = ["std"] }
 fs2 = { workspace = true }
 tokio-stream = { workspace = true, features = ["io-util"] }
 tempfile = { workspace = true }
-tokio-util = { workspace = true, features = ["compat"] }
+tokio-util = { workspace = true, features = ["compat", "rt"] }
 agent-client-protocol-schema = { workspace = true }
 agent-client-protocol = { workspace = true, features = [
     "unstable_end_turn_token_usage",

crates/goose/src/agents/agent.rs

@@ -22,7 +22,7 @@ use crate::agents::state_machine::{
     persist_tool_confirmation_decision, run_goose, Emitter, GooseEffect, StateMachine, SteerQueue,
 };
 use crate::agents::state_machine_builder::{build_state_machine, StateMachineConfig};
-use crate::agents::subagent_handler::SubagentRunner;
+use crate::agents::subagent_handler::ForegroundSubagentRunner;
 use crate::agents::types::{
     SessionConfig, DEFAULT_ON_FAILURE_TIMEOUT_SECONDS, DEFAULT_RETRY_TIMEOUT_SECONDS,
 };
@@ -607,7 +607,7 @@ impl Agent {
             tool_inspection_manager: &self.tool_inspection_manager,
             hook_manager: self.hook_manager.clone(),
             steer_queue,
-            subagent_runner: SubagentRunner::new(Arc::clone(self)),
+            subagent_runner: ForegroundSubagentRunner::new(Arc::clone(self)),
             retry_timeout: std::time::Duration::from_secs(retry_timeout),
             on_failure_timeout: std::time::Duration::from_secs(on_failure_timeout),
             stop_hook_block_cap,

crates/goose/src/agents/state_machine/mod.rs

@@ -45,6 +45,7 @@ pub use goose_agent::operation::{
     not_applicable, trailing_error, yielded, yielded_with, ConversationEffect, Emitter, Inference,
     InferenceInput, MachineEffect, Operation, OperationResult, RunStatus, SlashCommand, StepResult,
 };
+pub(super) use goose_agent::subagent::SubagentOperation;
 pub(crate) use tool_confirmation::{
     has_unapplied_tool_confirmation_response, pending_tool_confirmations,
     persist_tool_confirmation_decision,
@@ -67,7 +68,7 @@ pub(super) use ops_slash_command::SlashCommandOperation;
 pub(super) use ops_status::StatusOperation;
 pub(super) use ops_steer::{SteerOperation, SteerQueue};
 pub(super) use ops_stop_hook::StopHookOperation;
-pub(super) use ops_subagent::{SubagentOperation, DELEGATED_META_KEY};
+pub(super) use ops_subagent::DELEGATED_META_KEY;
 pub(super) use ops_tool_approval::ToolApprovalOperation;
 pub(super) use ops_tool_pair_compaction::ToolPairCompactionOperation;
 pub(super) use ops_toolcalling::ToolExecutionOperation;

crates/goose/src/agents/state_machine/ops_subagent.rs

@@ -1,38 +1,26 @@
-use std::collections::{HashMap, HashSet};
-
-use anyhow::{anyhow, Result};
 use async_trait::async_trait;
-use tokio::sync::{mpsc, Mutex};
-use tokio::task::{self, JoinSet};
+use futures::future;
+use futures::stream::{self, BoxStream, StreamExt};
+use goose_agent::subagent::{SubagentOutcome, SubagentRunner, SubagentUpdate};
 use tokio_util::sync::CancellationToken;
-use tracing::Instrument;
 
-use crate::agents::state_machine::{
-    applied, not_applicable, Emitter, GooseEffect, Operation, OperationResult,
-};
-use crate::agents::subagent_handler::{SubagentOutcome, SubagentRunner, SubagentStart};
+use crate::agents::final_output_tool::FinalOutputTool;
+use crate::agents::state_machine::Emitter;
+use crate::agents::subagent_handler::ForegroundSubagentRunner;
 use crate::conversation::message::{Message, MessageContent, SystemNotificationType};
 use crate::conversation::Conversation;
 use crate::session::{Session, SessionType};
 use crate::utils::safe_truncate;
 
-// Both keys predate background subagents' removal and live on in session histories:
-// renaming the note scope would deliver every past subagent's result again, and older
+// Predates background subagents' removal and lives on in session histories, where older
 // histories still hold background delegations marked `false`.
-const OPERATION_NAME: &str = "foreground_subagent";
 pub(crate) const DELEGATED_META_KEY: &str = "foreground_subagent";
-const DELIVERED: &str = "delivered";
-const CANCELLED: &str = "cancelled";
 const TASK_SNIPPET_CHARS: usize = 160;
 
 fn inline_notice(text: String) -> Message {
     Message::assistant().with_system_notification(SystemNotificationType::InlineMessage, text)
 }
 
-fn subagent_label(subagent_id: &str) -> String {
-    format!("Subagent {subagent_id}")
-}
-
 fn start_notice(subagent_id: &str, task: Option<&str>) -> String {
     let snippet = task
         .map(|task| {
@@ -43,6 +31,34 @@ fn start_notice(subagent_id: &str, task: Option<&str>) -> String {
     format!("Running subagent {subagent_id}{snippet}")
 }
 
+fn finished_notice(
+    subagent_id: &str,
+    outcome: &SubagentOutcome,
+    show_output: bool,
+) -> Option<String> {
+    let notice = match outcome {
+        SubagentOutcome::Completed(output) if show_output => {
+            format!(
+                "Subagent {subagent_id} completed\n\n{}",
+                readable_output(output)
+            )
+        }
+        SubagentOutcome::Completed(_) => format!("Subagent {subagent_id} completed"),
+        SubagentOutcome::Failed(reason) => format!(
+            "Subagent {subagent_id} failed: {}",
+            reason.lines().next().unwrap_or_default()
+        ),
+        SubagentOutcome::Cancelled => return None,
+    };
+    Some(notice)
+}
+
+fn ended(outcome: SubagentOutcome) -> BoxStream<'static, SubagentUpdate> {
+    Box::pin(stream::once(future::ready(SubagentUpdate::Finished(
+        outcome,
+    ))))
+}
+
 fn delegated_subagent_ids(content: &[MessageContent]) -> impl Iterator<Item = &str> {
     content.iter().filter_map(|content| {
         let MessageContent::ToolResponse(response) = content else {
@@ -59,38 +75,6 @@ fn delegated_subagent_ids(content: &[MessageContent]) -> impl Iterator<Item = &s
     })
 }
 
-fn delivered_or_cancelled_ids(message: &Message) -> impl Iterator<Item = &str> {
-    let delivered = message
-        .metadata
-        .operation_note(OPERATION_NAME, DELIVERED)
-        .and_then(serde_json::Value::as_str);
-    let cancelled = message
-        .metadata
-        .operation_note(OPERATION_NAME, CANCELLED)
-        .and_then(serde_json::Value::as_array)
-        .into_iter()
-        .flatten()
-        .filter_map(serde_json::Value::as_str);
-    delivered.into_iter().chain(cancelled)
-}
-
-fn pending_subagents(messages: &[Message]) -> Vec<String> {
-    let mut seen: HashSet<&str> = messages
-        .iter()
-        .flat_map(delivered_or_cancelled_ids)
-        .collect();
-    let mut pending: Vec<String> = Vec::new();
-    for subagent_id in messages
-        .iter()
-        .flat_map(|message| delegated_subagent_ids(&message.content))
-    {
-        if seen.insert(subagent_id) {
-            pending.push(subagent_id.to_owned());
-        }
-    }
-    pending
-}
-
 fn readable_output(output: &str) -> String {
     match serde_json::from_str::<serde_json::Value>(output) {
         Ok(serde_json::Value::Object(fields)) if fields.len() == 1 => fields
@@ -101,202 +85,67 @@ fn readable_output(output: &str) -> String {
     }
 }
 
-fn subagent_result_message(
-    subagent_id: &str,
-    outcome: SubagentOutcome,
-    others_waiting: bool,
-    emit: &Emitter,
-) -> Option<GooseEffect> {
-    let label = subagent_label(subagent_id);
-    let (notice, text) = match outcome {
-        SubagentOutcome::Completed(output) => (
-            if others_waiting {
-                format!("{label} completed\n\n{}", readable_output(&output))
-            } else {
-                format!("{label} completed")
-            },
-            format!("{label} completed: {output}"),
-        ),
-        SubagentOutcome::Failed(reason) => (
-            format!(
-                "{label} failed: {}",
-                reason.lines().next().unwrap_or_default()
-            ),
-            format!("{label} failed: {reason}"),
-        ),
-        SubagentOutcome::Cancelled => return None,
-    };
-    emit.message(inline_notice(notice));
-    let mut message = Message::user().with_text(text).with_visibility(false, true);
-    message
-        .metadata
-        .set_operation_note(OPERATION_NAME, DELIVERED, serde_json::json!(subagent_id));
-    Some(message.into())
-}
-
-fn subagent_cancelled_message(pending: Vec<String>) -> Option<Message> {
-    if pending.is_empty() {
-        return None;
-    }
-    let text = pending
-        .iter()
-        .map(|subagent_id| {
-            format!(
-                "{} was cancelled before it finished and will not run again.",
-                subagent_label(subagent_id)
-            )
-        })
-        .collect::<Vec<_>>()
-        .join("\n");
-    let mut message = Message::user().with_text(text).with_visibility(false, true);
-    message
-        .metadata
-        .set_operation_note(OPERATION_NAME, CANCELLED, serde_json::json!(pending));
-    Some(message)
-}
-
-#[derive(Default)]
-struct RunningSubagents {
-    tasks: JoinSet<SubagentOutcome>,
-    subagent_ids: HashMap<task::Id, String>,
-}
-
-impl RunningSubagents {
-    fn contains(&self, subagent_id: &str) -> bool {
-        self.subagent_ids
-            .values()
-            .any(|running| running == subagent_id)
-    }
-}
-
-pub struct SubagentOperation {
-    runner: SubagentRunner,
-    cancel: CancellationToken,
-    running: Mutex<RunningSubagents>,
-    forward_tx: mpsc::UnboundedSender<Message>,
-    forwarded: Mutex<mpsc::UnboundedReceiver<Message>>,
-}
-
-impl SubagentOperation {
-    pub fn new(runner: SubagentRunner, cancel: CancellationToken) -> Self {
-        let (forward_tx, forwarded) = mpsc::unbounded_channel();
-        Self {
-            runner,
-            cancel,
-            running: Mutex::default(),
-            forward_tx,
-            forwarded: Mutex::new(forwarded),
-        }
-    }
-}
-
 #[async_trait]
-impl Operation<Session, GooseEffect> for SubagentOperation {
-    fn name(&self) -> &'static str {
-        OPERATION_NAME
-    }
-
-    async fn finalize_cancellation(
+impl SubagentRunner<Session> for ForegroundSubagentRunner {
+    fn started_subagent_session_ids(
         &self,
-        session: &Session,
+        parent_session: &Session,
         conversation: &Conversation,
-        emit: &Emitter,
-    ) -> Vec<GooseEffect> {
-        if session.session_type == SessionType::SubAgent {
+    ) -> Vec<String> {
+        if parent_session.session_type == SessionType::SubAgent {
             return Vec::new();
         }
-        let mut stopped = std::mem::take(&mut *self.running.lock().await);
-        let mut effects = Vec::new();
-        let mut delivered = HashSet::new();
-        while let Some(finished) = stopped.tasks.try_join_next_with_id() {
-            let Ok((task_id, outcome)) = finished else {
-                continue;
-            };
-            let Some(subagent_id) = stopped.subagent_ids.remove(&task_id) else {
-                continue;
-            };
-            if let Some(message) = subagent_result_message(&subagent_id, outcome, true, emit) {
-                effects.push(message);
-                delivered.insert(subagent_id);
-            }
-        }
-        stopped.tasks.abort_all();
-
-        let pending = pending_subagents(conversation.messages())
-            .into_iter()
-            .filter(|subagent_id| !delivered.contains(subagent_id))
-            .collect();
-        effects.extend(subagent_cancelled_message(pending).map(GooseEffect::from));
-        effects
+        conversation
+            .messages()
+            .iter()
+            .flat_map(|message| delegated_subagent_ids(&message.content))
+            .map(str::to_owned)
+            .collect()
     }
 
-    async fn run(
+    async fn start(
         &self,
-        session: &Session,
-        conversation: &Conversation,
-        emit: &Emitter,
-    ) -> Result<OperationResult<GooseEffect>> {
-        if session.session_type == SessionType::SubAgent {
-            return not_applicable();
-        }
-        let pending = pending_subagents(conversation.messages());
-        if pending.is_empty() {
-            return not_applicable();
-        }
-
-        let mut running = self.running.lock().await;
-        for subagent_id in &pending {
-            if running.contains(subagent_id) {
-                continue;
-            }
-            match self
-                .runner
-                .start(
-                    &session.id,
-                    subagent_id,
-                    self.cancel.clone(),
-                    self.forward_tx.clone(),
-                )
-                .await
-            {
-                SubagentStart::HasOutcome(outcome) => {
-                    if let Some(message) =
-                        subagent_result_message(subagent_id, outcome, pending.len() > 1, emit)
-                    {
-                        return applied([message]);
-                    }
-                }
-                SubagentStart::Started { task, run } => {
-                    emit.message(inline_notice(start_notice(subagent_id, task.as_deref())));
-                    let handle = running.tasks.spawn(run.in_current_span());
-                    running
-                        .subagent_ids
-                        .insert(handle.id(), subagent_id.to_string());
-                }
-            }
-        }
-
-        let mut forwarded = self.forwarded.lock().await;
-        let finished = loop {
-            tokio::select! {
-                Some(message) = forwarded.recv() => {
-                    emit.message(message);
-                }
-                finished = running.tasks.join_next_with_id() => break finished,
-            }
-        }
-        .ok_or_else(|| anyhow!("No subagent is running"))?;
-        let (task_id, outcome) = match finished {
-            Ok(finished) => finished,
-            Err(error) => (error.id(), SubagentOutcome::Failed(error.to_string())),
+        parent_session: &Session,
+        subagent_session_id: &str,
+        cancel: CancellationToken,
+    ) -> BoxStream<'static, SubagentUpdate> {
+        let subagent = match self.load(subagent_session_id).await {
+            Ok(subagent) => subagent,
+            Err(error) => return ended(SubagentOutcome::Failed(error.to_string())),
         };
-        let subagent_id = running
-            .subagent_ids
-            .remove(&task_id)
-            .ok_or_else(|| anyhow!("Unknown subagent task {task_id}"))?;
-        match subagent_result_message(&subagent_id, outcome, pending.len() > 1, emit) {
-            Some(message) => applied([message]),
-            None => not_applicable(),
+        if subagent.session_type != SessionType::SubAgent
+            || subagent.parent_session_id.as_deref() != Some(parent_session.id.as_str())
+        {
+            return ended(SubagentOutcome::Failed(
+                "it does not belong to this session".to_string(),
+            ));
+        }
+        if let Some(output) = subagent
+            .conversation
+            .as_ref()
+            .and_then(|conversation| FinalOutputTool::successful_output(conversation.messages()))
+        {
+            return ended(SubagentOutcome::Completed(output));
+        }
+        let task = subagent
+            .recipe
+            .as_ref()
+            .and_then(|recipe| recipe.prompt.as_deref());
+        let notice =
+            SubagentUpdate::ToParent(inline_notice(start_notice(subagent_session_id, task)));
+        Box::pin(stream::once(future::ready(notice)).chain(self.spawn(subagent, cancel)))
+    }
+
+    fn on_subagent_finished(
+        &self,
+        subagent_session_id: &str,
+        outcome: &SubagentOutcome,
+        remaining: usize,
+        emit: &Emitter,
+    ) {
+        let show_output = remaining > 0 || emit.cancel_token().is_cancelled();
+        if let Some(notice) = finished_notice(subagent_session_id, outcome, show_output) {
+            emit.message(inline_notice(notice));
         }
     }
 }
@@ -304,20 +153,26 @@ impl Operation<Session, GooseEffect> for SubagentOperation {
 #[cfg(test)]
 mod tests {
     use std::sync::Arc;
-    use std::time::Duration;
 
+    use anyhow::Result;
     use goose_agent::events::AgentEvent;
+    use goose_agent::subagent::SubagentOperation;
     use rmcp::model::{CallToolRequestParams, CallToolResult, ContentBlock, MetaObject};
     use tempfile::TempDir;
     use tokio::sync::mpsc;
 
     use super::*;
     use crate::agents::final_output_tool::{FINAL_OUTPUT_SUCCESS_MESSAGE, FINAL_OUTPUT_TOOL_NAME};
-    use crate::agents::state_machine::{ConversationEffect, RunStatus};
+    use crate::agents::state_machine::{
+        ConversationEffect, GooseEffect, Operation, OperationResult, RunStatus,
+    };
     use crate::config::GooseMode;
     use crate::session::SessionManager;
     use goose_agent::machine::EffectHandler;
 
+    const OPERATION_NAME: &str = "foreground_subagent";
+    const DELIVERED: &str = "delivered";
+
     struct Fixture {
         temp_dir: TempDir,
         manager: Arc<SessionManager>,
@@ -326,7 +181,7 @@ mod tests {
     }
 
     impl Fixture {
-        fn runner(&self) -> SubagentRunner {
+        fn runner(&self) -> ForegroundSubagentRunner {
             let mut agent = crate::agents::Agent::with_config(crate::agents::AgentConfig::new(
                 self.manager.clone(),
                 crate::config::permission::PermissionManager::instance(),
@@ -335,11 +190,11 @@ mod tests {
                 crate::agents::GoosePlatform::GooseCli,
             ));
             agent.set_hook_manager_for_test(crate::hooks::HookManager::default());
-            SubagentRunner::new(agent)
+            ForegroundSubagentRunner::new(agent)
         }
 
-        fn operation(&self, cancel: CancellationToken) -> SubagentOperation {
-            SubagentOperation::new(self.runner(), cancel)
+        fn operation(&self) -> SubagentOperation<Session> {
+            SubagentOperation::new(Arc::new(self.runner()))
         }
     }
 
@@ -424,17 +279,16 @@ mod tests {
     }
 
     async fn run_step(
-        operation: &SubagentOperation,
-        fixture: &Fixture,
+        operation: &SubagentOperation<Session>,
+        manager: &SessionManager,
+        session_id: &str,
         emit: &Emitter,
     ) -> Result<(Session, OperationResult<GooseEffect>)> {
-        let parent = fixture
-            .manager
-            .get_session(&fixture.parent_id, true)
-            .await?;
-        let conversation = parent.conversation.as_ref().unwrap();
-        let result = operation.run(&parent, conversation, emit).await?;
-        Ok((parent, result))
+        let session = manager.get_session(session_id, true).await?;
+        let conversation = session.conversation.as_ref().unwrap();
+        let result =
+            Operation::<Session, GooseEffect>::run(operation, &session, conversation, emit).await?;
+        Ok((session, result))
     }
 
     fn delivery_effects(result: OperationResult<GooseEffect>) -> Vec<GooseEffect> {
@@ -445,49 +299,49 @@ mod tests {
         step.effects
     }
 
-    fn delivered(effect: &GooseEffect) -> (&str, &Message) {
+    fn delivered(effect: &GooseEffect) -> &str {
         let GooseEffect::Conversation(ConversationEffect::AppendMessage(message)) = effect else {
             panic!("expected a subagent delivery");
         };
-        let subagent_id = message
+        message
             .metadata
             .operation_note(OPERATION_NAME, DELIVERED)
             .and_then(serde_json::Value::as_str)
-            .expect("the result message records its subagent");
-        (subagent_id, message)
+            .expect("the result message records its subagent")
     }
 
-    #[test]
-    fn pending_subagents_are_delegated_until_delivered_or_cancelled() {
+    #[tokio::test]
+    async fn successful_foreground_delegations_are_the_started_subagents() -> Result<()> {
+        let fixture = fixture().await?;
         let mut failed_delegate = delegate_result("failed", true);
         failed_delegate.is_error = Some(true);
-        let mut messages = vec![
+        let conversation = Conversation::new_unvalidated(vec![
+            Message::user().with_text("delegate"),
             Message::user()
-                .with_tool_response("a", Ok(delegate_result("delivered", true)))
+                .with_tool_response("a", Ok(delegate_result("first", true)))
                 .with_tool_response("b", Ok(delegate_result("background", false)))
                 .with_tool_response("c", Ok(failed_delegate))
                 .with_tool_response(
                     "d",
                     Ok(CallToolResult::success(vec![ContentBlock::text("done")])),
                 ),
-            delegate_message("cancelled"),
-            delegate_message("waiting"),
-            delegate_message("waiting"),
-        ];
-        assert_eq!(
-            pending_subagents(&messages),
-            ["delivered", "cancelled", "waiting"]
-        );
+            delegate_message("second"),
+        ]);
+        let mut parent = fixture
+            .manager
+            .get_session(&fixture.parent_id, false)
+            .await?;
+        let runner = fixture.runner();
 
-        for (key, id) in [
-            (DELIVERED, serde_json::json!("delivered")),
-            (CANCELLED, serde_json::json!(["cancelled"])),
-        ] {
-            let mut message = Message::user().with_text(key);
-            message.metadata.set_operation_note(OPERATION_NAME, key, id);
-            messages.push(message);
-        }
-        assert_eq!(pending_subagents(&messages), ["waiting"]);
+        assert_eq!(
+            runner.started_subagent_session_ids(&parent, &conversation),
+            ["first", "second"]
+        );
+        parent.session_type = SessionType::SubAgent;
+        assert!(runner
+            .started_subagent_session_ids(&parent, &conversation)
+            .is_empty());
+        Ok(())
     }
 
     async fn save_final_output(manager: &SessionManager, subagent_id: &str) -> Result<()> {
@@ -548,7 +402,7 @@ mod tests {
         let cancel = CancellationToken::new();
         let (tx, mut rx) = mpsc::unbounded_channel();
         let emit = Emitter::new(tx, cancel.clone());
-        let operation = fixture.operation(cancel);
+        let operation = fixture.operation();
 
         for (expected_id, expected_notice) in [
             (
@@ -560,17 +414,11 @@ mod tests {
                 format!("Subagent {second_subagent_id} completed"),
             ),
         ] {
-            let (parent, result) = run_step(&operation, &fixture, &emit).await?;
+            let (parent, result) =
+                run_step(&operation, &fixture.manager, &fixture.parent_id, &emit).await?;
             let mut effects = delivery_effects(result);
             assert_eq!(effects.len(), 1);
-            let (subagent_id, message) = delivered(&effects[0]);
-            assert_eq!(subagent_id, expected_id);
-            assert!(!message.is_user_visible());
-            assert!(message.is_agent_visible());
-            assert_eq!(
-                message.as_concat_text(),
-                format!("Subagent {subagent_id} completed: {{\"summary\":\"done\"}}")
-            );
+            assert_eq!(delivered(&effects[0]), expected_id);
             assert_eq!(notices(&mut rx), vec![expected_notice]);
             fixture
                 .manager
@@ -599,20 +447,13 @@ mod tests {
         let cancel = CancellationToken::new();
         let (tx, mut rx) = mpsc::unbounded_channel();
         let emit = Emitter::new(tx, cancel.clone());
-        let operation = fixture.operation(cancel);
-        let (_, result) = run_step(&operation, &fixture, &emit).await?;
+        let operation = fixture.operation();
+        let (_, result) = run_step(&operation, &fixture.manager, &fixture.parent_id, &emit).await?;
         let effects = delivery_effects(result);
         assert_eq!(effects.len(), 1);
-        let (subagent_id, message) = delivered(&effects[0]);
-        assert_eq!(subagent_id, fixture.subagent_id);
-        assert!(!message.is_user_visible());
-        assert!(message.is_agent_visible());
+        assert_eq!(delivered(&effects[0]), fixture.subagent_id);
 
         let reason = format!("Subagent {} has no saved recipe", fixture.subagent_id);
-        assert_eq!(
-            message.as_concat_text(),
-            format!("Subagent {} failed: {reason}", fixture.subagent_id)
-        );
         assert_eq!(
             notices(&mut rx),
             vec![
@@ -624,150 +465,27 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn delivers_each_subagent_in_its_own_step() -> Result<()> {
+    async fn stop_shows_the_output_of_the_last_finished_subagent() -> Result<()> {
         let fixture = fixture().await?;
-        let second_subagent_id =
-            add_delegated_subagent(&fixture.manager, &fixture.temp_dir, &fixture.parent_id).await?;
         let cancel = CancellationToken::new();
-        let (tx, _rx) = mpsc::unbounded_channel();
-        let emit = Emitter::new(tx, cancel.clone());
-        let operation = fixture.operation(cancel);
-
-        let mut delivered_ids = Vec::new();
-        for _ in 0..2 {
-            let (parent, result) = run_step(&operation, &fixture, &emit).await?;
-            let mut effects = delivery_effects(result);
-            assert_eq!(effects.len(), 1);
-            delivered_ids.push(delivered(&effects[0]).0.to_string());
-            fixture
-                .manager
-                .apply_effects(&parent, &mut effects, &emit)
-                .await?;
-        }
-        delivered_ids.sort();
-        let mut expected = vec![fixture.subagent_id.clone(), second_subagent_id];
-        expected.sort();
-        assert_eq!(delivered_ids, expected);
-
-        let (parent, result) = run_step(&operation, &fixture, &emit).await?;
-        assert!(matches!(result, OperationResult::NotApplicable));
-        let hidden_deliveries = parent
-            .conversation
-            .unwrap()
-            .messages()
-            .iter()
-            .filter(|message| !message.is_user_visible())
-            .count();
-        assert_eq!(hidden_deliveries, 2);
-        Ok(())
-    }
-
-    struct DropFlag(Arc<std::sync::atomic::AtomicBool>);
-
-    impl Drop for DropFlag {
-        fn drop(&mut self) {
-            self.0.store(true, std::sync::atomic::Ordering::SeqCst);
-        }
-    }
-
-    async fn cancel_step(
-        operation: &SubagentOperation,
-        fixture: &Fixture,
-        emit: &Emitter,
-    ) -> Result<(Session, Vec<GooseEffect>)> {
-        let parent = fixture
-            .manager
-            .get_session(&fixture.parent_id, true)
-            .await?;
-        let effects = operation
-            .finalize_cancellation(&parent, parent.conversation.as_ref().unwrap(), emit)
-            .await;
-        Ok((parent, effects))
-    }
-
-    fn cancelled_ids(effect: &GooseEffect) -> Vec<String> {
-        let GooseEffect::Conversation(ConversationEffect::AppendMessage(message)) = effect else {
-            panic!("expected a cancelled note");
-        };
-        assert!(!message.is_user_visible());
-        assert!(message.is_agent_visible());
-        serde_json::from_value(
-            message
-                .metadata
-                .operation_note(OPERATION_NAME, CANCELLED)
-                .cloned()
-                .expect("the note records the cancelled subagents"),
-        )
-        .unwrap()
-    }
-
-    #[tokio::test]
-    async fn stop_delivers_finished_subagents_and_aborts_running_ones_without_waiting() -> Result<()>
-    {
-        let fixture = fixture().await?;
-        let stopped_subagent_id =
-            add_delegated_subagent(&fixture.manager, &fixture.temp_dir, &fixture.parent_id).await?;
-        let running_subagent_id =
-            add_delegated_subagent(&fixture.manager, &fixture.temp_dir, &fixture.parent_id).await?;
-        let cancel = CancellationToken::new();
-        let (tx, _rx) = mpsc::unbounded_channel();
-        let emit = Emitter::new(tx, cancel.clone());
-        let operation = fixture.operation(cancel.clone());
-        let aborted = Arc::new(std::sync::atomic::AtomicBool::new(false));
-        let finished = {
-            let mut running = operation.running.lock().await;
-            let mut finished = Vec::new();
-            for (subagent_id, outcome) in [
-                (
-                    &fixture.subagent_id,
-                    SubagentOutcome::Completed("done".to_string()),
-                ),
-                (&stopped_subagent_id, SubagentOutcome::Cancelled),
-            ] {
-                let handle = running.tasks.spawn(async { outcome });
-                running
-                    .subagent_ids
-                    .insert(handle.id(), subagent_id.clone());
-                finished.push(handle);
-            }
-            let flag = DropFlag(aborted.clone());
-            let stuck = running.tasks.spawn(async move {
-                let _flag = flag;
-                std::future::pending::<SubagentOutcome>().await
-            });
-            running
-                .subagent_ids
-                .insert(stuck.id(), running_subagent_id.clone());
-            finished
-        };
-        while !finished.iter().all(|handle| handle.is_finished()) {
-            tokio::task::yield_now().await;
-        }
         cancel.cancel();
+        let (tx, mut rx) = mpsc::unbounded_channel();
+        let emit = Emitter::new(tx, cancel);
 
-        let (_, effects) = tokio::time::timeout(
-            Duration::from_secs(5),
-            cancel_step(&operation, &fixture, &emit),
-        )
-        .await??;
+        fixture.runner().on_subagent_finished(
+            &fixture.subagent_id,
+            &SubagentOutcome::Completed(r#"{"summary":"done"}"#.to_string()),
+            0,
+            &emit,
+        );
 
-        assert_eq!(effects.len(), 2);
-        let (subagent_id, message) = delivered(&effects[0]);
-        assert_eq!(subagent_id, fixture.subagent_id);
         assert_eq!(
-            message.as_concat_text(),
-            format!("Subagent {} completed: done", fixture.subagent_id)
+            notices(&mut rx),
+            [format!(
+                "Subagent {} completed\n\ndone",
+                fixture.subagent_id
+            )]
         );
-        assert_eq!(
-            cancelled_ids(&effects[1]),
-            [stopped_subagent_id, running_subagent_id]
-        );
-        tokio::time::timeout(Duration::from_secs(5), async {
-            while !aborted.load(std::sync::atomic::Ordering::SeqCst) {
-                tokio::task::yield_now().await;
-            }
-        })
-        .await?;
         Ok(())
     }
 
@@ -776,48 +494,20 @@ mod tests {
         let fixture = fixture().await?;
         let cancel = CancellationToken::new();
         cancel.cancel();
-        let runner = fixture.runner();
-        let SubagentStart::Started { run, .. } = runner
-            .start(
-                &fixture.parent_id,
-                &fixture.subagent_id,
-                cancel,
-                mpsc::unbounded_channel().0,
-            )
+        let parent = fixture
+            .manager
+            .get_session(&fixture.parent_id, false)
+            .await?;
+        let updates: Vec<_> = fixture
+            .runner()
+            .start(&parent, &fixture.subagent_id, cancel)
             .await
-        else {
-            panic!("expected the subagent to start");
-        };
-        assert!(matches!(run.await, SubagentOutcome::Cancelled));
-        Ok(())
-    }
-
-    #[tokio::test]
-    async fn cancel_marks_only_undelivered_subagents_once() -> Result<()> {
-        let fixture = fixture().await?;
-        save_final_output(&fixture.manager, &fixture.subagent_id).await?;
-        let unfinished_subagent_id =
-            add_delegated_subagent(&fixture.manager, &fixture.temp_dir, &fixture.parent_id).await?;
-        let cancel = CancellationToken::new();
-        let (tx, _rx) = mpsc::unbounded_channel();
-        let emit = Emitter::new(tx, cancel.clone());
-        let operation = fixture.operation(cancel.clone());
-        let (parent, result) = run_step(&operation, &fixture, &emit).await?;
-        fixture
-            .manager
-            .apply_effects(&parent, &mut delivery_effects(result), &emit)
-            .await?;
-
-        cancel.cancel();
-        let (parent, mut effects) = cancel_step(&operation, &fixture, &emit).await?;
-        assert_eq!(effects.len(), 1);
-        assert_eq!(cancelled_ids(&effects[0]), [unfinished_subagent_id]);
-        fixture
-            .manager
-            .apply_effects(&parent, &mut effects, &emit)
-            .await?;
-        let (_, effects) = cancel_step(&fixture.operation(cancel), &fixture, &emit).await?;
-        assert!(effects.is_empty());
+            .collect()
+            .await;
+        assert!(matches!(
+            updates.last(),
+            Some(SubagentUpdate::Finished(SubagentOutcome::Cancelled))
+        ));
         Ok(())
     }
 
@@ -835,20 +525,18 @@ mod tests {
         let cancel = CancellationToken::new();
         let (tx, mut rx) = mpsc::unbounded_channel();
         let emit = Emitter::new(tx, cancel.clone());
-        let operation = fixture.operation(cancel);
+        let operation = fixture.operation();
 
         for _ in 0..2 {
-            let copy = fixture.manager.get_session(&copy.id, true).await?;
-            let result = operation
-                .run(&copy, copy.conversation.as_ref().unwrap(), &emit)
-                .await?;
+            let (copy, result) = run_step(&operation, &fixture.manager, &copy.id, &emit).await?;
             fixture
                 .manager
                 .apply_effects(&copy, &mut delivery_effects(result), &emit)
                 .await?;
         }
 
-        let copy = fixture.manager.get_session(&copy.id, true).await?;
+        let (copy, result) = run_step(&operation, &fixture.manager, &copy.id, &emit).await?;
+        assert!(matches!(result, OperationResult::NotApplicable));
         let hidden = hidden_texts(copy.conversation.as_ref().unwrap());
         assert_eq!(
             hidden[0],
@@ -858,7 +546,6 @@ mod tests {
             )
         );
         assert!(hidden[1].starts_with("Subagent missing-subagent failed: "));
-        assert!(pending_subagents(copy.conversation.as_ref().unwrap().messages()).is_empty());
         assert!(notices(&mut rx)
             .iter()
             .all(|notice| !notice.starts_with("Running subagent")));

crates/goose/src/agents/state_machine/tests/pipeline.rs

@@ -18,7 +18,7 @@ use crate::agents::state_machine::{
     messages_since_kickoff, Emitter, GooseEffect, RunStatus, StateMachine, SteerQueue,
 };
 use crate::agents::state_machine_builder::{build_state_machine, StateMachineConfig};
-use crate::agents::subagent_handler::SubagentRunner;
+use crate::agents::subagent_handler::ForegroundSubagentRunner;
 use crate::agents::AgentEvent;
 use crate::config::permission::{PermissionLevel, PermissionManager};
 use crate::config::GooseMode;
@@ -124,7 +124,7 @@ impl TestPipeline {
             tool_inspection_manager: &self.tool_inspection_manager,
             hook_manager: self.hook_manager.clone(),
             steer_queue: self.steer_queue.clone(),
-            subagent_runner: SubagentRunner::new(Arc::clone(&self.subagent_host)),
+            subagent_runner: ForegroundSubagentRunner::new(Arc::clone(&self.subagent_host)),
             retry_timeout: std::time::Duration::from_secs(1),
             on_failure_timeout: std::time::Duration::from_secs(1),
             stop_hook_block_cap: self.stop_hook_block_cap,

crates/goose/src/agents/state_machine_builder.rs

@@ -13,7 +13,7 @@ use crate::agents::state_machine::{
     StateMachine, StatusOperation, SteerOperation, SteerQueue, Step, StopHookOperation,
     SubagentOperation, ToolApprovalOperation, ToolExecutionOperation, ToolPairCompactionOperation,
 };
-use crate::agents::subagent_handler::SubagentRunner;
+use crate::agents::subagent_handler::ForegroundSubagentRunner;
 use crate::hooks::HookManager;
 use crate::providers::base::Provider;
 use crate::session::{Session, SessionManager};
@@ -32,7 +32,7 @@ pub(crate) struct StateMachineConfig<'a> {
     pub(crate) tool_inspection_manager: &'a ToolInspectionManager,
     pub(crate) hook_manager: HookManager,
     pub(crate) steer_queue: SteerQueue,
-    pub(crate) subagent_runner: SubagentRunner,
+    pub(crate) subagent_runner: ForegroundSubagentRunner,
     pub(crate) retry_timeout: Duration,
     pub(crate) on_failure_timeout: Duration,
     pub(crate) stop_hook_block_cap: u32,
@@ -78,7 +78,7 @@ pub(crate) fn build_state_machine(
             Arc::clone(&extension_lease),
         )),
         Arc::new(RemainingToolCallsOperation::new(hook_manager.clone())),
-        Arc::new(SubagentOperation::new(subagent_runner, cancel.clone())),
+        Arc::new(SubagentOperation::new(Arc::new(subagent_runner))),
         Arc::new(SteerOperation::new(steer_queue, hook_manager.clone())),
         Arc::new(BangShellOperation::new()),
     ];

crates/goose/src/agents/subagent_handler.rs

@@ -13,13 +13,17 @@ use crate::{
     session::{Session, SessionManager, SessionType},
 };
 use anyhow::{anyhow, Result};
-use futures::future::BoxFuture;
+use futures::stream::BoxStream;
 use futures::StreamExt;
+use goose_agent::subagent::{SubagentOutcome, SubagentUpdate};
 use rmcp::model::Role;
 use serde::Serialize;
+use std::future::Future;
 use std::sync::Arc;
 use tokio::sync::mpsc;
 use tokio_util::sync::CancellationToken;
+use tokio_util::task::AbortOnDropHandle;
+use tracing::Instrument;
 
 #[derive(Serialize)]
 pub struct SubagentPromptContext {
@@ -77,26 +81,12 @@ pub(crate) async fn prepare_subagent(agent: &Agent, session: &Session) -> Result
     })
 }
 
-pub(crate) enum SubagentOutcome {
-    Completed(String),
-    Failed(String),
-    Cancelled,
-}
-
-pub(crate) enum SubagentStart {
-    HasOutcome(SubagentOutcome),
-    Started {
-        task: Option<String>,
-        run: BoxFuture<'static, SubagentOutcome>,
-    },
-}
-
 #[derive(Clone)]
-pub(crate) struct SubagentRunner {
+pub(crate) struct ForegroundSubagentRunner {
     agent: Arc<Agent>,
 }
 
-impl SubagentRunner {
+impl ForegroundSubagentRunner {
     pub(crate) fn new(agent: Arc<Agent>) -> Self {
         Self { agent }
     }
@@ -105,47 +95,24 @@ impl SubagentRunner {
         &self.agent.config.session_manager
     }
 
-    pub(crate) async fn start(
+    pub(crate) async fn load(&self, subagent_id: &str) -> Result<Session> {
+        self.session_manager().get_session(subagent_id, true).await
+    }
+
+    pub(crate) fn spawn(
         &self,
-        parent_id: &str,
-        subagent_id: &str,
+        subagent: Session,
         cancel: CancellationToken,
-        forward: mpsc::UnboundedSender<Message>,
-    ) -> SubagentStart {
-        let subagent = match self.session_manager().get_session(subagent_id, true).await {
-            Ok(subagent) => subagent,
-            Err(error) => {
-                return SubagentStart::HasOutcome(SubagentOutcome::Failed(error.to_string()))
-            }
-        };
-        if subagent.session_type != SessionType::SubAgent
-            || subagent.parent_session_id.as_deref() != Some(parent_id)
-        {
-            return SubagentStart::HasOutcome(SubagentOutcome::Failed(
-                "it does not belong to this session".to_string(),
-            ));
-        }
-        if let Some(output) = subagent
-            .conversation
-            .as_ref()
-            .and_then(|conversation| FinalOutputTool::successful_output(conversation.messages()))
-        {
-            return SubagentStart::HasOutcome(SubagentOutcome::Completed(output));
-        }
-        SubagentStart::Started {
-            task: subagent
-                .recipe
-                .as_ref()
-                .and_then(|recipe| recipe.prompt.clone()),
-            run: Box::pin(self.clone().run(subagent, cancel, forward)),
-        }
+    ) -> BoxStream<'static, SubagentUpdate> {
+        let (to_parent, from_subagent) = mpsc::unbounded_channel();
+        spawn_run(self.clone().run(subagent, cancel, to_parent), from_subagent)
     }
 
     async fn run_to_end(
         &self,
         subagent: &Session,
         cancel: CancellationToken,
-        forward: mpsc::UnboundedSender<Message>,
+        to_parent: &mpsc::UnboundedSender<Message>,
     ) -> Result<()> {
         let parent_id = subagent
             .parent_session_id
@@ -159,7 +126,7 @@ impl SubagentRunner {
         while let Some(event) = events.next().await {
             if let AgentEvent::Message(message) = event? {
                 if asks_for_a_decision(&message) {
-                    let _ = forward.send(forwarded(
+                    let _ = to_parent.send(forwarded(
                         message,
                         parent_id,
                         &subagent.id,
@@ -175,14 +142,14 @@ impl SubagentRunner {
         self,
         subagent: Session,
         cancel: CancellationToken,
-        forward: mpsc::UnboundedSender<Message>,
+        to_parent: mpsc::UnboundedSender<Message>,
     ) -> SubagentOutcome {
         let ending = SubagentEnding {
             agent: Arc::clone(&self.agent),
             subagent_id: subagent.id.clone(),
             ended: false,
         };
-        let run_result = self.run_to_end(&subagent, cancel.clone(), forward).await;
+        let run_result = self.run_to_end(&subagent, cancel.clone(), &to_parent).await;
         ending.end().await;
         let stopped = cancel.is_cancelled();
         let subagent = match self.session_manager().get_session(&subagent.id, true).await {
@@ -210,6 +177,22 @@ impl SubagentRunner {
     }
 }
 
+fn spawn_run(
+    run: impl Future<Output = SubagentOutcome> + Send + 'static,
+    mut from_subagent: mpsc::UnboundedReceiver<Message>,
+) -> BoxStream<'static, SubagentUpdate> {
+    let task = AbortOnDropHandle::new(tokio::spawn(run.in_current_span()));
+    Box::pin(async_stream::stream! {
+        while let Some(message) = from_subagent.recv().await {
+            yield SubagentUpdate::ToParent(message);
+        }
+        yield SubagentUpdate::Finished(
+            task.await
+                .unwrap_or_else(|error| SubagentOutcome::Failed(error.to_string())),
+        );
+    })
+}
+
 /// Aborting the parent's turn drops a running subagent's task, so the session is also ended
 /// from `Drop`.
 struct SubagentEnding {
@@ -332,9 +315,77 @@ async fn build_subagent_prompt(
 
 #[cfg(test)]
 mod tests {
-    use super::failure_reason;
+    use std::time::Duration;
+
+    use futures::StreamExt;
+
+    use super::{failure_reason, spawn_run};
     use crate::agents::state_machine::MAX_TURNS_MESSAGE;
     use crate::conversation::message::Message;
+    use goose_agent::subagent::{SubagentOutcome, SubagentUpdate};
+    use tokio::sync::{mpsc, oneshot};
+
+    #[tokio::test]
+    async fn dropping_a_started_run_aborts_its_task() {
+        let (alive, aborted) = oneshot::channel::<()>();
+        let (to_parent, from_subagent) = mpsc::unbounded_channel();
+        let run = spawn_run(
+            async move {
+                let _alive = alive;
+                let _to_parent = to_parent;
+                std::future::pending::<SubagentOutcome>().await
+            },
+            from_subagent,
+        );
+
+        drop(run);
+
+        let ended = tokio::time::timeout(Duration::from_secs(5), aborted).await;
+        assert!(ended.expect("the task was not aborted").is_err());
+    }
+
+    #[tokio::test]
+    async fn a_run_reports_what_it_forwarded_then_its_outcome() {
+        let (to_parent, from_subagent) = mpsc::unbounded_channel();
+        let run = spawn_run(
+            async move {
+                to_parent
+                    .send(Message::assistant().with_text("may I run cargo test?"))
+                    .unwrap();
+                SubagentOutcome::Completed("done".to_string())
+            },
+            from_subagent,
+        );
+
+        let updates: Vec<_> = run.collect().await;
+
+        assert!(matches!(
+            updates.as_slice(),
+            [
+                SubagentUpdate::ToParent(_),
+                SubagentUpdate::Finished(SubagentOutcome::Completed(_))
+            ]
+        ));
+    }
+
+    #[tokio::test]
+    async fn a_panicking_run_is_reported_as_failed() {
+        let (to_parent, from_subagent) = mpsc::unbounded_channel::<Message>();
+        let run = spawn_run(
+            async move {
+                let _to_parent = to_parent;
+                panic!("subagent crashed")
+            },
+            from_subagent,
+        );
+
+        let updates: Vec<_> = run.collect().await;
+
+        assert!(matches!(
+            updates.as_slice(),
+            [SubagentUpdate::Finished(SubagentOutcome::Failed(_))]
+        ));
+    }
 
     #[test]
     fn failure_reason_describes_how_the_subagent_stopped() {