From 4838b94b33d48dfa992b44a96c8173842f47926c Mon Sep 17 00:00:00 2001 From: M3gA-Mind Date: Wed, 15 Jul 2026 19:22:39 +0530 Subject: [PATCH 1/3] fix(orchestration): surface async sub-agent failures and awaiting-user in chat Async sub-agent delegation re-injected only successful outcomes; the AwaitingUser and Err(...) branches published events but never called record_completion, so background_delivery (which drains only completions) could never surface them and the turn finalized on "Accepted" with the failure silently lost. Route failure and awaiting-user outcomes through the delivery path with a user-readable summary (task id + reason), and render outcome-specific items in background_delivery so the user learns a delegated task failed or needs input. Success tag/wording preserved; existing tests unaffected. Closes #4896 --- .../background_completions.rs | 247 +++++++++++++++++- .../background_delivery.rs | 5 +- .../tools/spawn_async_subagent.rs | 24 ++ 3 files changed, 271 insertions(+), 5 deletions(-) diff --git a/src/openhuman/agent_orchestration/background_completions.rs b/src/openhuman/agent_orchestration/background_completions.rs index 8e7be4ca7c..e3338a60e5 100644 --- a/src/openhuman/agent_orchestration/background_completions.rs +++ b/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, + /// 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,6 +129,30 @@ pub(crate) fn record_completion( agent_id: impl Into, summary: impl Into, parent_thread_id: Option, +) { + 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, + task_id: impl Into, + agent_id: impl Into, + summary: impl Into, + parent_thread_id: Option, + outcome: BackgroundAgentOutcome, ) { let parent_session = parent_session.into(); let entry = CompletedBackgroundAgent { @@ -118,6 +160,7 @@ pub(crate) fn record_completion( 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, + task_id: impl Into, + agent_id: impl Into, + error: &str, + parent_thread_id: Option, +) { + 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, + task_id: impl Into, + agent_id: impl Into, + question: &str, + parent_thread_id: Option, +) { + 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\n{}\n\n", + "\n<{tag} id=\"{}\" agent=\"{}\">\n{}\n\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("")); + assert!(notice.contains("")); + assert!(notice.contains("[SUBAGENT_FAILED] boom")); + assert!( + notice.contains("") + ); + 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); + } } diff --git a/src/openhuman/agent_orchestration/background_delivery.rs b/src/openhuman/agent_orchestration/background_delivery.rs index b1b0e0df21..de41d1bfc6 100644 --- a/src/openhuman/agent_orchestration/background_delivery.rs +++ b/src/openhuman/agent_orchestration/background_delivery.rs @@ -107,12 +107,15 @@ fn plan_delivery(session: &str) -> Option) { for c in batch { - background_completions::record_completion( + // Preserve the terminal outcome on requeue so a failed / awaiting-input + // result isn't downgraded to a success when a delivery turn fails (#4896). + background_completions::record_outcome( session, c.task_id, c.agent_id, c.summary, c.parent_thread_id, + c.outcome, ); } } diff --git a/src/openhuman/agent_orchestration/tools/spawn_async_subagent.rs b/src/openhuman/agent_orchestration/tools/spawn_async_subagent.rs index beac656df3..4e710e0b3f 100644 --- a/src/openhuman/agent_orchestration/tools/spawn_async_subagent.rs +++ b/src/openhuman/agent_orchestration/tools/spawn_async_subagent.rs @@ -663,6 +663,18 @@ impl Tool for SpawnAsyncSubagentTool { let error = format!( "async sub-agent requested user clarification and was not continued: {question}" ); + // #4896: a detached child that pauses for input won't + // continue on its own — queue a framed notice so the + // parent chat learns the delegated task needs input, + // instead of finalizing silently on "Accepted". Rides the + // same idle-gated background_delivery path as a success. + crate::openhuman::agent_orchestration::background_completions::record_awaiting_input( + background_parent_session.clone(), + outcome.task_id.clone(), + outcome.agent_id.clone(), + question, + background_parent_thread_id.clone(), + ); crate::openhuman::agent_orchestration::subagent_events::publish_subagent_failed( background_parent_session, outcome.task_id.clone(), @@ -699,6 +711,18 @@ impl Tool for SpawnAsyncSubagentTool { let _ = status_tx.send(SubagentStatus::Failed { error: error.clone(), }); + // #4896: a detached child that errors previously only + // published an event — nothing reached chat, so the parent + // turn finalized on "Accepted" and the failure was lost. + // Queue a framed failure notice so background_delivery + // surfaces it as a follow-up chat turn. + crate::openhuman::agent_orchestration::background_completions::record_failure( + background_parent_session.clone(), + background_task_id.clone(), + background_agent_id.clone(), + &error, + background_parent_thread_id.clone(), + ); crate::openhuman::agent_orchestration::subagent_events::publish_subagent_failed( background_parent_session, background_task_id.clone(), From 45bd52314478fb4168c53a31d68c7fe42e0d04fb Mon Sep 17 00:00:00 2001 From: M3gA-Mind Date: Thu, 16 Jul 2026 03:31:11 +0530 Subject: [PATCH 2/3] fix(orchestration): drain delivery on subagent failure/awaiting-user (#4896) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The background delivery handler only scheduled a drain on SubagentCompleted, so a subagent that FAILED (or paused awaiting user) after the parent turn was already idle never triggered a delivery attempt — the pending failure sat undelivered until some unrelated later turn, leaving the chat stuck on the original "Accepted" response (Codex P1). Match all three subagent terminal events (completed | failed | awaiting-user), each of which carries `parent_session`, in the same debounced-drain arm. Add a handler test that accepts every terminal event. --- .../background_delivery.rs | 57 ++++++++++++++++++- 1 file changed, 55 insertions(+), 2 deletions(-) diff --git a/src/openhuman/agent_orchestration/background_delivery.rs b/src/openhuman/agent_orchestration/background_delivery.rs index c46d54ae1d..c8e3dde228 100644 --- a/src/openhuman/agent_orchestration/background_delivery.rs +++ b/src/openhuman/agent_orchestration/background_delivery.rs @@ -70,8 +70,17 @@ impl EventHandler for BackgroundDeliveryHandler { busy().lock().expect("busy poisoned").remove(session_id); schedule_delivery(session_id.clone(), Duration::from_millis(300)); } - DomainEvent::SubagentCompleted { parent_session, .. } => { - // Debounce so a burst of completions batches into a single turn. + // Any subagent terminal state — completed, failed, or awaiting-user — + // can arrive after the parent turn already went idle. Schedule a + // debounced drain for all three so the pending result is delivered + // promptly instead of sitting until some unrelated later turn. Only + // `SubagentCompleted` used to trigger a drain, so a failure (or an + // awaiting-user pause) after the parent turn went idle left the chat + // stuck on the original "Accepted" response (#4896). Debounce so a + // burst batches into a single turn. + DomainEvent::SubagentCompleted { parent_session, .. } + | DomainEvent::SubagentFailed { parent_session, .. } + | DomainEvent::SubagentAwaitingUser { parent_session, .. } => { schedule_delivery(parent_session.clone(), DEBOUNCE); } _ => {} @@ -304,4 +313,48 @@ mod tests { .await; assert!(!is_busy(&sid)); } + + #[tokio::test] + async fn handler_accepts_every_subagent_terminal_event() { + // #4896 regression: a subagent that finishes after the parent turn is + // idle must still schedule a drain. Previously only `SubagentCompleted` + // was matched, so `SubagentFailed` / `SubagentAwaitingUser` fell through + // to `_ => {}` and their pending results were never delivered. The + // handler must now accept all three terminal events (each schedules a + // debounced drain via the shared arm); the drain behaviour itself is + // covered by the `plan_delivery` tests above. + let h = BackgroundDeliveryHandler; + let sid = "bd-subagent-terminal".to_string(); + let events = [ + DomainEvent::SubagentCompleted { + parent_session: sid.clone(), + task_id: "t".into(), + agent_id: "a".into(), + elapsed_ms: 0, + output_chars: 0, + iterations: 0, + }, + DomainEvent::SubagentFailed { + parent_session: sid.clone(), + task_id: "t".into(), + agent_id: "a".into(), + error: "boom".into(), + }, + DomainEvent::SubagentAwaitingUser { + parent_session: sid.clone(), + task_id: "t".into(), + agent_id: "a".into(), + question: "?".into(), + }, + ]; + for ev in events { + h.handle(&ev).await; + // A subagent terminal event schedules delivery for the parent but + // must never mark the parent session itself busy. + assert!( + !is_busy(&sid), + "subagent terminal event must not mark parent busy" + ); + } + } } From 834c8237bacf67f68b4af792469d32bbc8e2afc5 Mon Sep 17 00:00:00 2001 From: M3gA-Mind Date: Thu, 16 Jul 2026 04:08:15 +0530 Subject: [PATCH 3/3] test(orchestration): assert terminal events actually schedule a drain (#4896) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CodeRabbit: the previous handler test only asserted `!is_busy`, which also holds if the events fall through to `_ => {}`, so it couldn't catch a regression to the completed-only behaviour. Replace it with a behavioural test on the paused clock: queue a headless result per session, fire each terminal event, advance past the debounce, and assert the pending result was consumed — proving the arm scheduled a drain for SubagentFailed / SubagentAwaitingUser (not just SubagentCompleted). --- .../background_delivery.rs | 97 +++++++++++-------- 1 file changed, 57 insertions(+), 40 deletions(-) diff --git a/src/openhuman/agent_orchestration/background_delivery.rs b/src/openhuman/agent_orchestration/background_delivery.rs index c8e3dde228..3cf2423b1c 100644 --- a/src/openhuman/agent_orchestration/background_delivery.rs +++ b/src/openhuman/agent_orchestration/background_delivery.rs @@ -314,47 +314,64 @@ mod tests { assert!(!is_busy(&sid)); } - #[tokio::test] - async fn handler_accepts_every_subagent_terminal_event() { - // #4896 regression: a subagent that finishes after the parent turn is - // idle must still schedule a drain. Previously only `SubagentCompleted` - // was matched, so `SubagentFailed` / `SubagentAwaitingUser` fell through - // to `_ => {}` and their pending results were never delivered. The - // handler must now accept all three terminal events (each schedules a - // debounced drain via the shared arm); the drain behaviour itself is - // covered by the `plan_delivery` tests above. + #[tokio::test(start_paused = true)] + async fn every_subagent_terminal_event_schedules_a_drain() { + // #4896 regression: EVERY subagent terminal event must schedule a drain + // for the parent — not just `SubagentCompleted`. Before the fix, + // `SubagentFailed` / `SubagentAwaitingUser` fell through to `_ => {}`, so + // a failure/pause recorded after the parent turn went idle was never + // delivered. Prove behaviour, not just acceptance: queue a headless + // result (no thread → drains without a delivery sink) per session, fire + // the event, advance past the debounce, and assert the pending item was + // consumed. The paused clock elapses the debounce with no wall-clock wait. let h = BackgroundDeliveryHandler; - let sid = "bd-subagent-terminal".to_string(); - let events = [ - DomainEvent::SubagentCompleted { - parent_session: sid.clone(), - task_id: "t".into(), - agent_id: "a".into(), - elapsed_ms: 0, - output_chars: 0, - iterations: 0, - }, - DomainEvent::SubagentFailed { - parent_session: sid.clone(), - task_id: "t".into(), - agent_id: "a".into(), - error: "boom".into(), - }, - DomainEvent::SubagentAwaitingUser { - parent_session: sid.clone(), - task_id: "t".into(), - agent_id: "a".into(), - question: "?".into(), - }, - ]; - for ev in events { - h.handle(&ev).await; - // A subagent terminal event schedules delivery for the parent but - // must never mark the parent session itself busy. - assert!( - !is_busy(&sid), - "subagent terminal event must not mark parent busy" - ); + + background_completions::record_completion("bd-term-completed", "t", "a", "s", None); + background_completions::record_completion("bd-term-failed", "t", "a", "s", None); + background_completions::record_completion("bd-term-awaiting", "t", "a", "s", None); + + h.handle(&DomainEvent::SubagentCompleted { + parent_session: "bd-term-completed".into(), + task_id: "t".into(), + agent_id: "a".into(), + elapsed_ms: 0, + output_chars: 0, + iterations: 0, + }) + .await; + h.handle(&DomainEvent::SubagentFailed { + parent_session: "bd-term-failed".into(), + task_id: "t".into(), + agent_id: "a".into(), + error: "boom".into(), + }) + .await; + h.handle(&DomainEvent::SubagentAwaitingUser { + parent_session: "bd-term-awaiting".into(), + task_id: "t".into(), + agent_id: "a".into(), + question: "?".into(), + }) + .await; + + // Advance the virtual clock past the debounce so every scheduled drain + // runs; the headless `try_deliver` completes synchronously (no sink). + tokio::time::sleep(DEBOUNCE + Duration::from_millis(50)).await; + for _ in 0..8 { + tokio::task::yield_now().await; } + + assert!( + !background_completions::has_pending("bd-term-completed"), + "SubagentCompleted must schedule a drain that consumes the pending result" + ); + assert!( + !background_completions::has_pending("bd-term-failed"), + "SubagentFailed must schedule a drain (regression #4896)" + ); + assert!( + !background_completions::has_pending("bd-term-awaiting"), + "SubagentAwaitingUser must schedule a drain (regression #4896)" + ); } }