rho-coding-agent 2.9.0

//! Delegated-agent questionnaire and completion coordination.

use futures_util::FutureExt;
use ratatui::DefaultTerminal;
use tokio::sync::oneshot;

use super::subagent_delivery::{TurnBoundaryBatch, TurnBoundaryDelivery};
use crate::display_transcript::DisplayTranscript;

use super::{
    questionnaire::QuestionnaireResponseChannel, App, ComposerMode, Entry, InteractiveRuntime,
    PendingSubagentQuestionnaire, QuestionAnswerRequest, QuestionnaireReply, TurnOutcome,
};

#[derive(Clone, Copy)]
enum ParentActivity {
    Idle,
    Working(&'static str),
}

pub(super) enum SubagentCompletionTurn {
    NoDelivery,
    PendingConfirmation,
    Completed(TurnOutcome),
}

/// Already-scheduled provider requests may carry informational notices; notices
/// alone must not start another request, including at a completion checkpoint.
#[derive(Clone, Copy)]
enum BoundaryTrigger {
    ScheduledTurn,
    Notification,
}

fn subagent_completion_changed(outcome: &SubagentCompletionTurn) -> bool {
    !matches!(outcome, SubagentCompletionTurn::NoDelivery)
}

impl App {
    /// The runtime waits here, rather than racing a ToolFinished-driven steer.
    /// An empty final reply hands later arrivals back to idle delivery.
    pub(super) async fn deliver_running_boundary(
        &mut self,
        request: rho_sdk::BoundaryInputRequest,
        agent: &mut InteractiveRuntime,
    ) -> Option<(String, DisplayTranscript)> {
        if request.session_id() != agent.session_id() {
            request.respond(None).await;
            return None;
        }
        let trigger = match request.boundary() {
            rho_sdk::InputBoundary::BeforeProvider => BoundaryTrigger::ScheduledTurn,
            rho_sdk::InputBoundary::BeforeCompletion => BoundaryTrigger::Notification,
            // Unknown checkpoints must not let routine notices buy inference.
            _ => BoundaryTrigger::Notification,
        };
        let captured = {
            let _snapshot = crate::app::notification_delivery::lock();
            let batch = self.take_turn_boundary_batch_locked(agent, trigger);
            if batch.is_empty() {
                self.restore_turn_boundary_batch(agent, batch);
                // Send synchronously while publishers are excluded so later
                // arrivals belong to idle delivery. Await ack outside the gate.
                Err(request.respond(None))
            } else {
                Ok((batch, request))
            }
        };
        let (batch, request) = match captured {
            Ok(captured) => captured,
            Err(receipt) => {
                receipt.await;
                return None;
            }
        };
        let delivery = batch.prepare(agent);
        let input = rho_sdk::UserInput::text(format!(
            "[runtime notifications for session {} run {}]\nThis is internal background context, not a human message or a new task. Incorporate unseen findings into the ongoing work.\n\n{}",
            request.session_id(),
            request.run_id(),
            delivery.model,
        ));
        let model_message = rho_sdk::model::Message::User(input.blocks().to_vec());
        if request.respond(Some(input)).await {
            let transcript = delivery.transcript;
            agent.record_boundary_display(model_message, transcript.display_message());
            self.subagent_inbox
                .commit_delivered_notices(delivery.batch.notice_count());
            Some((delivery.model, transcript))
        } else {
            self.restore_turn_boundary_batch(agent, delivery.batch);
            None
        }
    }

    /// Wakes an idle session with a turn for finished background subagents.
    /// Real prompt turns drain these notifications themselves, while active
    /// goals deliver them before evaluating the goal again.
    pub(super) async fn poll_subagent_completions(
        &mut self,
        terminal: &mut DefaultTerminal,
        agent: &mut InteractiveRuntime,
    ) -> anyhow::Result<bool> {
        self.subagent_inbox.drain();
        if !self.should_deliver_idle_subagent_completions() {
            return Ok(false);
        }
        // Completions and child notices share one idle delivery path. Opening a
        // confirmation modal is itself a visible state change and needs redraw.
        let outcome = self.run_subagent_completion_turn(terminal, agent).await?;
        Ok(subagent_completion_changed(&outcome))
    }

    /// Surfaces delegated questionnaires when the parent can take user input.
    pub(super) async fn poll_subagent_questionnaires(
        &mut self,
        session_id: &rho_sdk::SessionId,
    ) -> anyhow::Result<bool> {
        let mut changed = self.subagent_inbox.drain();
        changed |= self.subagent_inbox.discard_stale(session_id);
        changed |= self
            .finish_pending_subagent_questionnaire(ParentActivity::Idle)
            .await?;
        if self.can_present_subagent_questionnaire() {
            changed |= self.present_next_subagent_questionnaire(session_id).await?;
        }
        Ok(changed)
    }

    /// Updates delegated questionnaire state without presenting another request.
    /// The active turn uses this while its shared interaction queue owns ordering.
    pub(super) async fn poll_running_subagent_questionnaire_state(
        &mut self,
        session_id: &rho_sdk::SessionId,
    ) -> anyhow::Result<bool> {
        let mut changed = self.subagent_inbox.drain();
        changed |= self.subagent_inbox.discard_stale(session_id);
        changed |= self
            .finish_pending_subagent_questionnaire(ParentActivity::Working("running"))
            .await?;
        Ok(changed)
    }

    /// Surfaces delegated questionnaires while a goal waits for its children.
    pub(super) async fn poll_waiting_subagent_questionnaires(
        &mut self,
        session_id: &rho_sdk::SessionId,
    ) -> anyhow::Result<bool> {
        let mut changed = self.subagent_inbox.drain();
        changed |= self.subagent_inbox.discard_stale(session_id);
        changed |= self
            .finish_pending_subagent_questionnaire(ParentActivity::Working(
                "waiting for delegated agents",
            ))
            .await?;
        if self.pending_subagent_questionnaire.is_none()
            && matches!(self.input_ui.composer(), ComposerMode::Input)
            && !self.input_ui.has_pending_draft()
        {
            changed |= self.present_next_subagent_questionnaire(session_id).await?;
        }
        Ok(changed)
    }

    /// Collects everything owed to the model at a turn boundary: background
    /// completions, delegated-child notices, workflow notifications, and
    /// finished process-tool jobs.
    ///
    /// Returns the joined model prompt, display summary, and the drained batch
    /// so callers can restore on setup failure, or `None` when nothing is
    /// pending. Real prompt turns fold this into the outgoing message; an idle
    /// parent sends completions and action requests as a turn of its own,
    /// carrying any queued informational notices with them. Informational
    /// notices alone never create an idle turn. Active turns reply to an SDK
    /// checkpoint instead. Removal is committed only after accepted delivery.
    pub(super) fn collect_turn_boundary_prompts(
        &mut self,
        agent: &mut InteractiveRuntime,
    ) -> Option<TurnBoundaryDelivery> {
        self.collect_boundary_prompts(agent, BoundaryTrigger::ScheduledTurn)
    }

    fn collect_boundary_prompts(
        &mut self,
        agent: &mut InteractiveRuntime,
        trigger: BoundaryTrigger,
    ) -> Option<TurnBoundaryDelivery> {
        let batch = {
            let _snapshot = crate::app::notification_delivery::lock();
            self.take_turn_boundary_batch_locked(agent, trigger)
        };
        if batch.is_empty() {
            self.restore_turn_boundary_batch(agent, batch);
            None
        } else {
            Some(batch.prepare(agent))
        }
    }

    fn take_turn_boundary_batch_locked(
        &mut self,
        agent: &mut InteractiveRuntime,
        trigger: BoundaryTrigger,
    ) -> TurnBoundaryBatch {
        let mut batch = TurnBoundaryBatch::default();
        self.subagent_inbox.drain();
        self.subagent_inbox.reconcile_observed();
        self.subagent_inbox.discard_stale(agent.session_id());
        if let Some(manager) = agent.subagents().cloned() {
            batch.subagent_notifications = manager.take_notifications(agent.session_id().as_str());
        }
        batch.workflow_notifications = agent
            .workflow_tracker()
            .take_notifications(agent.session_id().as_str());
        if let Some(processes) = agent.processes() {
            batch.process_notifications = processes.take_notifications();
        }
        #[cfg(debug_assertions)]
        if matches!(trigger, BoundaryTrigger::Notification) {
            super::smoke_injection::quiet_notice_boundary(
                self.subagent_inbox.queued_notice_count(),
            );
        }
        let deliver_notices = match trigger {
            BoundaryTrigger::ScheduledTurn => true,
            BoundaryTrigger::Notification => {
                !batch.is_empty() || self.subagent_inbox.has_parent_action_requests()
            }
        };
        if deliver_notices {
            batch.notices = self.subagent_inbox.take_notices(agent.session_id());
        }
        batch
    }

    /// Puts a drained turn-boundary batch back when provider start never began.
    pub(super) fn restore_turn_boundary_batch(
        &mut self,
        agent: &mut InteractiveRuntime,
        batch: TurnBoundaryBatch,
    ) {
        if let Some(manager) = agent.subagents() {
            manager.restore_notifications(&batch.subagent_notifications);
        }
        self.subagent_inbox.return_notices(batch.notices);
        agent
            .workflow_tracker()
            .restore_notifications(&batch.workflow_notifications);
        if let Some(processes) = agent.processes() {
            processes.restore_notifications(&batch.process_notifications);
        }
    }

    async fn finish_pending_subagent_questionnaire(
        &mut self,
        parent_activity: ParentActivity,
    ) -> anyhow::Result<bool> {
        let Some(pending) = self.pending_subagent_questionnaire.as_mut() else {
            return Ok(false);
        };
        if pending.response_tx.is_closed() {
            let pending = self
                .pending_subagent_questionnaire
                .take()
                .expect("pending questionnaire checked above");
            let composer = self.input_ui.take_composer();
            if matches!(composer, ComposerMode::Questionnaire(_)) {
                drop(composer);
                self.clear_submitted_input();
            } else {
                self.input_ui.set_composer(composer);
            }
            self.insert_entry(&Entry::Notice(format!(
                "questionnaire for agent {} ({}) is no longer active",
                pending.run_id, pending.agent_id
            )));
            self.restore_parent_activity_after_questionnaire(parent_activity)
                .await;
            return Ok(true);
        }
        let Some(reply) = (&mut pending.reply_rx).now_or_never() else {
            return Ok(false);
        };
        let pending = self
            .pending_subagent_questionnaire
            .take()
            .expect("pending questionnaire checked above");
        match reply {
            Ok(QuestionnaireReply::Answer(response)) => {
                let _ = pending.response_tx.send(Ok(response));
                self.insert_entry(&Entry::Notice(format!(
                    "answered questionnaire for agent {} ({})",
                    pending.run_id, pending.agent_id
                )));
            }
            Ok(QuestionnaireReply::Cancelled(reason)) => {
                let message = match reason {
                    super::QuestionnaireCancelReason::UserCancelled => {
                        "delegated questionnaire cancelled by user"
                    }
                    super::QuestionnaireCancelReason::UiUnavailable => {
                        "delegated questionnaire cancelled because the UI closed"
                    }
                };
                let _ = pending.response_tx.send(Err(rho_sdk::Error::Interrupted {
                    message: message.into(),
                }));
                self.insert_entry(&Entry::Notice(format!(
                    "cancelled questionnaire for agent {} ({})",
                    pending.run_id, pending.agent_id
                )));
            }
            Err(_) => {
                let _ = pending.response_tx.send(Err(rho_sdk::Error::Interrupted {
                    message: "delegated questionnaire reply channel closed".into(),
                }));
            }
        }
        self.restore_parent_activity_after_questionnaire(parent_activity)
            .await;
        Ok(true)
    }

    async fn restore_parent_activity_after_questionnaire(
        &mut self,
        parent_activity: ParentActivity,
    ) {
        match parent_activity {
            ParentActivity::Idle => {
                self.set_status("ready");
                self.report_resting_herdr_state().await;
            }
            ParentActivity::Working(status) => {
                self.set_status(status);
                self.report_herdr_working().await;
            }
        }
    }

    pub(super) async fn present_subagent_questionnaire(
        &mut self,
        pending: crate::app::subagent_host_input::SubagentHostInputRequest,
    ) -> anyhow::Result<bool> {
        if pending.response.is_closed() {
            return Ok(false);
        }
        let (reply_tx, reply_rx) = oneshot::channel();
        let title = pending.request.title().to_string();
        self.open_questionnaire(QuestionAnswerRequest {
            request: pending.request,
            response: QuestionnaireResponseChannel::new(reply_tx),
            notice: Some(format!(
                "agent {} ({}) asks: {title}",
                pending.run_id, pending.agent_id
            )),
        })
        .await?;
        self.pending_subagent_questionnaire = Some(PendingSubagentQuestionnaire {
            run_id: pending.run_id,
            agent_id: pending.agent_id,
            reply_rx,
            response_tx: pending.response,
        });
        Ok(true)
    }

    async fn present_next_subagent_questionnaire(
        &mut self,
        session_id: &rho_sdk::SessionId,
    ) -> anyhow::Result<bool> {
        let mut changed = false;
        let pending = loop {
            let Some(pending) = self.subagent_inbox.next_questionnaire() else {
                return Ok(changed);
            };
            if pending.response.is_closed() {
                changed = true;
                continue;
            }
            if &pending.parent_session_id != session_id {
                let _ = pending.response.send(Err(rho_sdk::Error::Interrupted {
                    message: "parent session changed before the delegated questionnaire was shown"
                        .into(),
                }));
                changed = true;
                continue;
            }
            break pending;
        };
        changed |= self.present_subagent_questionnaire(pending).await?;
        Ok(changed)
    }

    fn can_present_subagent_questionnaire(&self) -> bool {
        self.pending_subagent_questionnaire.is_none()
            && matches!(self.input_ui.composer(), ComposerMode::Input)
            && !self.input_ui.has_pending_draft()
            && self.allows_idle_subagent_delivery()
            && self
                .goal
                .as_ref()
                .is_none_or(crate::tui::goal::GoalState::is_blocked)
            && self.pending.queued_prompts().is_empty()
            && self.pending.steering_prompts().is_empty()
    }

    pub(super) async fn run_subagent_completion_turn(
        &mut self,
        terminal: &mut DefaultTerminal,
        agent: &mut InteractiveRuntime,
    ) -> anyhow::Result<SubagentCompletionTurn> {
        let Some(delivery) = self.collect_boundary_prompts(agent, BoundaryTrigger::Notification)
        else {
            return Ok(SubagentCompletionTurn::NoDelivery);
        };
        // The whole drained batch is one message and one model request, no
        // matter how many runs finished while the parent was busy. The send
        // gate owns the drained batch until confirmation; provider start owns
        // restoration after that point.
        let submission = super::send_confirm::SendSubmission::turn_boundary(delivery);
        let Some(submission) = self.gate_send(submission, agent) else {
            return Ok(SubagentCompletionTurn::PendingConfirmation);
        };
        let (payload, authorization, _allow_auto_compact) = submission.into_authorized();
        let super::send_confirm::SendPayload::TurnBoundary(delivery) = payload else {
            unreachable!("subagent delivery is a turn-boundary submission");
        };
        self.run_turn_boundary_prompt_turn(delivery, authorization, terminal, agent)
            .await
            .map(SubagentCompletionTurn::Completed)
    }

    pub(super) fn should_deliver_idle_subagent_completions(&self) -> bool {
        self.allows_idle_subagent_delivery()
            && self.goal.is_none()
            && self.pending.queued_prompts().is_empty()
            && self.pending_subagent_questionnaire.is_none()
            && matches!(self.input_ui.composer(), ComposerMode::Input)
            && !self.subagent_inbox.has_queued_questionnaires()
    }
}

#[cfg(test)]
#[path = "subagent_questionnaires_tests.rs"]
mod tests;