Skip to content

fix(channel): retry retrigger relay sends to avoid dropped final replies #582

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
brendandebeasi wants to merge 2 commits into
base: main
Choose a base branch
from
Open

fix(channel): retry retrigger relay sends to avoid dropped final replies #582

Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
d7b5c0d
fix(channel): retry retrigger relay sends and preserve delivery state
May 2, 2026
1010830
fix(channel): drop dead retry loop, keep replied_flag fix on fallback…
May 10, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
108 changes: 94 additions & 14 deletions src/agent/channel.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,7 @@ use std::collections::HashSet;
use std::sync::{Arc, Weak};
use tokio::sync::broadcast;
use tokio::sync::{RwLock, mpsc};
use tokio::time::{Duration, sleep};

/// Shared cache of in-flight worker transcript steps, keyed by worker ID.
pub type LiveWorkerTranscripts =
Expand Down Expand Up @@ -2959,8 +2960,7 @@ impl Channel {
})
}

/// Send outbound text and record send metrics.
async fn send_outbound_text(&self, text: String, error_context: &str) {
async fn send_outbound_text(&self, text: String, error_context: &str) -> bool {
match self.send_routed(OutboundResponse::Text(text)).await {
Ok(()) => {
#[cfg(feature = "metrics")]
Expand All @@ -2971,6 +2971,7 @@ impl Channel {
.with_label_values(&[&self.deps.agent_id, channel_type])
.inc();
}
true
}
Err(error) => {
#[cfg(feature = "metrics")]
Expand All @@ -2982,8 +2983,57 @@ impl Channel {
.inc();
}
tracing::error!(%error, channel_id = %self.id, "{error_context}");
false
}
}
}

async fn send_outbound_text_with_retry(
&self,
text: String,
error_context: &str,
max_attempts: usize,
) -> bool {
let attempts = max_attempts.max(1);
for attempt in 1..=attempts {
if self
.send_outbound_text(text.clone(), error_context)
.await
{
if attempt > 1 {
tracing::info!(
channel_id = %self.id,
attempt,
attempts,
"outbound relay succeeded after retry"
);
}
return true;
}

if attempt < attempts {
let delay_ms = match attempt {
1 => 250,
2 => 1_000,
_ => 2_000,
};
tracing::warn!(
channel_id = %self.id,
attempt,
attempts,
delay_ms,
"outbound relay failed; retrying"
);
sleep(Duration::from_millis(delay_ms)).await;
}
}

tracing::warn!(
channel_id = %self.id,
attempts,
"outbound relay failed after retries"
);
false
}
Comment thread
coderabbitai[bot] marked this conversation as resolved.
Outdated

/// Dispatch the LLM result: send fallback text, log errors, clean up typing.
Expand Down Expand Up @@ -3061,11 +3111,24 @@ impl Channel {
self.state
.conversation_logger
.log_bot_message(&self.state.channel_id, &final_text);
self.send_outbound_text(
final_text,
"failed to send retrigger fallback reply",
)
.await;
let delivered = self
.send_outbound_text_with_retry(
final_text,
"failed to send retrigger fallback reply",
3,
)
.await;
if delivered {
replied_flag.store(true, std::sync::atomic::Ordering::Relaxed);
} else {
let _ = self
.send_outbound_text_with_retry(
"Delivery issue: your background result is preserved. Send 'continue' to replay it.".to_string(),
"failed to send relay failure backup notice",
1,
)
.await;
}

@coderabbitai coderabbitai Bot May 2, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🛠️ Refactor suggestion | 🟠 Major | 🏗️ Heavy lift

Add an injected-failure test for the new retrigger fallback state machine.

These paths now couple retry timing, replied_flag mutation, preserved-result replay, and the backup "continue" notice, but the provided validation is still just cargo check. Please add a targeted test that forces controlled send failures and asserts both branches: success after retry and failure-after-all-attempts preserving replay state.

As per coding guidelines, "For changes in async/stateful paths (worker lifecycle, cancellation, retrigger, recall cache behavior), include explicit race/terminal-state reasoning in the PR summary and run targeted tests in addition to just just gate-pr."

Also applies to: 3192-3209, 3265-3270

🤖 Prompt for AI Agents 
Verify each finding against the current code and only fix it if needed.

In `@src/agent/channel.rs` around lines 3114 - 3131, Add an injected-failure
unit/integration test that exercises the retrigger fallback state machine around
send_outbound_text_with_retry: simulate controlled send failures and time/retry
behavior to assert both branches — (1) success after retry sets replied_flag to
true and does not enqueue the preserved replay, and (2) permanent failure leaves
replied_flag false and sends the backup "continue" notice while preserving the
background result for replay. Use dependency injection/mocking or an injectable
send function used by send_outbound_text_with_retry to force 0, partial, and
full-failure scenarios, advance any timers/async delays as needed, and assert
final state (replied_flag, preserved-result replay state, and that the correct
backup notice was sent).

}
}
} else {
Expand Down Expand Up @@ -3126,11 +3189,24 @@ impl Channel {
self.state
.conversation_logger
.log_bot_message(&self.state.channel_id, &final_text);
self.send_outbound_text(
final_text,
"failed to send retrigger fallback reply",
)
.await;
let delivered = self
.send_outbound_text_with_retry(
final_text,
"failed to send retrigger fallback reply",
3,
)
.await;
if delivered {
replied_flag.store(true, std::sync::atomic::Ordering::Relaxed);
} else {
let _ = self
.send_outbound_text_with_retry(
"Delivery issue: your background result is preserved. Send 'continue' to replay it.".to_string(),
"failed to send relay failure backup notice",
1,
)
.await;
}
}
}
} else {
Expand Down Expand Up @@ -3186,8 +3262,12 @@ impl Channel {
Some(self.agent_display_name()),
tool_calls_json,
);
self.send_outbound_text(final_text, "failed to send fallback reply")
.await;
self.send_outbound_text_with_retry(
final_text,
"failed to send fallback reply",
2,
)
.await;
}
}

Expand Down