huskies: merge 1211 story Deterministic crash notification when the agent PTY dies mid-turn
This commit is contained in:
@@ -601,4 +601,126 @@ mod tests {
|
||||
"Expected RateLimitWarning for status=allowed, got: {evt:?}"
|
||||
);
|
||||
}
|
||||
|
||||
// ── story 1211: deterministic crash notification on mid-turn death ──────
|
||||
|
||||
/// AC1/AC3/AC4/AC5: a child killed mid-turn (inactivity watchdog, never
|
||||
/// emits a `"result"` event) produces exactly one `AgentCrashed` watcher
|
||||
/// event, carrying the exit code and the last line the child printed
|
||||
/// before it died. The watchdog-kill path used to `return Err(...)`
|
||||
/// directly from inside the receive loop, bypassing the bottom
|
||||
/// cleanup/notification code entirely — this test guards against that
|
||||
/// regression.
|
||||
#[tokio::test]
|
||||
async fn killed_mid_turn_sends_exactly_one_crash_notification() {
|
||||
use std::os::unix::fs::PermissionsExt;
|
||||
|
||||
let tmp = tempfile::tempdir().unwrap();
|
||||
let script = tmp.path().join("crash_then_hang.sh");
|
||||
std::fs::write(
|
||||
&script,
|
||||
"#!/bin/sh\nprintf '%s\\n' 'assertion failed: output.write(&bytes).is_ok()'\nsleep 5\n",
|
||||
)
|
||||
.unwrap();
|
||||
std::fs::set_permissions(&script, std::fs::Permissions::from_mode(0o755)).unwrap();
|
||||
|
||||
let (tx, _rx) = broadcast::channel::<AgentEvent>(64);
|
||||
let (watcher_tx, mut watcher_rx) = broadcast::channel::<WatcherEvent>(16);
|
||||
let event_log = Arc::new(Mutex::new(Vec::new()));
|
||||
|
||||
let result = run_agent_pty_streaming(
|
||||
"1211_story_crash",
|
||||
"coder-1",
|
||||
"sh",
|
||||
&[script.to_string_lossy().to_string()],
|
||||
"--",
|
||||
"/tmp",
|
||||
&tx,
|
||||
&event_log,
|
||||
None,
|
||||
1, // inactivity_timeout_secs = 1s
|
||||
watcher_tx,
|
||||
None,
|
||||
None,
|
||||
)
|
||||
.await;
|
||||
|
||||
assert!(
|
||||
result.is_err(),
|
||||
"a watchdog-killed agent must still return the inactivity timeout error"
|
||||
);
|
||||
|
||||
let evt = watcher_rx
|
||||
.try_recv()
|
||||
.expect("Expected exactly one AgentCrashed watcher event");
|
||||
match evt {
|
||||
WatcherEvent::AgentCrashed {
|
||||
story_id,
|
||||
agent_name,
|
||||
exit_code,
|
||||
last_error_line,
|
||||
} => {
|
||||
assert_eq!(story_id, "1211_story_crash");
|
||||
assert_eq!(agent_name, "coder-1");
|
||||
assert!(exit_code.is_some(), "exit code should be captured");
|
||||
assert_eq!(
|
||||
last_error_line.as_deref(),
|
||||
Some("assertion failed: output.write(&bytes).is_ok()"),
|
||||
"last line before death should be captured"
|
||||
);
|
||||
}
|
||||
other => panic!("Expected AgentCrashed, got: {other:?}"),
|
||||
}
|
||||
|
||||
// AC5: exactly one crash notification — no second event queued.
|
||||
assert!(
|
||||
watcher_rx.try_recv().is_err(),
|
||||
"must not emit a duplicate AgentCrashed for the same death event"
|
||||
);
|
||||
}
|
||||
|
||||
/// AC2: a turn that ends cleanly (a `"result"` event observed) followed
|
||||
/// by a normal child exit must produce zero `AgentCrashed` notifications.
|
||||
#[tokio::test]
|
||||
async fn clean_turn_end_sends_zero_crash_notifications() {
|
||||
use std::os::unix::fs::PermissionsExt;
|
||||
|
||||
let tmp = tempfile::tempdir().unwrap();
|
||||
let script = tmp.path().join("clean_result.sh");
|
||||
std::fs::write(
|
||||
&script,
|
||||
"#!/bin/sh\nprintf '%s\\n' '{\"type\":\"result\",\"subtype\":\"success\"}'\n",
|
||||
)
|
||||
.unwrap();
|
||||
std::fs::set_permissions(&script, std::fs::Permissions::from_mode(0o755)).unwrap();
|
||||
|
||||
let (tx, _rx) = broadcast::channel::<AgentEvent>(64);
|
||||
let (watcher_tx, mut watcher_rx) = broadcast::channel::<WatcherEvent>(16);
|
||||
let event_log = Arc::new(Mutex::new(Vec::new()));
|
||||
|
||||
let result = run_agent_pty_streaming(
|
||||
"1211_story_clean",
|
||||
"coder-1",
|
||||
"sh",
|
||||
&[script.to_string_lossy().to_string()],
|
||||
"--",
|
||||
"/tmp",
|
||||
&tx,
|
||||
&event_log,
|
||||
None,
|
||||
0, // no inactivity timeout
|
||||
watcher_tx,
|
||||
None,
|
||||
None,
|
||||
)
|
||||
.await;
|
||||
|
||||
assert!(result.is_ok(), "PTY run should succeed: {:?}", result.err());
|
||||
assert!(result.unwrap().exit_ok, "clean exit should report exit_ok");
|
||||
|
||||
assert!(
|
||||
watcher_rx.try_recv().is_err(),
|
||||
"a clean turn end followed by normal exit must not send AgentCrashed"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -255,6 +255,20 @@ fn run_agent_pty_blocking(
|
||||
// tool_result clears the last in-flight id (story 1196).
|
||||
let mut tool_calls_in_flight: std::collections::HashSet<String> =
|
||||
std::collections::HashSet::new();
|
||||
// Tracks whether a `"result"` event was observed — the CLI's signal that
|
||||
// the current turn completed normally (see
|
||||
// llm/providers/claude_code/events/mod.rs). If the child dies before
|
||||
// this is set, the death is a mid-turn crash (AC1/AC2 of story 1211)
|
||||
// rather than a clean completion.
|
||||
let mut saw_result_event = false;
|
||||
// Most recent non-empty line emitted by the child before it died — used
|
||||
// as the "last error line" in the crash notification (AC4) since panics
|
||||
// and assertion failures print to the PTY's combined stdout/stderr.
|
||||
let mut last_line: Option<String> = None;
|
||||
// Set when the inactivity watchdog kills the child (AC3 of story 1211):
|
||||
// tracked instead of returning early so the death still flows through
|
||||
// the single crash-detection funnel below before the function returns.
|
||||
let mut timed_out = false;
|
||||
|
||||
loop {
|
||||
let effective_timeout = if !tool_calls_in_flight.is_empty() {
|
||||
@@ -291,10 +305,8 @@ fn run_agent_pty_blocking(
|
||||
{inactivity_timeout_secs}s with no output. Killing process."
|
||||
);
|
||||
let _ = child.kill();
|
||||
let _ = child.wait();
|
||||
return Err(format!(
|
||||
"Agent inactivity timeout: no output received for {inactivity_timeout_secs}s"
|
||||
));
|
||||
timed_out = true;
|
||||
break;
|
||||
}
|
||||
};
|
||||
|
||||
@@ -302,6 +314,7 @@ fn run_agent_pty_blocking(
|
||||
if trimmed.is_empty() {
|
||||
continue;
|
||||
}
|
||||
last_line = Some(trimmed.to_string());
|
||||
|
||||
// Try to parse as JSON
|
||||
let json: serde_json::Value = match serde_json::from_str(trimmed) {
|
||||
@@ -440,6 +453,10 @@ fn run_agent_pty_blocking(
|
||||
}
|
||||
}
|
||||
"result" => {
|
||||
// A "result" event signals the CLI turn completed — mark the
|
||||
// turn as clean so the post-loop crash check (AC1/AC2 of
|
||||
// story 1211) does not treat this exit as a mid-turn death.
|
||||
saw_result_event = true;
|
||||
// Extract token usage from the result event.
|
||||
if let Some(usage) = TokenUsage::from_result_event(&json) {
|
||||
slog!(
|
||||
@@ -481,6 +498,7 @@ fn run_agent_pty_blocking(
|
||||
false
|
||||
}
|
||||
};
|
||||
let exit_code = wait_result.as_ref().ok().map(|status| status.exit_code());
|
||||
|
||||
// Wait for the reader thread to finish so it releases the cloned PTY
|
||||
// master fd before we return. Without this, the next PTY spawn for the
|
||||
@@ -489,6 +507,32 @@ fn run_agent_pty_blocking(
|
||||
slog!("[agent:{story_id}:{agent_name}] Reader thread panicked: {e:?}");
|
||||
}
|
||||
|
||||
// AC1-AC3 (story 1211): single crash-detection funnel. Every exit path
|
||||
// above (EOF, reader disconnect, IO error, or watchdog kill on timeout)
|
||||
// breaks out of the loop into this one spot instead of returning early,
|
||||
// so a mid-turn death is detected and notified exactly once regardless
|
||||
// of which path triggered it (AC5). A turn is "clean" once a `"result"`
|
||||
// event has been observed; anything else was killed or crashed before
|
||||
// finishing its turn.
|
||||
if !saw_result_event {
|
||||
slog_warn!(
|
||||
"[agent:{story_id}:{agent_name}] Agent died mid-turn (no result event observed); \
|
||||
exit_code={exit_code:?}, last_line={last_line:?}"
|
||||
);
|
||||
let _ = watcher_tx.send(WatcherEvent::AgentCrashed {
|
||||
story_id: story_id.to_string(),
|
||||
agent_name: agent_name.to_string(),
|
||||
exit_code,
|
||||
last_error_line: last_line.clone(),
|
||||
});
|
||||
}
|
||||
|
||||
if timed_out {
|
||||
return Err(format!(
|
||||
"Agent inactivity timeout: no output received for {inactivity_timeout_secs}s"
|
||||
));
|
||||
}
|
||||
|
||||
// Log whether session was created — Session: None indicates CLI died
|
||||
// before emitting any events (possible causes: rate limit, budget
|
||||
// exhaustion, PTY write failure, CLI crash).
|
||||
|
||||
@@ -138,4 +138,18 @@ pub enum WatcherEvent {
|
||||
/// Identifier of the sled that observed the recovery.
|
||||
host_id: String,
|
||||
},
|
||||
/// An agent's PTY child process died before completing its current turn
|
||||
/// (no `"result"` event was observed) — a crash, kill, or watchdog
|
||||
/// termination rather than a clean finish (story 1211).
|
||||
/// Triggers a deterministic crash notification to configured chat rooms.
|
||||
AgentCrashed {
|
||||
/// Work item ID the agent was working on.
|
||||
story_id: String,
|
||||
/// Name of the agent whose process died.
|
||||
agent_name: String,
|
||||
/// Child process exit code, when available.
|
||||
exit_code: Option<u32>,
|
||||
/// Last non-empty line emitted by the child before it died, if any.
|
||||
last_error_line: Option<String>,
|
||||
},
|
||||
}
|
||||
|
||||
@@ -37,6 +37,8 @@ pub enum EventAction {
|
||||
},
|
||||
/// Post a disk-space recovery notification (story 1200).
|
||||
DiskRecovery,
|
||||
/// Post an agent-crashed notification (story 1211).
|
||||
AgentCrashed,
|
||||
/// Log server-side only; do not post to chat (e.g. hard rate-limit blocks).
|
||||
LogOnly,
|
||||
/// Reload the project configuration.
|
||||
@@ -68,6 +70,7 @@ pub fn classify(event: &WatcherEvent) -> EventAction {
|
||||
level: level.clone(),
|
||||
},
|
||||
WatcherEvent::DiskSpaceRecovered { .. } => EventAction::DiskRecovery,
|
||||
WatcherEvent::AgentCrashed { .. } => EventAction::AgentCrashed,
|
||||
_ => EventAction::Skip,
|
||||
}
|
||||
}
|
||||
@@ -216,4 +219,15 @@ mod tests {
|
||||
};
|
||||
assert_eq!(classify(&event), EventAction::MergeAutoRetry);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn agent_crashed_is_classified_correctly() {
|
||||
let event = WatcherEvent::AgentCrashed {
|
||||
story_id: "1_story_foo".to_string(),
|
||||
agent_name: "coder-1".to_string(),
|
||||
exit_code: Some(134),
|
||||
last_error_line: Some("thread panicked".to_string()),
|
||||
};
|
||||
assert_eq!(classify(&event), EventAction::AgentCrashed);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -221,6 +221,50 @@ pub fn format_agent_completed_notification(
|
||||
(plain, html)
|
||||
}
|
||||
|
||||
/// Format an agent-crashed notification message (story 1211).
|
||||
///
|
||||
/// Sent when an agent's PTY child process dies before completing its
|
||||
/// current turn (no `"result"` event observed) — a crash, kill, or
|
||||
/// watchdog termination rather than a clean finish. Includes the exit
|
||||
/// code and the last line emitted by the child when available.
|
||||
/// Returns `(plain_text, html)` suitable for `ChatTransport::send_message`.
|
||||
pub fn format_agent_crashed_notification(
|
||||
item_id: &str,
|
||||
story_name: &str,
|
||||
agent_name: &str,
|
||||
exit_code: Option<u32>,
|
||||
last_error_line: Option<&str>,
|
||||
) -> (String, String) {
|
||||
let number = extract_item_number(item_id).unwrap_or(item_id);
|
||||
let effective_name = if story_name.is_empty() {
|
||||
None
|
||||
} else {
|
||||
Some(story_name)
|
||||
};
|
||||
let name_plain = effective_name.map(|n| format!("{n} ")).unwrap_or_default();
|
||||
let name_html = effective_name
|
||||
.map(|n| format!("<em>{n}</em> "))
|
||||
.unwrap_or_default();
|
||||
|
||||
let exit_code_display = exit_code
|
||||
.map(|c| c.to_string())
|
||||
.unwrap_or_else(|| "unknown".to_string());
|
||||
|
||||
let mut plain = format!(
|
||||
"\u{1f480} #{number} {name_plain}\u{2014} {agent_name} crashed mid-turn \
|
||||
(exit code: {exit_code_display})"
|
||||
);
|
||||
let mut html = format!(
|
||||
"\u{1f480} <strong>#{number}</strong> {name_html}\u{2014} {agent_name} crashed mid-turn \
|
||||
(exit code: {exit_code_display})"
|
||||
);
|
||||
if let Some(line) = last_error_line {
|
||||
plain.push_str(&format!("\nLast line: {line}"));
|
||||
html.push_str(&format!("<br>Last line: <code>{line}</code>"));
|
||||
}
|
||||
(plain, html)
|
||||
}
|
||||
|
||||
/// Emoji used to badge a work item's kind in creation notifications.
|
||||
fn item_type_emoji(item_type: &str) -> &'static str {
|
||||
match item_type {
|
||||
|
||||
@@ -467,6 +467,44 @@ fn format_agent_completed_notification_falls_back_to_number() {
|
||||
);
|
||||
}
|
||||
|
||||
// ── format_agent_crashed_notification ─────────────────────────────────────
|
||||
|
||||
/// AC4: exit code and last error line both appear in the formatted text.
|
||||
#[test]
|
||||
fn format_agent_crashed_notification_includes_exit_code_and_last_line() {
|
||||
let (plain, html) = format_agent_crashed_notification(
|
||||
"1211_story_foo",
|
||||
"My Feature",
|
||||
"coder-1",
|
||||
Some(134),
|
||||
Some("assertion failed: output.write(&bytes).is_ok()"),
|
||||
);
|
||||
assert!(plain.contains("#1211"), "got: {plain}");
|
||||
assert!(plain.contains("coder-1"), "got: {plain}");
|
||||
assert!(plain.contains("crashed mid-turn"), "got: {plain}");
|
||||
assert!(plain.contains("exit code: 134"), "got: {plain}");
|
||||
assert!(
|
||||
plain.contains("assertion failed: output.write(&bytes).is_ok()"),
|
||||
"got: {plain}"
|
||||
);
|
||||
assert!(html.contains("exit code: 134"), "got: {html}");
|
||||
assert!(
|
||||
html.contains("<code>assertion failed: output.write(&bytes).is_ok()</code>"),
|
||||
"got: {html}"
|
||||
);
|
||||
}
|
||||
|
||||
/// AC4: exit code and last error line are both optional — an unavailable
|
||||
/// exit code renders as "unknown" and a missing last line is simply omitted.
|
||||
#[test]
|
||||
fn format_agent_crashed_notification_handles_missing_exit_code_and_line() {
|
||||
let (plain, html) =
|
||||
format_agent_crashed_notification("1211_story_foo", "", "coder-1", None, None);
|
||||
assert!(plain.contains("exit code: unknown"), "got: {plain}");
|
||||
assert!(!plain.contains("Last line"), "got: {plain}");
|
||||
assert!(!html.contains("Last line"), "got: {html}");
|
||||
}
|
||||
|
||||
// ── format_new_item_notification ──────────────────────────────────────────
|
||||
|
||||
#[test]
|
||||
|
||||
@@ -17,8 +17,9 @@ use super::super::filter::{
|
||||
};
|
||||
use super::super::format::{
|
||||
MERGE_FAILURE_TAIL_LINES, NewItemInfo, format_agent_completed_notification,
|
||||
format_agent_started_notification, format_blocked_notification,
|
||||
format_disk_recovery_notification, format_disk_warning_notification, format_error_notification,
|
||||
format_agent_crashed_notification, format_agent_started_notification,
|
||||
format_blocked_notification, format_disk_recovery_notification,
|
||||
format_disk_warning_notification, format_error_notification,
|
||||
format_merge_auto_retry_notification, format_new_items_notification,
|
||||
format_oauth_account_swapped, format_oauth_accounts_exhausted, format_rate_limit_notification,
|
||||
truncate_gate_output,
|
||||
@@ -423,6 +424,36 @@ pub fn spawn_notification_listener(
|
||||
}
|
||||
}
|
||||
}
|
||||
EventAction::AgentCrashed => {
|
||||
if !config.status_push_enabled {
|
||||
continue;
|
||||
}
|
||||
let WatcherEvent::AgentCrashed {
|
||||
ref story_id,
|
||||
ref agent_name,
|
||||
exit_code,
|
||||
ref last_error_line,
|
||||
} = event
|
||||
else {
|
||||
continue;
|
||||
};
|
||||
let story_name = find_story_name_any_stage(&project_root, story_id);
|
||||
let (plain, html) = format_agent_crashed_notification(
|
||||
story_id,
|
||||
&story_name,
|
||||
agent_name,
|
||||
exit_code,
|
||||
last_error_line.as_deref(),
|
||||
);
|
||||
slog!("[bot] Sending agent-crashed notification: {plain}");
|
||||
for room_id in &rooms_for_notification(&get_room_ids) {
|
||||
if let Err(e) = transport.send_message(room_id, &plain, &html).await {
|
||||
slog!(
|
||||
"[bot] Failed to send agent-crashed notification to {room_id}: {e}"
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
EventAction::DiskRecovery => {
|
||||
if !config.status_push_enabled {
|
||||
continue;
|
||||
|
||||
@@ -44,6 +44,8 @@ pub fn watcher_event_to_response(e: WatcherEvent) -> Option<WsResponse> {
|
||||
// Disk-space events are forwarded to chat transports only; no WebSocket message (story 1200).
|
||||
WatcherEvent::DiskSpaceWarning { .. } => None,
|
||||
WatcherEvent::DiskSpaceRecovered { .. } => None,
|
||||
// Crash notifications are forwarded to chat transports only; no WebSocket message (story 1211).
|
||||
WatcherEvent::AgentCrashed { .. } => None,
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user