scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-10-09 03:20:56 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 4

fix: keep forwarding a reused child's events after a broadcast lag

Browse source 
The subagent event forwarder left its loop on any `recv` error, including
`Lagged`. A lagged broadcast receiver stays usable, so one transient lag
silenced the child for the rest of its life while the task completed
normally and shutdown joined it without noticing. Session reuse widens
that window from a single turn to the whole parent session.

Also borrow each result's output when rendering a parent notification
instead of cloning it.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Bryan Helmkamp 2026-08-01 10:13:27 -04:00
parent 1bb70adafd
commit 8c19b1e94a
No known key found for this signature in database
1 changed files with 19 additions and 5 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

24
lib/components/fabro-agent/src/subagent.rs
View file

@ -1,3 +1,4 @@
use std::borrow::Cow;
use std::collections::{HashMap, VecDeque};
use std::sync::{Arc, Mutex, RwLock, Weak};
use std::time::Duration;
@ -6,7 +7,7 @@ use fabro_llm::types::ToolDefinition;
use fabro_types::INITIAL_SUBAGENT_GENERATION;
use fabro_util::error as util_error;
use futures::future;
use tokio::sync::{mpsc, oneshot, watch};
use tokio::sync::{broadcast, mpsc, oneshot, watch};
use tokio::task::{AbortHandle, JoinHandle};
use tokio::time::{Instant, timeout_at};
use tokio_util::sync::CancellationToken;
@ -48,9 +49,14 @@ fn format_parent_notification_batch(notifications: &[SubAgentParentNotification]
.iter()
.map(|notification| {
let (status, result) = match &notification.result {
Ok(result) if result.success => ("completed", result.output.clone()),
Ok(result) => ("failed", result.output.clone()),
Err(error) => ("failed", util_error::collect_chain(error).join(": ")),
Ok(result) if result.success => {
("completed", Cow::Borrowed(result.output.as_str()))
}
Ok(result) => ("failed", Cow::Borrowed(result.output.as_str())),
Err(error) => (
"failed",
Cow::Owned(util_error::collect_chain(error).join(": ")),
),
};
format!(
"<task-notification>\n <task-id>{}</task-id>\n <status>{status}</status>\n \
@ -621,7 +627,15 @@ impl SubAgentSupervisor {
let mut rx = session.subscribe();
let callback = Arc::clone(&self.event_callback);
Some(tokio::spawn(async move {
while let Ok(event) = rx.recv().await {
loop {
let event = match rx.recv().await {
Ok(event) => event,
// A lagged receiver stays usable, and a reused child
// forwards for the whole parent session. Giving up here
// would silence the child for the rest of its life.
Err(broadcast::error::RecvError::Lagged(_)) => continue,
Err(broadcast::error::RecvError::Closed) => break,
};
// Skip streaming / noise events
if event.event.is_streaming_noise()
|| matches!(