mirror of
https://github.com/block/buzz.git
synced 2026-11-03 16:01:35 +00:00
feat(acp): agent timeout resilience — idle margin, tool-call reset, death notices, keepalive
The 20s margin between the 600s max shell timeout and the 620s idle timeout caused spurious session kills during legitimate long-running tool calls. This changeset addresses the problem from four angles: - Raise DEFAULT_IDLE_TIMEOUT_SECS from 620 to 900 (300s margin) - Reset idle clock explicitly on tool_call session updates with observability logging - Post a visible channel message (death notice) when a session ends due to idle timeout or unexpected agent exit - Emit a keepalive session update every 30s while waiting on the LLM provider response, preventing idle timeout during slow completions Co-authored-by: Will Pfleger <pfleger.will@gmail.com> Signed-off-by: Will Pfleger <pfleger.will@gmail.com>
This commit is contained in:
parent
877048d68f
commit
903ab774bf
@@ -739,7 +739,7 @@ impl AcpClient {
|
||||
if let Some(method) = msg.get("method").and_then(|v| v.as_str()) {
|
||||
match method {
|
||||
"session/update" => {
|
||||
self.handle_session_update(&msg);
|
||||
let _ = self.handle_session_update(&msg);
|
||||
}
|
||||
"session/request_permission" => {
|
||||
self.handle_permission_request(&msg).await?;
|
||||
@@ -852,7 +852,14 @@ impl AcpClient {
|
||||
// Dispatch notifications and agent-initiated requests.
|
||||
if let Some(method) = msg.get("method").and_then(|v| v.as_str()) {
|
||||
match method {
|
||||
"session/update" => self.handle_session_update(&msg),
|
||||
"session/update" => {
|
||||
if self.handle_session_update(&msg) {
|
||||
// Tool call started — explicitly reset idle clock.
|
||||
// The agent will be silent while the tool executes.
|
||||
tracing::debug!("idle clock reset: tool call started");
|
||||
idle_deadline = Instant::now() + idle_timeout;
|
||||
}
|
||||
}
|
||||
"session/request_permission" => {
|
||||
self.handle_permission_request(&msg).await?;
|
||||
}
|
||||
@@ -893,7 +900,10 @@ impl AcpClient {
|
||||
/// Log a `session/update` notification via tracing.
|
||||
///
|
||||
/// The discriminator field is `sessionUpdate` (not `type`) per the ACP schema.
|
||||
fn handle_session_update(&self, msg: &serde_json::Value) {
|
||||
/// Returns `true` if the update indicates a tool call started, signaling that
|
||||
/// the idle clock should be explicitly reset (the agent will be silent while
|
||||
/// the tool executes).
|
||||
fn handle_session_update(&self, msg: &serde_json::Value) -> bool {
|
||||
let update = &msg["params"]["update"];
|
||||
let update_type = update
|
||||
.get("sessionUpdate")
|
||||
@@ -905,6 +915,7 @@ impl AcpClient {
|
||||
if let Some(text) = update["content"]["text"].as_str() {
|
||||
tracing::info!(target: "acp::stream", "{text}");
|
||||
}
|
||||
false
|
||||
}
|
||||
"tool_call" => {
|
||||
let title = update
|
||||
@@ -916,6 +927,7 @@ impl AcpClient {
|
||||
.and_then(|v| v.as_str())
|
||||
.unwrap_or("unknown");
|
||||
tracing::info!(target: "acp::tool", "tool_call: {title} ({kind})");
|
||||
true
|
||||
}
|
||||
"tool_call_update" => {
|
||||
let tool_id = update
|
||||
@@ -924,14 +936,17 @@ impl AcpClient {
|
||||
.unwrap_or("?");
|
||||
let status = update.get("status").and_then(|v| v.as_str()).unwrap_or("?");
|
||||
tracing::info!(target: "acp::tool", "tool_call_update: {tool_id} → {status}");
|
||||
false
|
||||
}
|
||||
"plan" => {
|
||||
tracing::info!(target: "acp::plan", "plan update received");
|
||||
false
|
||||
}
|
||||
"agent_thought_chunk" => {
|
||||
if let Some(text) = update["content"]["text"].as_str() {
|
||||
tracing::debug!(target: "acp::thought", "{text}");
|
||||
}
|
||||
false
|
||||
}
|
||||
"available_commands_update" => {
|
||||
// Advertised slash commands (ACP slash-commands extension).
|
||||
@@ -946,9 +961,11 @@ impl AcpClient {
|
||||
names.len(),
|
||||
names.join(", ")
|
||||
);
|
||||
false
|
||||
}
|
||||
other => {
|
||||
tracing::debug!(target: "acp::update", "session/update: {other}");
|
||||
false
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -20,9 +20,11 @@ use crate::filter::SubscriptionRule;
|
||||
///
|
||||
/// Sized for slow turns where the agent may go silent on its outer ACP channel
|
||||
/// while running long sub-tools (e.g. a sprout-agent running another agent, or
|
||||
/// codex/claude doing multi-minute single tool calls). 600s working budget +
|
||||
/// 20s buffer. Override via `--idle-timeout` / `SPROUT_ACP_IDLE_TIMEOUT`.
|
||||
pub(crate) const DEFAULT_IDLE_TIMEOUT_SECS: u64 = 620;
|
||||
/// codex/claude doing multi-minute single tool calls). 900s gives 300s of
|
||||
/// breathing room above the 600s max shell timeout, so legitimate long-running
|
||||
/// tool calls don't race the idle deadline.
|
||||
/// Override via `--idle-timeout` / `SPROUT_ACP_IDLE_TIMEOUT`.
|
||||
pub(crate) const DEFAULT_IDLE_TIMEOUT_SECS: u64 = 900;
|
||||
|
||||
// ── Errors ────────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
@@ -1717,6 +1717,7 @@ async fn tokio_main() -> Result<()> {
|
||||
&respawn_tx,
|
||||
&mut respawn_tasks,
|
||||
observer.clone(),
|
||||
&relay,
|
||||
) == LoopAction::Exit
|
||||
{
|
||||
break;
|
||||
@@ -1987,6 +1988,7 @@ fn handle_prompt_result(
|
||||
respawn_tx: &mpsc::Sender<RespawnResult>,
|
||||
respawn_tasks: &mut tokio::task::JoinSet<()>,
|
||||
observer: Option<observer::ObserverHandle>,
|
||||
relay: &HarnessRelay,
|
||||
) -> LoopAction {
|
||||
let before = pool.task_map().len();
|
||||
let agent_index = result.agent.index;
|
||||
@@ -2076,10 +2078,26 @@ fn handle_prompt_result(
|
||||
outcome = outcome_label,
|
||||
"agent_returned — respawning"
|
||||
);
|
||||
emit_turn_error(match outcome_label {
|
||||
let death_message = match outcome_label {
|
||||
"exited" => "Agent process exited unexpectedly",
|
||||
_ => "Agent timed out",
|
||||
});
|
||||
_ => "Agent session timed out due to inactivity",
|
||||
};
|
||||
emit_turn_error(death_message);
|
||||
|
||||
// Post a visible death notice to the channel so humans know why
|
||||
// the agent went silent.
|
||||
if let Some(ch) = channel_id {
|
||||
match relay.build_death_notice(ch, death_message) {
|
||||
Ok(event) => {
|
||||
if let Err(e) = relay.try_publish_event(event) {
|
||||
tracing::warn!("failed to publish death notice: {e}");
|
||||
}
|
||||
}
|
||||
Err(e) => {
|
||||
tracing::warn!("failed to build death notice: {e}");
|
||||
}
|
||||
}
|
||||
}
|
||||
let index = result.agent.index;
|
||||
let slot_history = &mut crash_history[index];
|
||||
if !spawn_respawn_task(
|
||||
|
||||
@@ -748,6 +748,19 @@ impl HarnessRelay {
|
||||
Ok(event)
|
||||
}
|
||||
|
||||
/// Build a channel message (kind:9) for death notices — posted when the
|
||||
/// agent session ends due to timeout or unexpected exit.
|
||||
pub fn build_death_notice(&self, channel_id: Uuid, message: &str) -> Result<Event, RelayError> {
|
||||
use sprout_core::kind::KIND_STREAM_MESSAGE;
|
||||
|
||||
let h_tag = Tag::parse(["h", &channel_id.to_string()])
|
||||
.map_err(|e| RelayError::AuthFailed(e.to_string()))?;
|
||||
let event = EventBuilder::new(Kind::Custom(KIND_STREAM_MESSAGE as u16), message)
|
||||
.tags([h_tag])
|
||||
.sign_with_keys(&self.keys)?;
|
||||
Ok(event)
|
||||
}
|
||||
|
||||
/// Set the startup watermark timestamp (Finding #22).
|
||||
///
|
||||
/// Call this once after `connect()` with the Unix timestamp captured just
|
||||
|
||||
@@ -95,6 +95,27 @@ impl RunCtx<'_> {
|
||||
biased;
|
||||
_ = self.cancel.changed() => return Ok(StopReason::Cancelled),
|
||||
r = self.llm.complete(self.cfg, self.system_prompt, self.history, &tools) => r?,
|
||||
_ = async {
|
||||
// Keepalive ticker: emit a lightweight session update every 30s
|
||||
// while waiting on the LLM provider. This resets the ACP harness
|
||||
// idle clock so long provider responses don't trigger timeout.
|
||||
let mut interval = tokio::time::interval(std::time::Duration::from_secs(30));
|
||||
interval.tick().await; // first tick fires immediately — skip it
|
||||
loop {
|
||||
interval.tick().await;
|
||||
tracing::debug!("llm keepalive tick");
|
||||
wire::send(
|
||||
self.wire,
|
||||
wire::session_update(
|
||||
self.session_id,
|
||||
json!({
|
||||
"sessionUpdate": "keepalive",
|
||||
}),
|
||||
),
|
||||
)
|
||||
.await;
|
||||
}
|
||||
} => unreachable!(),
|
||||
};
|
||||
|
||||
// Record provider-reported input usage so the next loop iteration's
|
||||
|
||||
Reference in New Issue
Block a user