huskies: merge 1196 bug Inactivity watchdog kills agents mid tool-call: awaited MCP calls produce no PTY output
This commit is contained in:
@@ -247,6 +247,104 @@ mod tests {
|
|||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// ── story 1196: in-flight MCP tool calls suspend the inactivity deadline ──
|
||||||
|
|
||||||
|
/// A `tool_use` block in an `assistant` message marks a call as in
|
||||||
|
/// flight; the inactivity deadline must be suspended while it is
|
||||||
|
/// outstanding (so a slow MCP tool like `run_tests` doesn't get the
|
||||||
|
/// agent killed), and must resume as soon as the matching `tool_result`
|
||||||
|
/// arrives in a `user` message.
|
||||||
|
///
|
||||||
|
/// Script: emits a tool_use, sleeps 2s (would fail a 1s timeout without
|
||||||
|
/// suspension), emits the matching tool_result, then sleeps 2s again
|
||||||
|
/// (now with no call in flight, the 1s timeout must fire).
|
||||||
|
#[tokio::test]
|
||||||
|
async fn tool_call_in_flight_suspends_then_resumes_inactivity_deadline() {
|
||||||
|
use std::os::unix::fs::PermissionsExt;
|
||||||
|
|
||||||
|
let tmp = tempfile::tempdir().unwrap();
|
||||||
|
let script = tmp.path().join("tool_call_then_silence.sh");
|
||||||
|
let body = "#!/bin/sh\n\
|
||||||
|
printf '%s\\n' '{\"type\":\"assistant\",\"message\":{\"content\":[{\"type\":\"tool_use\",\"id\":\"tool1\",\"name\":\"run_tests\",\"input\":{}}]}}'\n\
|
||||||
|
sleep 2\n\
|
||||||
|
printf '%s\\n' '{\"type\":\"user\",\"message\":{\"content\":[{\"type\":\"tool_result\",\"tool_use_id\":\"tool1\",\"content\":\"ok\"}]}}'\n\
|
||||||
|
sleep 2\n";
|
||||||
|
std::fs::write(&script, body).unwrap();
|
||||||
|
std::fs::set_permissions(&script, std::fs::Permissions::from_mode(0o755)).unwrap();
|
||||||
|
|
||||||
|
let (tx, _rx) = broadcast::channel::<AgentEvent>(64);
|
||||||
|
let (watcher_tx, _watcher_rx) = broadcast::channel::<WatcherEvent>(16);
|
||||||
|
let event_log = Arc::new(Mutex::new(Vec::new()));
|
||||||
|
|
||||||
|
let result = run_agent_pty_streaming(
|
||||||
|
"1196_story_tool_call_in_flight",
|
||||||
|
"coder-1",
|
||||||
|
"sh",
|
||||||
|
&[script.to_string_lossy().to_string()],
|
||||||
|
"--",
|
||||||
|
"/tmp",
|
||||||
|
&tx,
|
||||||
|
&event_log,
|
||||||
|
None,
|
||||||
|
1, // inactivity_timeout_secs = 1s
|
||||||
|
watcher_tx,
|
||||||
|
None,
|
||||||
|
None,
|
||||||
|
)
|
||||||
|
.await;
|
||||||
|
|
||||||
|
match result {
|
||||||
|
Err(err) => assert!(
|
||||||
|
err.contains("inactivity timeout"),
|
||||||
|
"expected an inactivity timeout error after the tool call resolved, got: {err}"
|
||||||
|
),
|
||||||
|
Ok(_) => panic!(
|
||||||
|
"agent must still be killed once the tool call resolves and \
|
||||||
|
the process falls genuinely silent again"
|
||||||
|
),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// A genuinely hung agent (no output at all, no tool call in flight)
|
||||||
|
/// must still be killed by the inactivity watchdog after the configured
|
||||||
|
/// timeout — the suspension added for in-flight tool calls must not
|
||||||
|
/// mask a real hang.
|
||||||
|
#[tokio::test]
|
||||||
|
async fn genuinely_silent_agent_with_no_in_flight_call_is_killed() {
|
||||||
|
use std::os::unix::fs::PermissionsExt;
|
||||||
|
|
||||||
|
let tmp = tempfile::tempdir().unwrap();
|
||||||
|
let script = tmp.path().join("silent.sh");
|
||||||
|
std::fs::write(&script, "#!/bin/sh\nsleep 2\n").unwrap();
|
||||||
|
std::fs::set_permissions(&script, std::fs::Permissions::from_mode(0o755)).unwrap();
|
||||||
|
|
||||||
|
let (tx, _rx) = broadcast::channel::<AgentEvent>(64);
|
||||||
|
let (watcher_tx, _watcher_rx) = broadcast::channel::<WatcherEvent>(16);
|
||||||
|
let event_log = Arc::new(Mutex::new(Vec::new()));
|
||||||
|
|
||||||
|
let result = run_agent_pty_streaming(
|
||||||
|
"1196_story_genuine_hang",
|
||||||
|
"coder-1",
|
||||||
|
"sh",
|
||||||
|
&[script.to_string_lossy().to_string()],
|
||||||
|
"--",
|
||||||
|
"/tmp",
|
||||||
|
&tx,
|
||||||
|
&event_log,
|
||||||
|
None,
|
||||||
|
1, // inactivity_timeout_secs = 1s
|
||||||
|
watcher_tx,
|
||||||
|
None,
|
||||||
|
None,
|
||||||
|
)
|
||||||
|
.await;
|
||||||
|
|
||||||
|
match result {
|
||||||
|
Err(err) => assert!(err.contains("inactivity timeout")),
|
||||||
|
Ok(_) => panic!("a genuinely silent agent with no in-flight tool call must be killed"),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn test_emit_event_writes_to_log_writer() {
|
fn test_emit_event_writes_to_log_writer() {
|
||||||
let tmp = tempfile::tempdir().unwrap();
|
let tmp = tempfile::tempdir().unwrap();
|
||||||
|
|||||||
@@ -246,14 +246,27 @@ fn run_agent_pty_blocking(
|
|||||||
// can distinguish a rate-limit exit from a genuine no-progress exit (bug 1053).
|
// can distinguish a rate-limit exit from a genuine no-progress exit (bug 1053).
|
||||||
let mut rate_limit_hard_block_seen = false;
|
let mut rate_limit_hard_block_seen = false;
|
||||||
let mut rate_limit_reset_at_captured: Option<chrono::DateTime<chrono::Utc>> = None;
|
let mut rate_limit_reset_at_captured: Option<chrono::DateTime<chrono::Utc>> = None;
|
||||||
|
// Tool-use ids from `assistant` messages that haven't yet seen a matching
|
||||||
|
// `tool_result` in a `user` message. While an MCP tool call (e.g.
|
||||||
|
// run_tests) is in flight, the server can legitimately run for many
|
||||||
|
// minutes with no PTY output at all — the CLI is blocked waiting on the
|
||||||
|
// MCP response, not hung. The inactivity deadline is suspended entirely
|
||||||
|
// while this set is non-empty, and resumes as soon as the matching
|
||||||
|
// tool_result clears the last in-flight id (story 1196).
|
||||||
|
let mut tool_calls_in_flight: std::collections::HashSet<String> =
|
||||||
|
std::collections::HashSet::new();
|
||||||
|
|
||||||
loop {
|
loop {
|
||||||
let effective_timeout = base_timeout.map(|base| {
|
let effective_timeout = if !tool_calls_in_flight.is_empty() {
|
||||||
let extra = block_until
|
None
|
||||||
.and_then(|t| (t - chrono::Utc::now()).to_std().ok())
|
} else {
|
||||||
.unwrap_or(std::time::Duration::ZERO);
|
base_timeout.map(|base| {
|
||||||
base + extra
|
let extra = block_until
|
||||||
});
|
.and_then(|t| (t - chrono::Utc::now()).to_std().ok())
|
||||||
|
.unwrap_or(std::time::Duration::ZERO);
|
||||||
|
base + extra
|
||||||
|
})
|
||||||
|
};
|
||||||
|
|
||||||
let recv_result = match effective_timeout {
|
let recv_result = match effective_timeout {
|
||||||
Some(dur) => line_rx.recv_timeout(dur),
|
Some(dur) => line_rx.recv_timeout(dur),
|
||||||
@@ -339,8 +352,31 @@ fn run_agent_pty_blocking(
|
|||||||
}
|
}
|
||||||
// Complete assistant events are skipped for content extraction
|
// Complete assistant events are skipped for content extraction
|
||||||
// because thinking and text already arrived via stream_event.
|
// because thinking and text already arrived via stream_event.
|
||||||
// The raw JSON is still forwarded as AgentJson below.
|
// The raw JSON is still forwarded as AgentJson below. A tool_use
|
||||||
"assistant" | "user" => {}
|
// block marks an MCP call as in flight so the inactivity deadline
|
||||||
|
// is suspended until its tool_result arrives (story 1196).
|
||||||
|
"assistant" => {
|
||||||
|
if let Some(blocks) = json.pointer("/message/content").and_then(|c| c.as_array()) {
|
||||||
|
for block in blocks {
|
||||||
|
if block.get("type").and_then(|t| t.as_str()) == Some("tool_use")
|
||||||
|
&& let Some(id) = block.get("id").and_then(|i| i.as_str())
|
||||||
|
{
|
||||||
|
tool_calls_in_flight.insert(id.to_string());
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
"user" => {
|
||||||
|
if let Some(blocks) = json.pointer("/message/content").and_then(|c| c.as_array()) {
|
||||||
|
for block in blocks {
|
||||||
|
if block.get("type").and_then(|t| t.as_str()) == Some("tool_result")
|
||||||
|
&& let Some(id) = block.get("tool_use_id").and_then(|i| i.as_str())
|
||||||
|
{
|
||||||
|
tool_calls_in_flight.remove(id);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
"rate_limit_event" => {
|
"rate_limit_event" => {
|
||||||
let rate_limit_info = json.get("rate_limit_info");
|
let rate_limit_info = json.get("rate_limit_info");
|
||||||
let status = rate_limit_info
|
let status = rate_limit_info
|
||||||
|
|||||||
Reference in New Issue
Block a user