fix(orchestration): surface async sub-agent failures/awaiting-user in chat (#4896)

src/openhuman/agent_orchestration/background_completions.rs

@@ -12,6 +12,21 @@
 use std::collections::{HashMap, HashSet, VecDeque};
 use std::sync::{Mutex, OnceLock};
 
+/// Terminal disposition of a finished background sub-agent. Drives distinct
+/// rendering in [`build_batched_notice`] so a failed / awaiting-input async
+/// sub-agent surfaces in chat as such instead of being dropped or mistaken for a
+/// success (#4896).
+#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
+pub(crate) enum BackgroundAgentOutcome {
+    /// Ran to a usable result (or partial progress framed as such).
+    #[default]
+    Completed,
+    /// The child errored before producing a result.
+    Failed,
+    /// The child paused asking the user a question and was not continued.
+    AwaitingInput,
+}
+
 /// One finished background sub-agent's deliverable result.
 #[derive(Clone, Debug, PartialEq, Eq)]
 pub(crate) struct CompletedBackgroundAgent {
@@ -24,6 +39,9 @@ pub(crate) struct CompletedBackgroundAgent {
     /// Parent chat thread id to stream the delivery turn into (captured at
     /// spawn). `None` for a headless spawn with no originating thread.
     pub(crate) parent_thread_id: Option<String>,
+    /// Terminal disposition — success, failure, or awaiting-user — so delivery
+    /// can render failures/awaiting distinctly (#4896).
+    pub(crate) outcome: BackgroundAgentOutcome,
 }
 
 /// Upper bound on the cancelled-thread tombstone set. A thread id is a one-shot
@@ -111,13 +129,38 @@ pub(crate) fn record_completion(
     agent_id: impl Into<String>,
     summary: impl Into<String>,
     parent_thread_id: Option<String>,
+) {
+    record_outcome(
+        parent_session,
+        task_id,
+        agent_id,
+        summary,
+        parent_thread_id,
+        BackgroundAgentOutcome::Completed,
+    );
+}
+
+/// Record a finished background sub-agent carrying an explicit terminal
+/// [`BackgroundAgentOutcome`]. This is the general enqueue behind
+/// [`record_completion`] (success) and the [`record_failure`] /
+/// [`record_awaiting_input`] framing helpers, so a failed or awaiting-input
+/// async sub-agent is delivered back into chat too — not only successes (#4896).
+/// Same tombstone / idempotency guarantees as [`record_completion`].
+pub(crate) fn record_outcome(
+    parent_session: impl Into<String>,
+    task_id: impl Into<String>,
+    agent_id: impl Into<String>,
+    summary: impl Into<String>,
+    parent_thread_id: Option<String>,
+    outcome: BackgroundAgentOutcome,
 ) {
     let parent_session = parent_session.into();
     let entry = CompletedBackgroundAgent {
         task_id: task_id.into(),
         agent_id: agent_id.into(),
         summary: summary.into(),
         parent_thread_id,
+        outcome,
     };
     let mut state = queue()
         .lock()
@@ -151,6 +194,55 @@ pub(crate) fn record_completion(
     pending.push(entry);
 }
 
+/// Queue a **failed** async sub-agent for chat delivery (#4896). The summary is
+/// framed with the `[SUBAGENT_FAILED]` envelope the parent agent is prompted to
+/// relay, so the user learns the delegated task errored instead of the turn
+/// silently finalizing on "Accepted". Enqueues via [`record_outcome`], so it
+/// rides the same idle-gated `background_delivery` path as a success.
+pub(crate) fn record_failure(
+    parent_session: impl Into<String>,
+    task_id: impl Into<String>,
+    agent_id: impl Into<String>,
+    error: &str,
+    parent_thread_id: Option<String>,
+) {
+    let summary =
+        format!("[SUBAGENT_FAILED] the async sub-agent errored before producing a result: {error}");
+    record_outcome(
+        parent_session,
+        task_id,
+        agent_id,
+        summary,
+        parent_thread_id,
+        BackgroundAgentOutcome::Failed,
+    );
+}
+
+/// Queue an **awaiting-user** async sub-agent for chat delivery (#4896). A
+/// detached child that pauses to ask a question will not continue on its own, so
+/// the framed `[SUBAGENT_NEEDS_INPUT]` notice is delivered back into chat for the
+/// parent agent to relay to (or answer for) the user.
+pub(crate) fn record_awaiting_input(
+    parent_session: impl Into<String>,
+    task_id: impl Into<String>,
+    agent_id: impl Into<String>,
+    question: &str,
+    parent_thread_id: Option<String>,
+) {
+    let summary = format!(
+        "[SUBAGENT_NEEDS_INPUT] the async sub-agent paused to ask the user a question and will \
+         not continue on its own: {question}"
+    );
+    record_outcome(
+        parent_session,
+        task_id,
+        agent_id,
+        summary,
+        parent_thread_id,
+        BackgroundAgentOutcome::AwaitingInput,
+    );
+}
+
 /// Is anything waiting to be delivered for this session? Cheap idle-loop check.
 pub(crate) fn has_pending(parent_session: &str) -> bool {
     queue()
@@ -272,18 +364,34 @@ pub(crate) fn build_batched_notice(completed: &[CompletedBackgroundAgent]) -> Op
     let mut out = String::new();
     out.push_str(&format!(
         "[{n} background sub-agent{} finished while you were busy. Review each result \
-         below and present what is relevant to the user. Each is tagged with its \
-         sub-agent process id.]\n",
+         below — including any that FAILED or NEED INPUT — and present what is relevant \
+         to the user (never silently drop a failure or an awaiting-input pause). Each is \
+         tagged with its sub-agent process id.]\n",
         if n == 1 { "" } else { "s" },
     ));
     for c in completed {
+        // Distinct tag per terminal outcome so a failure / awaiting-input result
+        // is not presented as a normal completion (#4896).
+        let (tag, empty_fallback) = match c.outcome {
+            BackgroundAgentOutcome::Completed => {
+                ("background_agent_result", "(no output reported)")
+            }
+            BackgroundAgentOutcome::Failed => (
+                "background_agent_failure",
+                "(failed with no detail reported)",
+            ),
+            BackgroundAgentOutcome::AwaitingInput => (
+                "background_agent_needs_input",
+                "(the sub-agent paused awaiting user input)",
+            ),
+        };
         let summary = if c.summary.trim().is_empty() {
-            "(no output reported)"
+            empty_fallback
         } else {
             c.summary.trim()
         };
         out.push_str(&format!(
-            "\n<background_agent_result id=\"{}\" agent=\"{}\">\n{}\n</background_agent_result>\n",
+            "\n<{tag} id=\"{}\" agent=\"{}\">\n{}\n</{tag}>\n",
             c.task_id, c.agent_id, summary,
         ));
     }
@@ -311,6 +419,19 @@ mod tests {
             agent_id: agent.into(),
             summary: summary.into(),
             parent_thread_id: Some("thread-1".into()),
+            outcome: BackgroundAgentOutcome::Completed,
+        }
+    }
+
+    fn c_outcome(
+        task: &str,
+        agent: &str,
+        summary: &str,
+        outcome: BackgroundAgentOutcome,
+    ) -> CompletedBackgroundAgent {
+        CompletedBackgroundAgent {
+            outcome,
+            ..c(task, agent, summary)
         }
     }
 
@@ -531,4 +652,122 @@ mod tests {
             "collected tombstone must stay bounded, got {len}"
         );
     }
+
+    // ── #4896: failure / awaiting-user delivery ─────────────────────────────
+
+    #[test]
+    fn record_failure_queues_a_framed_failure_for_delivery() {
+        let _guard = test_guard();
+        let s = "sess-fail";
+        record_failure(
+            s,
+            "sub-f",
+            "researcher",
+            "provider 500: inference failed",
+            Some("thread-F".into()),
+        );
+        // The failure rode the SAME queue successes use → it will be delivered.
+        assert_eq!(pending_count(s), 1);
+        let drained = take_pending(s);
+        assert_eq!(drained[0].outcome, BackgroundAgentOutcome::Failed);
+        assert!(drained[0].summary.starts_with("[SUBAGENT_FAILED]"));
+        assert!(drained[0]
+            .summary
+            .contains("provider 500: inference failed"));
+    }
+
+    #[test]
+    fn record_awaiting_input_queues_a_framed_needs_input_for_delivery() {
+        let _guard = test_guard();
+        let s = "sess-await";
+        record_awaiting_input(
+            s,
+            "sub-a",
+            "researcher",
+            "Which repo should I open the PR against?",
+            Some("thread-A".into()),
+        );
+        assert_eq!(pending_count(s), 1);
+        let drained = take_pending(s);
+        assert_eq!(drained[0].outcome, BackgroundAgentOutcome::AwaitingInput);
+        assert!(drained[0].summary.starts_with("[SUBAGENT_NEEDS_INPUT]"));
+        assert!(drained[0].summary.contains("Which repo"));
+    }
+
+    #[test]
+    fn notice_renders_failure_and_awaiting_with_distinct_tags() {
+        let notice = build_batched_notice(&[
+            c_outcome(
+                "sub-ok",
+                "researcher",
+                "all good",
+                BackgroundAgentOutcome::Completed,
+            ),
+            c_outcome(
+                "sub-bad",
+                "researcher",
+                "[SUBAGENT_FAILED] boom",
+                BackgroundAgentOutcome::Failed,
+            ),
+            c_outcome(
+                "sub-ask",
+                "researcher",
+                "[SUBAGENT_NEEDS_INPUT] which repo?",
+                BackgroundAgentOutcome::AwaitingInput,
+            ),
+        ])
+        .expect("non-empty batch");
+
+        // The header now tells the agent to surface failures / awaiting-input.
+        assert!(notice.contains("FAILED or NEED INPUT"));
+        // Each outcome renders under its own tag so a failure is not presented
+        // as a normal completion.
+        assert!(notice.contains("<background_agent_result id=\"sub-ok\" agent=\"researcher\">"));
+        assert!(notice.contains("<background_agent_failure id=\"sub-bad\" agent=\"researcher\">"));
+        assert!(notice.contains("[SUBAGENT_FAILED] boom"));
+        assert!(
+            notice.contains("<background_agent_needs_input id=\"sub-ask\" agent=\"researcher\">")
+        );
+        assert!(notice.contains("[SUBAGENT_NEEDS_INPUT] which repo?"));
+    }
+
+    #[test]
+    fn empty_summary_fallback_is_outcome_specific() {
+        let failed = build_batched_notice(&[c_outcome(
+            "sub-e",
+            "r",
+            "   ",
+            BackgroundAgentOutcome::Failed,
+        )])
+        .unwrap();
+        assert!(failed.contains("(failed with no detail reported)"));
+
+        let awaiting = build_batched_notice(&[c_outcome(
+            "sub-e",
+            "r",
+            "",
+            BackgroundAgentOutcome::AwaitingInput,
+        )])
+        .unwrap();
+        assert!(awaiting.contains("(the sub-agent paused awaiting user input)"));
+    }
+
+    #[test]
+    fn record_outcome_preserves_the_outcome_through_a_drain() {
+        // Guards the requeue path (background_delivery::requeue re-enqueues via
+        // record_outcome): a failed batch that fails delivery must not be
+        // downgraded to a success on retry.
+        let _guard = test_guard();
+        let s = "sess-preserve";
+        record_outcome(
+            s,
+            "sub-p",
+            "researcher",
+            "[SUBAGENT_FAILED] x",
+            None,
+            BackgroundAgentOutcome::Failed,
+        );
+        let drained = take_pending(s);
+        assert_eq!(drained[0].outcome, BackgroundAgentOutcome::Failed);
+    }
 }