diff --git a/.alfonso/release-notes/next/watch-gap-notice.md b/.alfonso/release-notes/next/watch-gap-notice.md new file mode 100644 index 000000000..75a1b3016 --- /dev/null +++ b/.alfonso/release-notes/next/watch-gap-notice.md @@ -0,0 +1 @@ +- When a stalled connection drops `bash_watch` match notices, the next notice for that watch says how many were lost and points at the task output, and health reports the gap until it is acknowledged; a gap notice alone never uses up a one-shot watch. diff --git a/benchmarks/aft-search/slice-descriptors/land-watch-gap-notice.json b/benchmarks/aft-search/slice-descriptors/land-watch-gap-notice.json new file mode 100644 index 000000000..7208dc684 --- /dev/null +++ b/benchmarks/aft-search/slice-descriptors/land-watch-gap-notice.json @@ -0,0 +1,12 @@ +{ + "slice_class": "non_ranking", + "targeted_mechanism": "none", + "kind": "harness", + "fixtures": [ + "harness-goldens" + ], + "slices": [ + "watch-gap-notice" + ], + "note": "bash_watch delivery-gap notice and health counters; no changed path matches RANKING_FENCE_PREFIXES." +} diff --git a/crates/aft/src/bash_background/registry.rs b/crates/aft/src/bash_background/registry.rs index 724248b94..dc60ca05f 100644 --- a/crates/aft/src/bash_background/registry.rs +++ b/crates/aft/src/bash_background/registry.rs @@ -2376,6 +2376,9 @@ impl BgTaskRegistry { .speech_db_path .write() .unwrap_or_else(|error| error.into_inner()) = path; + if let Ok(connection) = conn.lock() { + crate::db::bash_watches::refresh_watch_gap_health(&connection); + } if let Ok(mut slot) = self.inner.db_pool.write() { if slot.as_ref().is_none_or(|old| !Arc::ptr_eq(old, &conn)) { if let Err(error) = crate::db::call_ledger::recover_pool_once(&conn) { @@ -3855,7 +3858,19 @@ impl BgTaskRegistry { match crate::db::bash_watches::list_bash_pattern_watches_for_session( &conn, &harness, session_id, ) { - Ok(rows) => rows, + Ok(mut rows) => { + match crate::db::bash_watches::list_delivery_gaps( + &conn, + &harness, + Some(session_id), + ) { + Ok(gaps) => rows.extend(gaps), + Err(error) => crate::slog_warn!( + "failed to load retained watch gaps for session {session_id}: {error}" + ), + } + rows + } Err(error) => { crate::slog_warn!( "failed to load pending bash watches for session {session_id}: {error}" @@ -3902,6 +3917,13 @@ impl BgTaskRegistry { .match_batch .as_deref() .and_then(|batch| serde_json::from_str(batch).ok()); + if frame + .batch + .as_ref() + .is_some_and(|batch| !batch.has_matches()) + { + frame.once = false; + } Some(frame) }) .collect() @@ -4138,6 +4160,247 @@ impl BgTaskRegistry { } } + pub(crate) fn retain_retry_gap( + &self, + harness: &str, + candidate: &crate::subc::watch_delivery::Candidate, + ) -> Result<(), String> { + use super::watch_lines::{LineBatch, RetryGap, WatchLine}; + let frame = &candidate.frame; + let pool = self + .inner + .db_pool + .read() + .map_err(|e| e.to_string())? + .clone() + .ok_or_else(|| "watch persistence unavailable".to_string())?; + let conn = pool.lock().map_err(|e| e.to_string())?; + let acknowledged = crate::db::bash_watches::delivery_ack_version( + &conn, + harness, + &frame.session_id, + &frame.task_id, + &frame.watch_id, + ) + .map_err(|e| e.to_string())?; + if acknowledged + .as_ref() + .is_some_and(|(version, epoch, through)| { + (epoch == &candidate.eviction_epoch && *through >= candidate.last_eviction) + || (frame.batch.is_some() && *version >= candidate.last_version) + || (frame.batch.is_none() + && candidate.same_occurrence + && *version == candidate.last_version) + }) + { + return Ok(()); + } + let found = crate::db::bash_watches::get_bash_pattern_watch( + &conn, + harness, + &frame.session_id, + &frame.task_id, + &frame.watch_id, + ) + .map_err(|e| e.to_string())?; + let registered = found.is_some(); + let found = match found { + Some(row) => Some(row), + None => { + crate::db::bash_watches::list_delivery_gaps(&conn, harness, Some(&frame.session_id)) + .map_err(|e| e.to_string())? + .into_iter() + .find(|row| row.task_id == frame.task_id && row.watch_id == frame.watch_id) + } + }; + let mut row = + found + .clone() + .unwrap_or_else(|| crate::db::bash_watches::BashPatternWatchRow { + harness: harness.to_string(), + session_id: frame.session_id.clone(), + task_id: frame.task_id.clone(), + watch_id: frame.watch_id.clone(), + pattern_kind: "substring".to_string(), + pattern: String::new(), + once: true, + each_line: false, + match_batch: None, + created_at: 0, + stdout_offset: 0, + stderr_offset: 0, + pty_offset: 0, + scanning: false, + pending_match: false, + match_text: None, + match_offset: None, + match_context: None, + }); + let previous = row + .match_batch + .as_deref() + .and_then(|b| serde_json::from_str::(b).ok()); + if frame.batch.is_some() + && previous + .as_ref() + .and_then(|b| b.acknowledged_version) + .is_some_and(|v| v >= candidate.last_version) + { + return Ok(()); + } + let recoverable_batch = candidate.same_generation + && row.pending_match + && frame.batch.as_ref().is_some_and(|evicted| { + previous.as_ref().is_some_and(|saved| { + evicted.pending_id.is_some() + && saved.pending_id == evicted.pending_id + && saved.version.is_some_and(|v| v >= candidate.last_version) + }) + }); + let recoverable_scalar = frame.batch.is_none() + && candidate.same_occurrence + && row.pending_match + && candidate.first_version == candidate.last_version + && ((row.match_offset == Some(candidate.first_version as i64) + && row.match_text.as_ref().is_some_and(|text| { + use sha2::Digest; + sha2::Sha256::digest(text.as_bytes()).as_slice() + == candidate.fingerprint.as_slice() + })) + || previous.as_ref().is_some_and(|b| { + b.lines.iter().any(|line| { + use sha2::Digest; + line.offset == candidate.first_version + && sha2::Sha256::digest(line.text.as_bytes()).as_slice() + == candidate.fingerprint.as_slice() + }) + })); + if recoverable_batch || recoverable_scalar { + return Ok(()); + } + let mut batch = if row.pending_match { + previous.clone().unwrap_or_default() + } else { + LineBatch::default() + }; + if batch.retry_gap.as_ref().is_some_and(|g| { + g.eviction_epoch == candidate.eviction_epoch + && g.evicted_through >= candidate.last_eviction + }) { + return Ok(()); + } + // A superseded scalar has one identifiable occurrence. Overlapping + // snapshots and uncertain persistence never imply a lost-line count. + let known = u64::from( + registered + && frame.batch.is_none() + && candidate.same_occurrence + && acknowledged + .as_ref() + .is_none_or(|(version, _, _)| *version < candidate.first_version) + && batch + .retry_gap + .as_ref() + .is_none_or(|gap| gap.last_match_version < candidate.first_version) + && row.pending_match + && row + .match_offset + .is_some_and(|v| v > candidate.last_version as i64) + && candidate.first_version == candidate.last_version, + ); + if row.pending_match && row.match_batch.is_none() { + if let Some(text) = &row.match_text { + let offset = row.match_offset.unwrap_or_default().max(0) as u64; + batch.lines.push(WatchLine { + stream: "output".to_string(), + offset, + end_offset: offset.saturating_add(1), + text: text.clone(), + count: 1, + unavailable: false, + }); + } + } + let mut gap = batch.retry_gap.take().unwrap_or_else(|| RetryGap { + reason: "retry_overflow".to_string(), + ..RetryGap::default() + }); + gap.dropped_notifications = gap + .dropped_notifications + .saturating_add(candidate.notifications); + gap.known_missing_occurrences = gap.known_missing_occurrences.saturating_add(known); + if gap.dropped_notifications == candidate.notifications { + gap.first_offset = candidate.first_offset; + gap.last_offset = candidate.last_offset; + gap.stream = candidate.stream.clone(); + } else if gap.stream == candidate.stream { + gap.first_offset = gap + .first_offset + .zip(candidate.first_offset) + .map(|(a, b)| a.min(b)); + gap.last_offset = gap + .last_offset + .zip(candidate.last_offset) + .map(|(a, b)| a.max(b)); + } else { + gap.first_offset = None; + gap.last_offset = None; + gap.stream = None; + } + gap.evicted_through = candidate.last_eviction; + gap.eviction_epoch = candidate.eviction_epoch.clone(); + gap.last_match_version = gap.last_match_version.max(candidate.last_version); + let version = previous + .as_ref() + .and_then(|b| b.version) + .unwrap_or_default() + .max(row.match_offset.unwrap_or_default().max(0) as u64) + .max(candidate.last_version) + .saturating_add(1); + batch.version = Some(version); + batch + .pending_id + .get_or_insert_with(|| format!("{}:{version}", frame.watch_id)); + batch.snapshot = true; + batch.retry_gap = Some(gap); + row.pending_match = true; + row.match_offset = Some(version as i64); + row.match_text + .get_or_insert_with(|| "watch delivery gap".to_string()); + row.match_context = Some(crate::subc_format::format_watch_line_batch( + &frame.task_id, + &batch, + )); + row.match_batch = Some(serde_json::to_string(&batch).map_err(|e| e.to_string())?); + if registered { + crate::db::bash_watches::upsert_bash_pattern_watch(&conn, &row) + .map_err(|e| e.to_string())?; + } else { + crate::db::bash_watches::upsert_delivery_gap(&conn, &row).map_err(|e| e.to_string())?; + } + drop(conn); + if let Some(sender) = self + .inner + .progress_sender + .lock() + .map_err(|e| e.to_string())? + .clone() + { + let mut notice = BashPatternMatchFrame::new( + &frame.task_id, + &frame.session_id, + &frame.watch_id, + row.match_text.unwrap_or_default(), + version, + row.match_context.unwrap_or_default(), + row.once && !row.scanning && batch.has_matches(), + ); + notice.batch = Some(batch); + sender(PushFrame::BashPatternMatch(notice)); + } + Ok(()) + } + fn persist_watch_match( &self, session_id: &str, @@ -4166,6 +4429,26 @@ impl BgTaskRegistry { ) else { return; }; + if pattern_match.batch.is_none() { + if let Some(previous) = row + .match_batch + .as_deref() + .and_then(|b| serde_json::from_str::(b).ok()) + .filter(|b| b.version.is_some()) + { + let mut batch = super::watch_lines::LineBatch::default(); + batch.acknowledged_version = previous.acknowledged_version; + batch.lines = vec![super::watch_lines::WatchLine { + stream: "output".to_string(), + offset: pattern_match.match_offset, + end_offset: pattern_match.match_offset.saturating_add(1), + text: pattern_match.match_text.clone(), + count: 1, + unavailable: false, + }]; + pattern_match.batch = Some(batch); + } + } if let Some(batch) = pattern_match.batch.as_mut() { let previous = row.match_batch.as_deref().and_then(|value| { serde_json::from_str::(value).ok() @@ -4329,11 +4612,20 @@ impl BgTaskRegistry { .ok_or_else(|| "watch persistence unavailable".to_string())?; let (removed, settled) = { let conn = pool.lock().map_err(|error| error.to_string())?; - let Some(mut row) = crate::db::bash_watches::get_bash_pattern_watch( + let registered = crate::db::bash_watches::get_bash_pattern_watch( &conn, &harness, session_id, task_id, watch_id, ) - .map_err(|error| error.to_string())? - else { + .map_err(|error| error.to_string())?; + let retained = registered.is_none(); + let Some(mut row) = (match registered { + Some(row) => Some(row), + None => { + crate::db::bash_watches::list_delivery_gaps(&conn, &harness, Some(session_id)) + .map_err(|e| e.to_string())? + .into_iter() + .find(|row| row.task_id == task_id && row.watch_id == watch_id) + } + }) else { return Ok(None); }; if !row.pending_match @@ -4342,13 +4634,44 @@ impl BgTaskRegistry { return Ok(None); } let removed = (row.once && !row.scanning) || terminal; + crate::db::bash_watches::acknowledge_delivery( + &conn, + &harness, + session_id, + task_id, + watch_id, + match_offset, + row.match_batch + .as_deref() + .and_then(|value| { + serde_json::from_str::(value).ok() + }) + .and_then(|batch| batch.retry_gap) + .as_ref(), + ) + .map_err(|error| error.to_string())?; if removed { - crate::db::bash_watches::delete_bash_pattern_watch( - &conn, &harness, session_id, task_id, watch_id, - ) - .map_err(|error| error.to_string())?; + if retained { + crate::db::bash_watches::delete_delivery_gap( + &conn, &harness, session_id, task_id, watch_id, + ) + .map_err(|e| e.to_string())?; + } else { + crate::db::bash_watches::delete_bash_pattern_watch( + &conn, &harness, session_id, task_id, watch_id, + ) + .map_err(|error| error.to_string())?; + } } else { row.pending_match = false; + if let Some(mut batch) = row + .match_batch + .as_deref() + .and_then(|b| serde_json::from_str::(b).ok()) + { + batch.acknowledged_version = Some(match_offset); + row.match_batch = serde_json::to_string(&batch).ok(); + } row.match_text = None; row.match_offset = None; row.match_context = None; @@ -4361,7 +4684,11 @@ impl BgTaskRegistry { ) .map_err(|error| error.to_string())? .iter() - .any(|row| row.pending_match); + .any(|row| row.pending_match) + && !crate::db::bash_watches::list_delivery_gaps(&conn, &harness, Some(session_id)) + .map_err(|e| e.to_string())? + .iter() + .any(|row| row.task_id == task_id); (removed, settled) }; self.sync_memory_watches_from_persistence(task_id); @@ -4392,19 +4719,24 @@ impl BgTaskRegistry { let Ok(conn) = pool.lock() else { return; }; - if task_terminal { - let _ = crate::db::bash_watches::delete_bash_pattern_watches_for_task( - &conn, &harness, session_id, task_id, - ); - return; - } let Ok(rows) = crate::db::bash_watches::list_bash_pattern_watches_for_task( &conn, &harness, session_id, task_id, ) else { return; }; for mut row in rows { - if row.once && !row.scanning { + // A task acknowledgement carries no watch version and cannot + // confirm delivery of a newer retry-overflow warning. + if row.pending_match + && row + .match_batch + .as_deref() + .and_then(|b| serde_json::from_str::(b).ok()) + .is_some_and(|b| b.retry_gap.is_some()) + { + continue; + } + if task_terminal || (row.once && !row.scanning) { let _ = crate::db::bash_watches::delete_bash_pattern_watch( &conn, &harness, @@ -10471,6 +10803,14 @@ impl BgTaskRegistry { ) })); } + if let Ok(rows) = crate::db::bash_watches::list_delivery_gaps(&conn, harness, None) { + keys.extend(rows.into_iter().map(|row| { + format!( + "match\0{}\0{}\0{}", + row.session_id, row.task_id, row.watch_id + ) + })); + } } keys } @@ -12523,17 +12863,19 @@ impl BgTaskRegistry { row.match_offset, row.match_context.clone(), ) { + let batch = row.match_batch.as_deref().and_then(|batch| { + serde_json::from_str::(batch).ok() + }); pending_to_emit.push(PatternMatch { watch_id: row.watch_id.clone(), task_id: row.task_id.clone(), match_text, match_offset: match_offset.max(0) as u64, context, - once: row.once && !row.scanning, - batch: row - .match_batch - .as_deref() - .and_then(|batch| serde_json::from_str(batch).ok()), + once: row.once + && !row.scanning + && batch.as_ref().is_none_or(|b| b.has_matches()), + batch, // The saved match text and context were rendered // before this row was reloaded. literal_pattern: false, @@ -27472,6 +27814,275 @@ mod tests { assert!(masked.ends_with("page final\n")); } + #[test] + fn retry_overflow_sticky_gap_survives_restart_and_exact_ack() { + let dir = tempfile::tempdir().unwrap(); + let (registry, task, _, _) = each_line_running_fixture(dir.path()); + let _reap = KillTasksOnDrop(registry.clone()); + let watch = registry + .register_watch( + task.task_id.clone(), + WatchPattern::Substring("READY".into()), + false, + ) + .unwrap(); + let mut first = PatternMatch { + task_id: task.task_id.clone(), + watch_id: watch.clone(), + match_text: "READY first".into(), + match_offset: 10, + context: "READY first".into(), + once: false, + batch: None, + literal_pattern: false, + }; + registry.persist_watch_match("session", &task.task_id, &mut first, 20, 0, 0); + let mut second = first.clone(); + second.match_offset = 30; + second.match_text = "READY second".into(); + registry.persist_watch_match("session", &task.task_id, &mut second, 40, 0, 0); + let frame = BashPatternMatchFrame::new( + &task.task_id, + "session", + &watch, + "READY first", + 10, + "READY first", + false, + ); + let candidate = crate::subc::watch_delivery::stalled_route_eviction(frame) + .expect("stalled FIFO overflow must retain a watch-gap candidate"); + registry.retain_retry_gap("opencode", &candidate).unwrap(); + let pending = registry + .pending_pattern_matches_for_session("session") + .into_iter() + .find(|f| f.watch_id == watch) + .unwrap(); + let gap = pending.batch.as_ref().unwrap().retry_gap.as_ref().unwrap(); + assert_eq!(gap.reason, "retry_overflow"); + assert_eq!(gap.dropped_notifications, 1); + assert_eq!(gap.known_missing_occurrences, 1); + assert_eq!(gap.first_offset, Some(10)); + assert!(pending.context.contains("they cannot be recovered.")); + for terminal in [false, true] { + registry.ack_persisted_watches_for_task("session", &task.task_id, terminal); + assert!(registry + .pending_pattern_matches_for_session("session") + .iter() + .any(|frame| frame.watch_id == watch)); + } + assert!(registry + .ack_pattern_match_for_session("session", &task.task_id, &watch, 30) + .unwrap() + .is_none()); + let (restarted, _, _) = registry_with_db_and_frames(dir.path()); + let restored = restarted + .pending_pattern_matches_for_session("session") + .into_iter() + .find(|f| f.watch_id == watch) + .unwrap(); + assert_eq!(restored.match_offset, pending.match_offset); + assert_eq!( + restored + .batch + .unwrap() + .retry_gap + .unwrap() + .dropped_notifications, + 1 + ); + restarted + .ack_pattern_match_for_session("session", &task.task_id, &watch, pending.match_offset) + .unwrap() + .unwrap(); + assert!(!restarted + .pending_pattern_matches_for_session("session") + .iter() + .any(|f| f.watch_id == watch)); + // An evicted duplicate of the acknowledged frame must not reopen the gap. + restarted.retain_retry_gap("opencode", &candidate).unwrap(); + assert!(!restarted + .pending_pattern_matches_for_session("session") + .iter() + .any(|f| f.watch_id == watch)); + let later = BashPatternMatchFrame::new( + &task.task_id, + "session", + &watch, + "another lost notice", + 20, + "another lost notice", + false, + ); + let later = crate::subc::watch_delivery::stalled_route_eviction(later).unwrap(); + restarted.retain_retry_gap("opencode", &later).unwrap(); + let newer = restarted + .pending_pattern_matches_for_session("session") + .into_iter() + .find(|f| f.watch_id == watch) + .unwrap(); + assert!(newer.match_offset > pending.match_offset); + assert!(restarted + .ack_pattern_match_for_session("session", &task.task_id, &watch, pending.match_offset) + .unwrap() + .is_none()); + restarted + .ack_pattern_match_for_session("session", &task.task_id, &watch, newer.match_offset) + .unwrap() + .unwrap(); + } + + #[test] + fn retry_overflow_gap_only_does_not_consume_once_watch_and_recoverable_batches_are_not_lost() { + let dir = tempfile::tempdir().unwrap(); + let (registry, task, _, watch) = each_line_running_fixture_with_once(dir.path(), true); + let _reap = KillTasksOnDrop(registry.clone()); + let frame = BashPatternMatchFrame::new( + &task.task_id, + "session", + &watch, + "unpersisted", + 10, + "unpersisted", + true, + ); + let candidate = crate::subc::watch_delivery::stalled_route_eviction(frame).unwrap(); + registry.retain_retry_gap("opencode", &candidate).unwrap(); + let pending = registry + .pending_pattern_matches_for_session("session") + .pop() + .unwrap(); + assert!(!pending.once); + assert!(!pending.batch.as_ref().unwrap().has_matches()); + assert_eq!( + pending + .batch + .as_ref() + .unwrap() + .retry_gap + .as_ref() + .unwrap() + .known_missing_occurrences, + 0 + ); + registry + .ack_pattern_match_for_session("session", &task.task_id, &watch, pending.match_offset) + .unwrap() + .unwrap(); + assert_eq!(registry.active_watch_count(&task.task_id), 1); + append_each_line_output(&task.paths.stdout, b"READY after gap\n"); + registry.scan_task_watch_output(&task); + let matched = registry + .pending_pattern_matches_for_session("session") + .pop() + .unwrap(); + assert!(matched.once); + assert!(matched.batch.as_ref().unwrap().retry_gap.is_none()); + let candidate = + crate::subc::watch_delivery::stalled_route_eviction(matched.clone()).unwrap(); + registry.retain_retry_gap("opencode", &candidate).unwrap(); + let recovered = registry + .pending_pattern_matches_for_session("session") + .pop() + .unwrap(); + assert_eq!(recovered.match_offset, matched.match_offset); + assert!(recovered.batch.unwrap().retry_gap.is_none()); + } + + #[test] + fn retry_overflow_retired_watch_gap_ack_prevents_replay_after_restart() { + let dir = tempfile::tempdir().unwrap(); + let (registry, task, _, _) = each_line_running_fixture(dir.path()); + let _reap = KillTasksOnDrop(registry.clone()); + let frame = BashPatternMatchFrame::new( + &task.task_id, + "session", + "retired-watch", + "old match", + 10, + "old match", + false, + ); + let candidate = crate::subc::watch_delivery::stalled_route_eviction(frame).unwrap(); + registry.retain_retry_gap("opencode", &candidate).unwrap(); + let notice = registry + .pending_pattern_matches_for_session("session") + .into_iter() + .find(|f| f.watch_id == "retired-watch") + .unwrap(); + assert!(!notice.once); + assert!(notice.batch.as_ref().unwrap().lines.is_empty()); + registry + .ack_pattern_match_for_session( + "session", + &task.task_id, + "retired-watch", + notice.match_offset, + ) + .unwrap() + .unwrap(); + let (restarted, _, _) = registry_with_db_and_frames(dir.path()); + restarted.retain_retry_gap("opencode", &candidate).unwrap(); + assert!(!restarted + .pending_pattern_matches_for_session("session") + .iter() + .any(|f| f.watch_id == "retired-watch")); + } + + #[test] + fn retry_overflow_missing_task_keeps_a_session_gap_until_exact_ack() { + let dir = tempfile::tempdir().unwrap(); + let (registry, _, _) = registry_with_db_and_frames(dir.path()); + let frame = BashPatternMatchFrame::new( + "missing-task", + "session", + "missing-watch", + "match", + 10, + "match", + false, + ); + let candidate = crate::subc::watch_delivery::stalled_route_eviction(frame).unwrap(); + registry.retain_retry_gap("opencode", &candidate).unwrap(); + let (restarted, _, _) = registry_with_db_and_frames(dir.path()); + let notice = restarted + .pending_pattern_matches_for_session("session") + .pop() + .unwrap(); + assert_eq!(notice.task_id, "missing-task"); + assert!(!notice.once); + assert!(restarted.has_unacked_wakes_for_session("session")); + assert!(restarted + .unacked_wake_keys() + .contains("match\0session\0missing-task\0missing-watch")); + assert!(restarted + .ack_pattern_match_for_session( + "session", + "missing-task", + "missing-watch", + notice.match_offset - 1 + ) + .unwrap() + .is_none()); + restarted + .ack_pattern_match_for_session( + "session", + "missing-task", + "missing-watch", + notice.match_offset, + ) + .unwrap() + .unwrap(); + assert!(restarted + .pending_pattern_matches_for_session("session") + .is_empty()); + assert!(!restarted.has_unacked_wakes_for_session("session")); + restarted.retain_retry_gap("opencode", &candidate).unwrap(); + assert!(restarted + .pending_pattern_matches_for_session("session") + .is_empty()); + } + fn each_line_running_fixture( storage: &Path, ) -> ( diff --git a/crates/aft/src/bash_background/watch_lines.rs b/crates/aft/src/bash_background/watch_lines.rs index 472d1583e..fc6d01671 100644 --- a/crates/aft/src/bash_background/watch_lines.rs +++ b/crates/aft/src/bash_background/watch_lines.rs @@ -60,8 +60,31 @@ pub struct WatchLine { pub unavailable: bool, } +#[derive(Debug, Clone, Default, Serialize, Deserialize)] +pub struct RetryGap { + pub reason: String, + pub dropped_notifications: u64, + pub known_missing_occurrences: u64, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub first_offset: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub last_offset: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub stream: Option, + #[serde(default)] + pub evicted_through: u64, + #[serde(default)] + pub eviction_epoch: String, + #[serde(default)] + pub last_match_version: u64, +} + #[derive(Debug, Clone, Default, Serialize, Deserialize)] pub struct LineBatch { + #[serde(default, skip_serializing_if = "Option::is_none")] + pub acknowledged_version: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub retry_gap: Option, #[serde(default, skip_serializing_if = "Option::is_none")] pub masking_limit_bytes: Option, #[serde(default, skip_serializing_if = "Option::is_none")] @@ -85,6 +108,9 @@ impl LineBatch { } pub fn merge(&mut self, next: LineBatch) { + if next.retry_gap.is_some() { + self.retry_gap = next.retry_gap.clone(); + } self.masking_limit_bytes = next.masking_limit_bytes.or(self.masking_limit_bytes); self.terminal_summary = next.terminal_summary; let mut retained = self.lines.iter().map(|line| line.count).sum::(); @@ -123,7 +149,13 @@ impl LineBatch { } fn is_empty(&self) -> bool { - self.lines.is_empty() && self.dropped_lines == 0 && self.skipped_long_lines == 0 + self.lines.is_empty() + && self.dropped_lines == 0 + && self.skipped_long_lines == 0 + && self + .retry_gap + .as_ref() + .is_none_or(|gap| gap.dropped_notifications == 0) } } diff --git a/crates/aft/src/db/bash_watches.rs b/crates/aft/src/db/bash_watches.rs index fd0db51bb..6f8b57689 100644 --- a/crates/aft/src/db/bash_watches.rs +++ b/crates/aft/src/db/bash_watches.rs @@ -77,6 +77,7 @@ pub fn upsert_bash_pattern_watch( row.match_batch, ], )?; + refresh_watch_gap_health(conn); Ok(()) } @@ -87,11 +88,13 @@ pub fn delete_bash_pattern_watch( task_id: &str, watch_id: &str, ) -> rusqlite::Result { - conn.execute( + let deleted = conn.execute( "DELETE FROM bash_pattern_watches WHERE harness = ?1 AND session_id = ?2 AND task_id = ?3 AND watch_id = ?4", params![harness, session_id, task_id, watch_id], - ) + )?; + refresh_watch_gap_health(conn); + Ok(deleted) } pub fn delete_bash_pattern_watches_for_task( @@ -100,11 +103,13 @@ pub fn delete_bash_pattern_watches_for_task( session_id: &str, task_id: &str, ) -> rusqlite::Result { - conn.execute( + let deleted = conn.execute( "DELETE FROM bash_pattern_watches WHERE harness = ?1 AND session_id = ?2 AND task_id = ?3", params![harness, session_id, task_id], - ) + )?; + refresh_watch_gap_health(conn); + Ok(deleted) } pub fn list_bash_pattern_watches_for_task( @@ -187,9 +192,9 @@ pub fn count_pending_bash_pattern_watches_for_session( session_id: &str, ) -> rusqlite::Result { conn.query_row( - "SELECT COUNT(*) + "SELECT (SELECT COUNT(*) FROM bash_pattern_watches - WHERE harness = ?1 AND session_id = ?2 AND pending_match = 1", + WHERE harness = ?1 AND session_id = ?2 AND pending_match = 1) + (SELECT COUNT(*) FROM bash_watch_delivery_gaps WHERE harness=?1 AND session_id=?2)", params![harness, session_id], |row| row.get(0), ) @@ -200,9 +205,9 @@ pub fn count_pending_bash_pattern_watches( harness: &str, ) -> rusqlite::Result { conn.query_row( - "SELECT COUNT(*) + "SELECT (SELECT COUNT(*) FROM bash_pattern_watches - WHERE harness = ?1 AND pending_match = 1", + WHERE harness = ?1 AND pending_match = 1) + (SELECT COUNT(*) FROM bash_watch_delivery_gaps WHERE harness=?1)", params![harness], |row| row.get(0), ) @@ -273,3 +278,216 @@ fn map_watch_row(row: &Row<'_>) -> rusqlite::Result { match_batch: row.get(17)?, }) } + +// Keep the last census for every opened database, even after its actor retires. +// Health reads only this compact cache, never SQLite on the frame loop. +type GapCensus = (u64, u64, bool); +static WATCH_GAP_CENSUS: std::sync::OnceLock< + std::sync::Mutex>, +> = std::sync::OnceLock::new(); + +/// The census entry for one database: SQLite's own file name for it, so every +/// connection to the same file shares an entry, or a per-connection key for an +/// in-memory database. +fn census_key(conn: &Connection) -> String { + conn.path() + .filter(|path| !path.is_empty()) + .map(str::to_owned) + .unwrap_or_else(|| format!("memory:{:p}", conn)) +} + +pub(crate) fn refresh_watch_gap_health(conn: &Connection) { + let key = census_key(conn); + let census = (|| -> rusqlite::Result { + let mut stmt = conn.prepare("SELECT match_batch FROM bash_pattern_watches WHERE pending_match = 1 AND match_batch IS NOT NULL UNION ALL SELECT batch FROM bash_watch_delivery_gaps")?; + let batches = stmt.query_map([], |row| row.get::<_, String>(0))?; + let (mut outstanding, mut missing) = (0u64, 0u64); + let mut uncertain = false; + for batch in batches { + match serde_json::from_str::(&batch?) { + Ok(batch) => { + if let Some(gap) = batch.retry_gap { + outstanding = outstanding.saturating_add(1); + missing = missing.saturating_add(gap.known_missing_occurrences); + } + } + Err(_) => uncertain = true, + } + } + Ok((outstanding, missing, uncertain)) + })(); + let mut cache = WATCH_GAP_CENSUS + .get_or_init(Default::default) + .lock() + .unwrap(); + match census { + Ok((0, 0, false)) => { + cache.remove(&key); + } + Ok(value) => { + cache.insert(key, value); + } + Err(_) => { + cache.entry(key).or_default().2 = true; + } + } +} + +pub(crate) fn watch_gap_health() -> GapCensus { + WATCH_GAP_CENSUS + .get_or_init(Default::default) + .lock() + .unwrap() + .values() + .fold((0u64, 0u64, false), |(gaps, missing, uncertain), next| { + ( + gaps.saturating_add(next.0), + missing.saturating_add(next.1), + uncertain || next.2, + ) + }) +} + +pub(crate) fn delivery_ack_version( + conn: &Connection, + harness: &str, + session: &str, + task: &str, + watch: &str, +) -> rusqlite::Result> { + conn.query_row("SELECT version, eviction_epoch, evicted_through FROM bash_watch_delivery_acks WHERE harness=?1 AND session_id=?2 AND task_id=?3 AND watch_id=?4", params![harness, session, task, watch], |r| Ok((r.get::<_, i64>(0)?.max(0) as u64, r.get(1)?, r.get::<_, i64>(2)?.max(0) as u64))).optional() +} + +pub(crate) fn acknowledge_delivery( + conn: &Connection, + harness: &str, + session: &str, + task: &str, + watch: &str, + version: u64, + gap: Option<&crate::bash_background::watch_lines::RetryGap>, +) -> rusqlite::Result<()> { + let epoch = gap + .map(|gap| gap.eviction_epoch.as_str()) + .unwrap_or(crate::subc::watch_delivery::eviction_epoch()); + let through = gap.map(|gap| gap.evicted_through).unwrap_or_default(); + conn.execute("INSERT INTO bash_watch_delivery_acks(harness, session_id, task_id, watch_id, version, acknowledged_at, eviction_epoch, evicted_through) VALUES(?1,?2,?3,?4,?5,unixepoch(),?6,?7) ON CONFLICT(harness, session_id, task_id, watch_id) DO UPDATE SET version=MAX(version,excluded.version), acknowledged_at=excluded.acknowledged_at, eviction_epoch=CASE WHEN excluded.eviction_epoch != '' THEN excluded.eviction_epoch ELSE eviction_epoch END, evicted_through=CASE WHEN eviction_epoch = excluded.eviction_epoch THEN MAX(evicted_through,excluded.evicted_through) WHEN excluded.eviction_epoch != '' THEN excluded.evicted_through ELSE evicted_through END", params![harness, session, task, watch, version as i64, epoch, through as i64])?; + // Only prune old acknowledgements from an earlier daemon: its process-local + // retry copies no longer exist, even if the current daemon is long-lived. + conn.execute( + "DELETE FROM bash_watch_delivery_acks WHERE acknowledged_at < unixepoch() - 2592000 AND eviction_epoch != ?1", + [crate::subc::watch_delivery::eviction_epoch()], + )?; + Ok(()) +} + +/// One database's census entry. Uses the connection's own key rather than a +/// path the test spells, since SQLite reports its file name in its own form +/// (symlinks resolved on macOS, no verbatim prefix on Windows). +#[cfg(test)] +pub(crate) fn watch_gap_health_for(conn: &Connection) -> GapCensus { + WATCH_GAP_CENSUS + .get_or_init(Default::default) + .lock() + .unwrap() + .get(&census_key(conn)) + .copied() + .unwrap_or_default() +} + +pub(crate) fn list_delivery_gaps( + conn: &Connection, + harness: &str, + session: Option<&str>, +) -> rusqlite::Result> { + let mut statement = conn.prepare("SELECT session_id, task_id, watch_id, batch FROM bash_watch_delivery_gaps WHERE harness=?1 AND (?2 IS NULL OR session_id=?2)")?; + let rows = statement.query_map(params![harness, session], |row| { + let session_id = row.get(0)?; + let task_id: String = row.get(1)?; + let watch_id = row.get(2)?; + let saved: String = row.get(3)?; + let batch = serde_json::from_str::(&saved) + .map_err(|error| { + rusqlite::Error::FromSqlConversionFailure( + 3, + rusqlite::types::Type::Text, + Box::new(error), + ) + })?; + Ok(BashPatternWatchRow { + harness: harness.into(), + session_id, + task_id: task_id.clone(), + watch_id, + pattern_kind: "substring".into(), + pattern: String::new(), + once: true, + each_line: false, + match_batch: Some(saved), + created_at: 0, + stdout_offset: 0, + stderr_offset: 0, + pty_offset: 0, + scanning: false, + pending_match: true, + match_text: Some("watch delivery gap".into()), + match_offset: batch.version.map(|version| version as i64), + match_context: Some(crate::subc_format::format_watch_line_batch( + &task_id, &batch, + )), + }) + })?; + rows.collect() +} + +pub(crate) fn upsert_delivery_gap( + conn: &Connection, + row: &BashPatternWatchRow, +) -> rusqlite::Result<()> { + conn.execute("INSERT INTO bash_watch_delivery_gaps(harness,session_id,task_id,watch_id,batch) VALUES(?1,?2,?3,?4,?5) ON CONFLICT(harness,session_id,task_id,watch_id) DO UPDATE SET batch=excluded.batch", params![row.harness,row.session_id,row.task_id,row.watch_id,row.match_batch])?; + refresh_watch_gap_health(conn); + Ok(()) +} + +pub(crate) fn delete_delivery_gap( + conn: &Connection, + harness: &str, + session: &str, + task: &str, + watch: &str, +) -> rusqlite::Result<()> { + conn.execute("DELETE FROM bash_watch_delivery_gaps WHERE harness=?1 AND session_id=?2 AND task_id=?3 AND watch_id=?4", params![harness,session,task,watch])?; + refresh_watch_gap_health(conn); + Ok(()) +} + +#[cfg(test)] +mod delivery_ack_tests { + use super::*; + + #[test] + fn retry_overflow_ack_tombstones_survive_a_long_lived_daemon() { + let dir = tempfile::tempdir().unwrap(); + let conn = crate::db::open(&dir.path().join("aft.db")).unwrap(); + acknowledge_delivery(&conn, "opencode", "session", "task", "live", 5, None).unwrap(); + conn.execute( + "UPDATE bash_watch_delivery_acks SET acknowledged_at=unixepoch()-2592001", + [], + ) + .unwrap(); + conn.execute("INSERT INTO bash_watch_delivery_acks VALUES('opencode','session','task','old',5,unixepoch()-2592001,'earlier-daemon',0)", []).unwrap(); + acknowledge_delivery(&conn, "opencode", "session", "task", "another", 6, None).unwrap(); + assert_eq!( + delivery_ack_version(&conn, "opencode", "session", "task", "live") + .unwrap() + .unwrap() + .0, + 5 + ); + assert!( + delivery_ack_version(&conn, "opencode", "session", "task", "old") + .unwrap() + .is_none() + ); + } +} diff --git a/crates/aft/src/db/mod.rs b/crates/aft/src/db/mod.rs index 4ff8e40dd..7b0a2e0ec 100644 --- a/crates/aft/src/db/mod.rs +++ b/crates/aft/src/db/mod.rs @@ -35,7 +35,7 @@ pub mod state; #[cfg(test)] pub(crate) mod wal_credit_probe; -pub const CURRENT_SCHEMA_VERSION: u32 = 23; +pub const CURRENT_SCHEMA_VERSION: u32 = 24; const MIGRATION_VERSIONS: &[u32] = &[ 1, @@ -61,6 +61,7 @@ const MIGRATION_VERSIONS: &[u32] = &[ 21, 22, 23, + 24, ]; // Performance-only indexes do not change the table format or reader floor. @@ -995,6 +996,7 @@ fn migration_already_applied(conn: &Connection, version: u32) -> rusqlite::Resul ) .map(|object_count| object_count == 1), 14 => conn.query_row("SELECT count(*) FROM sqlite_master WHERE name IN ('remote_exec_policies','idx_remote_exec_policies_used')", [], |r| r.get::<_,u32>(0)).map(|n| n == 2), + 24 => conn.query_row("SELECT count(*) FROM sqlite_master WHERE type='table' AND name IN ('bash_watch_delivery_acks','bash_watch_delivery_gaps')", [], |row| row.get::<_, u32>(0)).map(|count| count == 2), _ => Ok(false), } } @@ -1024,6 +1026,16 @@ fn apply_migration_statements(conn: &Connection, version: u32) -> rusqlite::Resu 21 => apply_migration_v21(conn), 22 => conn.execute_batch(gh_speech::MIGRATION), 23 => conn.execute_batch(gh_speech::LEGACY_DISPATCH_MIGRATION), + 24 => conn.execute_batch("CREATE TABLE IF NOT EXISTS bash_watch_delivery_acks ( + harness TEXT NOT NULL, session_id TEXT NOT NULL, task_id TEXT NOT NULL, watch_id TEXT NOT NULL, + version INTEGER NOT NULL, acknowledged_at INTEGER NOT NULL, + eviction_epoch TEXT NOT NULL, evicted_through INTEGER NOT NULL, + PRIMARY KEY (harness, session_id, task_id, watch_id) + ); + CREATE TABLE IF NOT EXISTS bash_watch_delivery_gaps ( + harness TEXT NOT NULL, session_id TEXT NOT NULL, task_id TEXT NOT NULL, watch_id TEXT NOT NULL, + batch TEXT NOT NULL, PRIMARY KEY (harness, session_id, task_id, watch_id) + )"), _ => Ok(()), } } @@ -1070,7 +1082,7 @@ mod tests { apply_migration(&mut conn, version).unwrap(); } assert_eq!(current_schema_version(&conn).unwrap(), 21); - assert_eq!(run_migrations(&mut conn).unwrap(), 23); + assert_eq!(run_migrations(&mut conn).unwrap(), CURRENT_SCHEMA_VERSION); for table in [ "gh_speech_authorities", "gh_speech_task_revocations", @@ -1089,7 +1101,7 @@ mod tests { .unwrap(); assert!(present, "missing {table}"); } - assert_eq!(MIGRATION_VERSIONS, (1..=23).collect::>()); + assert_eq!(MIGRATION_VERSIONS, (1..=24).collect::>()); } #[test] @@ -1106,7 +1118,7 @@ mod tests { absent, "schema 22 must not silently acquire schema 23 tables" ); - assert_eq!(run_migrations(&mut conn).unwrap(), 23); + assert_eq!(run_migrations(&mut conn).unwrap(), CURRENT_SCHEMA_VERSION); let reason: String = conn .query_row( "SELECT reason FROM gh_speech_task_revocations WHERE incarnation='old-task'", @@ -1124,6 +1136,52 @@ mod tests { .unwrap(); assert_eq!(count, 0); } + #[test] + fn schema_twenty_three_upgrade_adds_watch_gap_state_after_speech_authority() { + let mut conn = Connection::open_in_memory().unwrap(); + conn.execute_batch("CREATE TABLE schema_version (version INTEGER NOT NULL PRIMARY KEY)") + .unwrap(); + for version in 1..=23 { + apply_migration(&mut conn, version).unwrap(); + } + conn.execute("INSERT INTO gh_speech_task_revocations(incarnation,reason,recorded_at_ms) VALUES('old-task','cancelled',1)", []).unwrap(); + assert!(!migration_already_applied(&conn, 24).unwrap()); + let count: u32 = conn.query_row("SELECT COUNT(*) FROM sqlite_master WHERE name IN ('bash_watch_delivery_acks','bash_watch_delivery_gaps')", [], |row| row.get(0)).unwrap(); + assert_eq!(count, 0); + assert_eq!(run_migrations(&mut conn).unwrap(), 24); + assert!(migration_already_applied(&conn, 24).unwrap()); + bash_watches::acknowledge_delivery(&conn, "opencode", "session", "task", "watch", 7, None) + .unwrap(); + assert_eq!( + bash_watches::delivery_ack_version(&conn, "opencode", "session", "task", "watch") + .unwrap() + .unwrap() + .0, + 7 + ); + assert!( + bash_watches::list_delivery_gaps(&conn, "opencode", Some("session")) + .unwrap() + .is_empty() + ); + let reason: String = conn + .query_row( + "SELECT reason FROM gh_speech_task_revocations WHERE incarnation='old-task'", + [], + |row| row.get(0), + ) + .unwrap(); + assert_eq!(reason, "cancelled"); + let dispatch_count: u32 = conn + .query_row( + "SELECT COUNT(*) FROM gh_speech_legacy_dispatch", + [], + |row| row.get(0), + ) + .unwrap(); + assert_eq!(dispatch_count, 0); + } + use rusqlite::params; use tempfile::tempdir; @@ -1131,6 +1189,8 @@ mod tests { "schema_version", "bash_tasks", "bash_pattern_watches", + "bash_watch_delivery_acks", + "bash_watch_delivery_gaps", "compression_events", "backups", "harness_state", diff --git a/crates/aft/src/subc/health.rs b/crates/aft/src/subc/health.rs index 395b399f4..4cf4fee04 100644 --- a/crates/aft/src/subc/health.rs +++ b/crates/aft/src/subc/health.rs @@ -558,6 +558,7 @@ struct FrameOperation { } pub(super) struct DispatchPathMetrics { + pub(super) watch_delivery: Arc, unanswered_tools: StdMutex>, pub(super) origin: Instant, pub(super) frame_loop_last_tick_ms: AtomicU64, @@ -620,6 +621,7 @@ pub(super) struct DispatchPathMetrics { impl DispatchPathMetrics { pub(super) fn new() -> Self { Self { + watch_delivery: Arc::default(), unanswered_tools: StdMutex::new(HashMap::new()), origin: Instant::now(), frame_loop_last_tick_ms: AtomicU64::new(0), @@ -1423,6 +1425,7 @@ impl DispatchPathMetrics { "oldest_age_ms": oldest_pending_age_ms, }, "bind_acks": self.bind_ack_snapshot(), + "reliable_push": self.watch_delivery.health(), "tool_call_presets": self.tool_call_preset_snapshot(), "completion_channels": { "control": self.control_completion_queued.load(Ordering::Relaxed), @@ -3971,7 +3974,14 @@ fn build_health_report_with_snapshots( metrics.insert("headline".to_string(), json!(headline)); budget_health_metrics(&mut metrics); - let warning = if scheduler_busy { + let gap_health = dispatch_path_metrics.watch_delivery.health(); + let watch_gap_unresolved = gap_health["status"] != "ok"; + let warning = if watch_gap_unresolved { + Some(format!( + "watch delivery is degraded: {}", + gap_health["reason"].as_str().unwrap_or("retry_overflow") + )) + } else if scheduler_busy { Some(EXECUTOR_SCHEDULER_CONTENTION_DETAIL.to_string()) } else { rollup.detail.clone() @@ -3984,7 +3994,7 @@ fn build_health_report_with_snapshots( (None, warning) => warning, }; HealthReport { - status: if scheduler_busy { + status: if scheduler_busy || watch_gap_unresolved { HealthStatus::Degraded } else { rollup.status.clone() @@ -9559,3 +9569,75 @@ mod observability_retention_tests { assert!(events.keys().all(|key| key.session != "inactive")); } } + +#[cfg(test)] +mod watch_gap_tests { + + #[test] + fn retry_overflow_health_keeps_evictions_separate_and_clears_after_ack() { + use crate::bash_background::watch_lines::{LineBatch, RetryGap}; + use crate::db::bash_watches::*; + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("aft.db"); + let conn = crate::db::open(&path).unwrap(); + let batch = LineBatch { + retry_gap: Some(RetryGap { + reason: "retry_overflow".into(), + dropped_notifications: 3, + known_missing_occurrences: 1, + ..Default::default() + }), + version: Some(5), + ..Default::default() + }; + let row = BashPatternWatchRow { + harness: "opencode".into(), + session_id: "session".into(), + task_id: "task".into(), + watch_id: "watch".into(), + pattern_kind: "substring".into(), + pattern: "READY".into(), + once: false, + each_line: false, + match_batch: Some(serde_json::to_string(&batch).unwrap()), + created_at: 0, + stdout_offset: 0, + stderr_offset: 0, + pty_offset: 0, + scanning: true, + pending_match: true, + match_text: Some("watch delivery gap".into()), + match_offset: Some(5), + match_context: None, + }; + upsert_delivery_gap(&conn, &row).unwrap(); + let (gaps, missing, uncertain) = watch_gap_health_for(&conn); + assert_eq!((gaps, missing, uncertain), (1, 1, false)); + let health = super::super::watch_delivery::health_from_census(gaps, missing, uncertain, 0); + assert_eq!(health["status"], "degraded"); + assert_eq!(health["reason"], "retry_overflow"); + assert_eq!(health["outstanding_watch_gaps"], 1); + assert_eq!(health["known_missing_occurrences"], 1); + assert!(health["evicted_transport_copies_total"].is_u64()); + acknowledge_delivery( + &conn, + "opencode", + "session", + "task", + "watch", + 5, + batch.retry_gap.as_ref(), + ) + .unwrap(); + delete_delivery_gap(&conn, "opencode", "session", "task", "watch").unwrap(); + let (gaps, missing, uncertain) = watch_gap_health_for(&conn); + let cleared = super::super::watch_delivery::health_from_census(gaps, missing, uncertain, 0); + assert_eq!(cleared["status"], "ok"); + assert!(cleared["reason"].is_null()); + assert_eq!(cleared["outstanding_watch_gaps"], 0); + assert_eq!(cleared["known_missing_occurrences"], 0); + let unconfirmed = super::super::watch_delivery::health_from_census(0, 0, true, 1); + assert_eq!(unconfirmed["status"], "degraded"); + assert_eq!(unconfirmed["reason"], "watch_gap_persistence_unconfirmed"); + } +} diff --git a/crates/aft/src/subc/mod.rs b/crates/aft/src/subc/mod.rs index a9274b3e6..d72546a9a 100644 --- a/crates/aft/src/subc/mod.rs +++ b/crates/aft/src/subc/mod.rs @@ -253,6 +253,7 @@ mod remote_policy; pub(crate) mod result_body; mod retained_result; mod rich_ui; +pub(crate) mod watch_delivery; // Drives the chunk queries and Occipital's route shapes through the real // frame loop, so it lives beside the chunk module but runs as a child of this // module to reach the loop and its test fixtures. @@ -4035,7 +4036,8 @@ async fn teardown_installed_route( channel.channel ); } - let migrated = push::migrate_retry_buffer_to_push_buffer(retry_buffer, channel, push_buffer); + let migrated = + push::migrate_retry_buffer_to_push_buffer(metrics, retry_buffer, channel, push_buffer); if let Some(identity) = routes.get(&channel) { let has_pending_bind = pending_binds .values() @@ -5121,6 +5123,7 @@ where }; let dispatch_path_metrics = Arc::new(DispatchPathMetrics::new()); + dispatch_path_metrics.watch_delivery.install(&executor); if let Some(probe) = lifecycle_probe.clone() { *dispatch_path_metrics .lifecycle_probe @@ -5490,6 +5493,7 @@ where &routes, &mut retry_buffer, ); + dispatch_path_metrics.watch_delivery.kick(); if retried > 0 { log::debug!( "subc attach: retried {retried} reliable Push frame(s) after writer backpressure" diff --git a/crates/aft/src/subc/push.rs b/crates/aft/src/subc/push.rs index 45640bed4..630a1a2d6 100644 --- a/crates/aft/src/subc/push.rs +++ b/crates/aft/src/subc/push.rs @@ -593,9 +593,13 @@ pub(super) fn clear_stale_bg_wakes_for_empty_sessions( static RELIABLE_PUSH_EVICTIONS: AtomicU64 = AtomicU64::new(0); -fn bounded_push_back(queue: &mut VecDeque, item: T) { - if queue.len() >= PUSH_BUFFER_MAX_PER_KEY { - queue.pop_front(); +pub(super) fn reliable_push_evictions() -> u64 { + RELIABLE_PUSH_EVICTIONS.load(Ordering::Relaxed) +} + +fn bounded_push_back(queue: &mut VecDeque, item: T) -> Option { + let evicted = if queue.len() >= PUSH_BUFFER_MAX_PER_KEY { + let evicted = queue.pop_front(); let total = RELIABLE_PUSH_EVICTIONS .fetch_add(1, Ordering::Relaxed) .saturating_add(1); @@ -603,28 +607,41 @@ fn bounded_push_back(queue: &mut VecDeque, item: T) { if total.is_power_of_two() { log::warn!("subc reliable Push buffer overflow: evicted_total={total}"); } - } + evicted + } else { + None + }; queue.push_back(item); + evicted } fn buffer_push_frame( + metrics: &DispatchPathMetrics, push_buffer: &mut HashMap>, key: ReplayKey, frame: PushFrame, ) { - bounded_push_back(push_buffer.entry(key).or_default(), frame); + if let Some(evicted) = bounded_push_back(push_buffer.entry(key.clone()).or_default(), frame) { + metrics.watch_delivery.evicted(key, evicted); + } } -fn buffer_retry_frame( +pub(super) fn buffer_retry_frame( + metrics: &DispatchPathMetrics, retry_buffer: &mut RetryBuffer, channel: RouteChannel, key: ReplayKey, frame: PushFrame, ) { - bounded_push_back(retry_buffer.entry(channel).or_default(), (key, frame)); + if let Some((key, frame)) = + bounded_push_back(retry_buffer.entry(channel).or_default(), (key, frame)) + { + metrics.watch_delivery.evicted(key, frame); + } } pub(super) fn migrate_retry_buffer_to_push_buffer( + metrics: &DispatchPathMetrics, retry_buffer: &mut RetryBuffer, channel: RouteChannel, push_buffer: &mut HashMap>, @@ -634,7 +651,7 @@ pub(super) fn migrate_retry_buffer_to_push_buffer( }; let migrated = frames.len(); for (key, frame) in frames { - buffer_push_frame(push_buffer, key, frame); + buffer_push_frame(metrics, push_buffer, key, frame); } migrated } @@ -814,6 +831,7 @@ fn matching_session_route_channels( } fn buffer_detached_reliable_push_frame( + metrics: &DispatchPathMetrics, push_buffer: &mut HashMap>, session_identity: &HashMap<(ProjectRootId, String), RetainedSessionIdentity>, root: &ProjectRootId, @@ -832,7 +850,7 @@ fn buffer_detached_reliable_push_frame( if frame_is_bash_observation(frame) && !trust.allows_bash_observation() { return; } - buffer_push_frame(push_buffer, key, frame.clone()); + buffer_push_frame(metrics, push_buffer, key, frame.clone()); if let (Some(probe), Some(task_id)) = (lifecycle_probe, completed_task_id(frame)) { probe.reliable_completion_retained(task_id, session); } @@ -981,6 +999,7 @@ fn fan_out_reliable_push_frame( let matched_channels = matching_channels.len(); if matched_channels == 0 { buffer_detached_reliable_push_frame( + metrics, push_buffer, session_identity, root, @@ -1004,14 +1023,14 @@ fn fan_out_reliable_push_frame( .get(&channel) .is_some_and(|queue| !queue.is_empty()) { - buffer_retry_frame(retry_buffer, channel, key, frame.clone()); + buffer_retry_frame(metrics, retry_buffer, channel, key, frame.clone()); continue; } match try_send_push_frame(writer_tx, metrics, channel, frame) { PushSendOutcome::Sent => sent_frames += 1, PushSendOutcome::Backpressure => { - buffer_retry_frame(retry_buffer, channel, key, frame.clone()); + buffer_retry_frame(metrics, retry_buffer, channel, key, frame.clone()); } PushSendOutcome::PermanentFailure => { log::warn!( @@ -2183,6 +2202,7 @@ mod tests { for index in 0..total { buffer_push_frame( + &DispatchPathMetrics::new(), &mut push_buffer, key.clone(), completion_frame(&format!("task-{index}")), @@ -2214,8 +2234,18 @@ mod tests { let (writer_tx, mut writer_rx) = mpsc::channel::(4); let metrics = DispatchPathMetrics::new(); let mut push_buffer = HashMap::new(); - buffer_push_frame(&mut push_buffer, key.clone(), completion_frame("task-a")); - buffer_push_frame(&mut push_buffer, key.clone(), completion_frame("task-b")); + buffer_push_frame( + &DispatchPathMetrics::new(), + &mut push_buffer, + key.clone(), + completion_frame("task-a"), + ); + buffer_push_frame( + &DispatchPathMetrics::new(), + &mut push_buffer, + key.clone(), + completion_frame("task-b"), + ); let replayed = replay_buffered_push_frames( &writer_tx, @@ -2250,7 +2280,12 @@ mod tests { let (writer_tx, mut writer_rx) = mpsc::channel::(4); let metrics = DispatchPathMetrics::new(); let mut push_buffer = HashMap::new(); - buffer_push_frame(&mut push_buffer, key.clone(), completion_frame("task-a")); + buffer_push_frame( + &DispatchPathMetrics::new(), + &mut push_buffer, + key.clone(), + completion_frame("task-a"), + ); let replayed = replay_buffered_push_frames( &writer_tx, @@ -2941,7 +2976,12 @@ mod tests { .expect("prefill writer queue"); let mut push_buffer = HashMap::new(); for task in ["replay-1", "replay-2", "replay-3"] { - buffer_push_frame(&mut push_buffer, key.clone(), completion_frame(task)); + buffer_push_frame( + &DispatchPathMetrics::new(), + &mut push_buffer, + key.clone(), + completion_frame(task), + ); } assert_eq!( @@ -2997,6 +3037,7 @@ mod tests { }; let mut retry_buffer = HashMap::new(); buffer_retry_frame( + &DispatchPathMetrics::new(), &mut retry_buffer, route_key(5, 1), key.clone(), @@ -3006,6 +3047,7 @@ mod tests { assert_eq!( migrate_retry_buffer_to_push_buffer( + &DispatchPathMetrics::new(), &mut retry_buffer, route_key(5, 1), &mut push_buffer @@ -3035,6 +3077,7 @@ mod tests { let mut push_buffer = HashMap::new(); buffer_push_frame( + &DispatchPathMetrics::new(), &mut push_buffer, key.clone(), completion_frame("closed-replay"), @@ -3055,6 +3098,7 @@ mod tests { let mut retry_buffer = HashMap::new(); buffer_retry_frame( + &DispatchPathMetrics::new(), &mut retry_buffer, route_key(4, 1), key, diff --git a/crates/aft/src/subc/watch_delivery.rs b/crates/aft/src/subc/watch_delivery.rs new file mode 100644 index 000000000..1b8b708bf --- /dev/null +++ b/crates/aft/src/subc/watch_delivery.rs @@ -0,0 +1,300 @@ +//! Coalesced retry-eviction work. SQLite never runs on the SubC frame loop. +use super::*; +use crate::protocol::BashPatternMatchFrame; +use std::sync::{Mutex, Weak}; + +static PENDING_CHECKS: AtomicUsize = AtomicUsize::new(0); +static NEXT_EVICTION: AtomicU64 = AtomicU64::new(1); +static EVICTION_EPOCH: std::sync::OnceLock = std::sync::OnceLock::new(); + +pub(crate) fn eviction_epoch() -> &'static str { + EVICTION_EPOCH.get_or_init(|| { + format!( + "{}:{}", + std::process::id(), + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_nanos() + ) + }) +} + +#[derive(Clone)] +pub(crate) struct Candidate { + pub eviction_epoch: String, + pub last_eviction: u64, + pub fingerprint: Vec, + pub frame: BashPatternMatchFrame, + pub notifications: u64, + pub first_version: u64, + pub last_version: u64, + pub first_offset: Option, + pub last_offset: Option, + pub stream: Option, + pub same_generation: bool, + pub same_occurrence: bool, +} + +impl Candidate { + fn new(mut frame: BashPatternMatchFrame) -> Self { + use sha2::Digest; + let fingerprint = sha2::Sha256::digest(frame.match_text.as_bytes()).to_vec(); + let (first_offset, last_offset, stream) = if let Some(batch) = &frame.batch { + let stream = batch.lines.first().map(|line| line.stream.clone()); + if batch + .lines + .iter() + .all(|line| Some(&line.stream) == stream.as_ref()) + { + ( + batch + .lines + .first() + .map(|line| line.offset) + .or(batch.first_dropped_offset), + batch.lines.last().map(|line| { + line.offset.saturating_add( + line.count + .saturating_mul(line.end_offset.saturating_sub(line.offset)), + ) + }), + stream.or(batch.first_dropped_stream.clone()), + ) + } else { + (None, None, None) + } + } else { + (Some(frame.match_offset), Some(frame.match_offset), None) + }; + let version = frame.match_offset; + frame.match_text = String::new(); + frame.context = String::new(); + if let Some(batch) = &mut frame.batch { + batch.lines = Vec::new(); + } + Self { + eviction_epoch: eviction_epoch().to_string(), + last_eviction: NEXT_EVICTION.fetch_add(1, Ordering::Relaxed), + fingerprint, + frame, + notifications: 1, + first_version: version, + last_version: version, + first_offset, + last_offset, + stream, + same_generation: true, + same_occurrence: true, + } + } + + fn merge(&mut self, next: Self) { + self.same_occurrence &= next.same_occurrence + && self.fingerprint == next.fingerprint + && self.first_version == next.first_version; + self.last_eviction = self.last_eviction.max(next.last_eviction); + self.notifications = self.notifications.saturating_add(next.notifications); + self.first_version = self.first_version.min(next.first_version); + self.last_version = self.last_version.max(next.last_version); + self.same_generation &= next.same_generation + && self + .frame + .batch + .as_ref() + .and_then(|b| b.pending_id.as_ref()) + == next + .frame + .batch + .as_ref() + .and_then(|b| b.pending_id.as_ref()); + if self.stream == next.stream { + self.first_offset = self + .first_offset + .zip(next.first_offset) + .map(|(a, b)| a.min(b)); + self.last_offset = self + .last_offset + .zip(next.last_offset) + .map(|(a, b)| a.max(b)); + } else { + self.first_offset = None; + self.last_offset = None; + self.stream = None; + } + } +} + +type Key = (push::ReplayKey, String, String); + +#[derive(Default)] +pub(super) struct WatchDelivery { + pending: Mutex>, + executor: Mutex>, + running: AtomicBool, + in_flight: AtomicUsize, +} + +impl WatchDelivery { + pub(super) fn install(&self, executor: &Arc) { + *self.executor.lock().unwrap() = Arc::downgrade(executor); + } + + pub(super) fn evicted(self: &Arc, key: push::ReplayKey, frame: PushFrame) { + let PushFrame::BashPatternMatch(frame) = frame else { + return; + }; + if frame.reason == "task_exit" { + return; + } + let key = (key, frame.task_id.clone(), frame.watch_id.clone()); + let candidate = Candidate::new(frame); + self.pending + .lock() + .unwrap() + .entry(key) + .and_modify(|old| old.merge(candidate.clone())) + .or_insert_with(|| { + PENDING_CHECKS.fetch_add(1, Ordering::Relaxed); + candidate + }); + self.kick(); + } + + pub(super) fn health(&self) -> serde_json::Value { + let attached = self.executor.lock().unwrap().upgrade().is_some(); + let (gaps, missing, uncertain) = if attached { + crate::db::bash_watches::watch_gap_health() + } else { + (0, 0, false) + }; + let pending = if attached { + PENDING_CHECKS.load(Ordering::Relaxed) + } else { + 0 + }; + health_from_census(gaps, missing, uncertain, pending) + } + + pub(super) fn kick(self: &Arc) { + let Some(executor) = self.executor.lock().unwrap().upgrade() else { + return; + }; + if self.pending.lock().unwrap().is_empty() || self.running.swap(true, Ordering::AcqRel) { + return; + } + let worker = self.clone(); + tokio::task::spawn_blocking(move || { + let candidates = { + let mut pending = worker.pending.lock().unwrap(); + worker.in_flight.store(pending.len(), Ordering::Relaxed); + std::mem::take(&mut *pending) + }; + for (key, candidate) in candidates { + let result = executor + .actor_context(&key.0.root) + .ok_or_else(|| "watch registry unavailable".to_string()) + .and_then(|ctx| { + ctx.bash_background() + .retain_retry_gap(&key.0.harness, &candidate) + }); + if let Err(error) = result { + log::warn!("retaining watch retry gap failed: {error}"); + worker + .pending + .lock() + .unwrap() + .entry(key) + .and_modify(|new| { + new.merge(candidate.clone()); + PENDING_CHECKS.fetch_sub(1, Ordering::Relaxed); + }) + .or_insert(candidate); + } else { + PENDING_CHECKS.fetch_sub(1, Ordering::Relaxed); + } + worker.in_flight.fetch_sub(1, Ordering::Relaxed); + } + worker.running.store(false, Ordering::Release); + if !worker.pending.lock().unwrap().is_empty() { + tokio::spawn(async move { + tokio::time::sleep(Duration::from_secs(1)).await; + worker.kick(); + }); + } + }); + } +} + +pub(super) fn health_from_census( + gaps: u64, + missing: u64, + uncertain: bool, + pending: usize, +) -> serde_json::Value { + let reason = if gaps > 0 { + Some("retry_overflow") + } else if pending > 0 || uncertain { + Some("watch_gap_persistence_unconfirmed") + } else { + None + }; + serde_json::json!({ + "evicted_transport_copies_total": push::reliable_push_evictions(), + "outstanding_watch_gaps": gaps, + "known_missing_occurrences": missing, + "pending_gap_checks": pending, + "status": if reason.is_some() { "degraded" } else { "ok" }, + "reason": reason, + }) +} + +#[cfg(test)] +pub(crate) fn stalled_route_eviction(frame: BashPatternMatchFrame) -> Option { + let metrics = DispatchPathMetrics::new(); + let key = push::ReplayKey { + root: ProjectRootId::from_path(std::env::current_dir().unwrap()).unwrap(), + harness: "opencode".to_string(), + session: frame.session_id.clone(), + }; + let mut retry = RetryBuffer::new(); + let channel = RouteChannel { + channel: 1, + epoch: 1, + }; + push::buffer_retry_frame( + &metrics, + &mut retry, + channel, + key.clone(), + PushFrame::BashPatternMatch(frame), + ); + for _ in 0..PUSH_BUFFER_MAX_PER_KEY { + let mut filler = crate::protocol::BashPatternMatchFrame::task_exit( + "filler", + &key.session, + "exit", + "exit", + ); + filler.once = false; + push::buffer_retry_frame( + &metrics, + &mut retry, + channel, + key.clone(), + PushFrame::BashPatternMatch(filler), + ); + } + let candidate = metrics + .watch_delivery + .pending + .lock() + .unwrap() + .drain() + .next() + .map(|(_, c)| c); + if candidate.is_some() { + PENDING_CHECKS.fetch_sub(1, Ordering::Relaxed); + } + candidate +} diff --git a/crates/aft/src/subc_format.rs b/crates/aft/src/subc_format.rs index 17ebd1cab..ed6552304 100644 --- a/crates/aft/src/subc_format.rs +++ b/crates/aft/src/subc_format.rs @@ -5502,13 +5502,46 @@ pub fn format_watch_line_batch( }; text.push_str(&format!("{} matching lines dropped.{incomplete} Read them with bash_status({{ task_id: {}, {cursor}: {} }}).\n", batch.dropped_lines, serde_json::to_string(task_id).unwrap_or_default(), batch.first_dropped_offset.unwrap_or(0))); } + if let Some(gap) = &batch.retry_gap { + let missing = match gap.known_missing_occurrences { + 0 => "Some matches may be missing.".to_string(), + 1 => "At least 1 match is missing.".to_string(), + count => format!("At least {count} matches are missing."), + }; + let cursor = if gap.stream.as_deref() == Some("stderr") { + "stderr_offset" + } else { + "output_offset" + }; + text.push_str(&format!("Some watch notices were lost: a stalled connection dropped {} of them, and they cannot be recovered. {missing} Read the task output with bash_status({{ task_id: {}, {cursor}: {} }}).\n", gap.dropped_notifications, serde_json::to_string(task_id).unwrap_or_default(), gap.first_offset.unwrap_or(0))); + } text.trim_end().to_string() } #[cfg(test)] mod watch_line_format_tests { use super::format_watch_line_batch; - use crate::bash_background::watch_lines::{LineBatch, WatchLine}; + use crate::bash_background::watch_lines::{LineBatch, RetryGap, WatchLine}; + + #[test] + fn retry_gap_wording_golden() { + for (notices, missing, expected) in [ + (2, 1, "Some watch notices were lost: a stalled connection dropped 2 of them, and they cannot be recovered. At least 1 match is missing. Read the task output with bash_status({ task_id: \"bash-gap\", stderr_offset: 16 })."), + (1, 0, "Some watch notices were lost: a stalled connection dropped 1 of them, and they cannot be recovered. Some matches may be missing. Read the task output with bash_status({ task_id: \"bash-gap\", stderr_offset: 16 })."), + (1, 1, "Some watch notices were lost: a stalled connection dropped 1 of them, and they cannot be recovered. At least 1 match is missing. Read the task output with bash_status({ task_id: \"bash-gap\", stderr_offset: 16 })."), + (2, 3, "Some watch notices were lost: a stalled connection dropped 2 of them, and they cannot be recovered. At least 3 matches are missing. Read the task output with bash_status({ task_id: \"bash-gap\", stderr_offset: 16 })."), + ] { + let batch = LineBatch { + retry_gap: Some(RetryGap { + reason: "retry_overflow".into(), dropped_notifications: notices, + known_missing_occurrences: missing, first_offset: Some(16), stream: Some("stderr".into()), + ..Default::default() + }), + ..Default::default() + }; + assert_eq!(format_watch_line_batch("bash-gap", &batch), expected); + } + } #[test] fn each_line_row_wording_golden() { diff --git a/crates/aft/tests/scope_end_kill.rs b/crates/aft/tests/scope_end_kill.rs index a3eda70da..9d0250497 100644 --- a/crates/aft/tests/scope_end_kill.rs +++ b/crates/aft/tests/scope_end_kill.rs @@ -766,7 +766,10 @@ async fn refused_shell_start_fails_fast_with_the_terminal_reason() { let mut request = shell(None, false, 15); request["arguments"]["env"] = json!({"PATH":"blocked"}); let corr = daemon.start(1, 1, request).await; - tokio::time::timeout(Duration::from_secs(2), running_task(&mut daemon, 1, corr)) + // The refusal must arrive as its own terminal reply, well before the 15 s + // startup timeout; 10 s keeps that proof while a loaded macOS runner, which + // took over 2 s, still fits. + tokio::time::timeout(Duration::from_secs(10), running_task(&mut daemon, 1, corr)) .await .expect("startup refusal was ignored instead of reporting its terminal reason"); } diff --git a/packages/aft-bridge/src/__tests__/bash-watch-progress.test.ts b/packages/aft-bridge/src/__tests__/bash-watch-progress.test.ts index 36ed47996..057d24794 100644 --- a/packages/aft-bridge/src/__tests__/bash-watch-progress.test.ts +++ b/packages/aft-bridge/src/__tests__/bash-watch-progress.test.ts @@ -38,6 +38,42 @@ test("watch progress dedups folded lines and cumulative counters after admission expect(unseen.skipped_long_lines).toBe(1); }); +test("retry gap-only batches are admitted once while newer gap versions remain visible", () => { + const progress = new WatchBatchProgress(); + const first: BashWatchLineBatch = { + ...snapshot(), + lines: [], + dropped_lines: 0, + skipped_long_lines: 0, + retry_gap: { reason: "retry_overflow", dropped_notifications: 2, known_missing_occurrences: 1 }, + }; + expect(watchBatchHasContent(progress.unseen(first))).toBe(true); + progress.admit(first); + expect(watchBatchHasContent(progress.unseen(first))).toBe(false); + const next = { + ...first, + version: 2, + retry_gap: { ...first.retry_gap!, dropped_notifications: 3, known_missing_occurrences: 2 }, + }; + expect(progress.unseen(next).retry_gap?.dropped_notifications).toBe(1); + expect(progress.unseen(next).retry_gap?.known_missing_occurrences).toBe(1); + progress.admit(next); + expect(watchBatchHasContent(progress.unseen(next))).toBe(false); +}); + +test("legacy scalar output retained beside a retry gap is not repeated", () => { + const progress = new WatchBatchProgress(); + const batch = snapshot(); + batch.lines[0].stream = "output"; + batch.retry_gap = { + reason: "retry_overflow", + dropped_notifications: 1, + known_missing_occurrences: 0, + }; + progress.admit(batch); + expect(watchBatchHasContent(progress.unseen(batch))).toBe(false); +}); + test("watch progress does not reserve rows before admission or across pending generations", () => { const progress = new WatchBatchProgress(); expect(progress.unseen(snapshot()).lines).toHaveLength(1); diff --git a/packages/aft-bridge/src/bash-watch-lines.ts b/packages/aft-bridge/src/bash-watch-lines.ts index 307237e62..076594e85 100644 --- a/packages/aft-bridge/src/bash-watch-lines.ts +++ b/packages/aft-bridge/src/bash-watch-lines.ts @@ -27,6 +27,7 @@ export function mergeWatchLineBatches( merged.masking_limit_bytes = next.masking_limit_bytes ?? previous.masking_limit_bytes; merged.pending_id = next.pending_id ?? previous.pending_id; merged.version = next.version ?? previous.version; + if (next.retry_gap) merged.retry_gap = next.retry_gap; let retained = merged.lines.reduce((sum, line) => sum + line.count, 0); for (const line of next.lines) { const keep = Math.min(line.count, MAX_PENDING_WATCH_LINES - retained); @@ -82,6 +83,21 @@ export function formatWatchLineBatch( `${batch.dropped_lines} matching lines dropped.${incomplete} Read them with bash_status({ ${taskKey}: ${JSON.stringify(taskId)}, ${offsetKey}: ${batch.first_dropped_offset ?? 0} }).`, ); } + if (batch.retry_gap && batch.retry_gap.dropped_notifications > 0) { + const gap = batch.retry_gap; + const taskKey = camelCase ? "taskId" : "task_id"; + const cursor = gap.stream === "stderr" ? "stderr" : "output"; + const offsetKey = camelCase ? `${cursor}Offset` : `${cursor}_offset`; + const missing = + gap.known_missing_occurrences === 0 + ? "Some matches may be missing." + : gap.known_missing_occurrences === 1 + ? "At least 1 match is missing." + : `At least ${gap.known_missing_occurrences} matches are missing.`; + lines.push( + `Some watch notices were lost: a stalled connection dropped ${gap.dropped_notifications} of them, and they cannot be recovered. ${missing} Read the task output with bash_status({ ${taskKey}: ${JSON.stringify(taskId)}, ${offsetKey}: ${gap.first_offset ?? 0} }).`, + ); + } return lines.join("\n"); } diff --git a/packages/aft-bridge/src/bash-watch-progress.ts b/packages/aft-bridge/src/bash-watch-progress.ts index 3c5f69b11..ca5df986d 100644 --- a/packages/aft-bridge/src/bash-watch-progress.ts +++ b/packages/aft-bridge/src/bash-watch-progress.ts @@ -1,19 +1,21 @@ import type { BashAckTarget } from "./bash-pattern-ack.js"; import type { BashWatchLineBatch } from "./bridge.js"; -type Stream = "stdout" | "stderr" | "pty"; +type Stream = "stdout" | "stderr" | "pty" | "output"; function knownStream(stream: string): stream is Stream { - return stream === "stdout" || stream === "stderr" || stream === "pty"; + return stream === "stdout" || stream === "stderr" || stream === "pty" || stream === "output"; } /** Admission watermarks for one pending watch generation; no line text is retained. */ export class WatchBatchProgress { private pendingID: string | undefined; private version: number | undefined; - private ends: Record = { stdout: 0, stderr: 0, pty: 0 }; + private ends: Record = { stdout: 0, stderr: 0, pty: 0, output: 0 }; private dropped = 0; private skipped = 0; + private gapNotices = 0; + private gapOccurrences = 0; get admittedVersion(): number | undefined { return this.version; @@ -50,6 +52,20 @@ export class WatchBatchProgress { skipped_long_lines: obsolete ? 0 : Math.max(0, batch.skipped_long_lines - (same && batch.snapshot ? this.skipped : 0)), + retry_gap: batch.retry_gap + ? { + ...batch.retry_gap, + dropped_notifications: obsolete + ? 0 + : Math.max(0, batch.retry_gap.dropped_notifications - (same ? this.gapNotices : 0)), + known_missing_occurrences: obsolete + ? 0 + : Math.max( + 0, + batch.retry_gap.known_missing_occurrences - (same ? this.gapOccurrences : 0), + ), + } + : undefined, }; } @@ -58,9 +74,11 @@ export class WatchBatchProgress { if (this.pendingID !== pendingID) { this.pendingID = pendingID; this.version = undefined; - this.ends = { stdout: 0, stderr: 0, pty: 0 }; + this.ends = { stdout: 0, stderr: 0, pty: 0, output: 0 }; this.dropped = 0; this.skipped = 0; + this.gapNotices = 0; + this.gapOccurrences = 0; } if ( !batch.snapshot && @@ -81,13 +99,23 @@ export class WatchBatchProgress { this.skipped = batch.snapshot ? Math.max(this.skipped, batch.skipped_long_lines) : this.skipped + batch.skipped_long_lines; + this.gapNotices = Math.max(this.gapNotices, batch.retry_gap?.dropped_notifications ?? 0); + this.gapOccurrences = Math.max( + this.gapOccurrences, + batch.retry_gap?.known_missing_occurrences ?? 0, + ); if (batch.version !== undefined) this.version = Math.max(this.version ?? batch.version, batch.version); } } export function watchBatchHasContent(batch: BashWatchLineBatch): boolean { - return batch.lines.length > 0 || batch.dropped_lines > 0 || batch.skipped_long_lines > 0; + return ( + batch.lines.length > 0 || + batch.dropped_lines > 0 || + batch.skipped_long_lines > 0 || + (batch.retry_gap?.dropped_notifications ?? 0) > 0 + ); } type BatchMatch = { diff --git a/packages/aft-bridge/src/bridge.ts b/packages/aft-bridge/src/bridge.ts index 564a85350..973feda58 100644 --- a/packages/aft-bridge/src/bridge.ts +++ b/packages/aft-bridge/src/bridge.ts @@ -501,6 +501,18 @@ export interface BashLongRunningPayload { } export interface BashWatchLineBatch { + retry_gap?: { + reason: "retry_overflow"; + dropped_notifications: number; + known_missing_occurrences: number; + first_offset?: number; + last_offset?: number; + stream?: string; + evicted_through?: number; + eviction_epoch?: string; + last_match_version?: number; + }; + acknowledged_version?: number; pending_id?: string; version?: number; terminal_summary?: boolean; diff --git a/packages/opencode-plugin/src/__tests__/bg-notifications.test.ts b/packages/opencode-plugin/src/__tests__/bg-notifications.test.ts index 3ae08e164..b22f1ef0c 100644 --- a/packages/opencode-plugin/src/__tests__/bg-notifications.test.ts +++ b/packages/opencode-plugin/src/__tests__/bg-notifications.test.ts @@ -2707,6 +2707,58 @@ function preAdmissionRejection(message: string) { }; } +test.each([ + [ + 2, + 1, + 'Some watch notices were lost: a stalled connection dropped 2 of them, and they cannot be recovered. At least 1 match is missing. Read the task output with bash_status({ taskId: "bash-gap", stderrOffset: 16 }).', + ], + [ + 1, + 0, + 'Some watch notices were lost: a stalled connection dropped 1 of them, and they cannot be recovered. Some matches may be missing. Read the task output with bash_status({ taskId: "bash-gap", stderrOffset: 16 }).', + ], + [ + 1, + 1, + 'Some watch notices were lost: a stalled connection dropped 1 of them, and they cannot be recovered. At least 1 match is missing. Read the task output with bash_status({ taskId: "bash-gap", stderrOffset: 16 }).', + ], + [ + 2, + 3, + 'Some watch notices were lost: a stalled connection dropped 2 of them, and they cannot be recovered. At least 3 matches are missing. Read the task output with bash_status({ taskId: "bash-gap", stderrOffset: 16 }).', + ], +] as const)("retry gap-only reminder pins wording (%i notices, %i missing)", (notices, missing, expected) => { + const reminder = formatPatternMatchReminder([ + { + task_id: "bash-gap", + session_id: "s1", + watch_id: "watch-gap", + match_text: "watch delivery gap", + match_offset: 9, + context: "", + once: false, + batch: { + lines: [], + dropped_lines: 0, + skipped_long_lines: 0, + first_dropped_offset: null, + first_dropped_stream: null, + retry_gap: { + reason: "retry_overflow", + dropped_notifications: notices, + known_missing_occurrences: missing, + first_offset: 16, + last_offset: 24, + stream: "stderr", + }, + }, + }, + ]); + expect(reminder).toContain("[BG BASH NOTIFY]"); + expect(reminder).toContain(expected); +}); + test("each_line reminder keeps bounded rows and drop recovery hint", () => { const reminder = formatPatternMatchReminder([ { @@ -3015,3 +3067,95 @@ for (const route of ["in-turn", "wake"] as const) { }); }); } + +for (const route of ["in-turn", "wake"] as const) { + test(`admitted retry gap-only snapshots wake once per new version after a failed ack (${route})`, async () => { + const first = { + task_id: "task-batch", + session_id: "s1", + watch_id: "watch-batch", + match_offset: 0, + match_text: "matching lines", + context: "batch", + once: false, + reason: "pattern_match", + batch: { + pending_id: "pending-a", + version: 0, + snapshot: true, + lines: [], + retry_gap: { + reason: "retry_overflow" as const, + dropped_notifications: 2, + known_missing_occurrences: 1, + first_offset: 10, + }, + dropped_lines: 0, + skipped_long_lines: 0, + first_dropped_offset: 20, + first_dropped_stream: "stdout", + }, + }; + let current = first; + let failAck = true; + const acknowledgements: Record[] = []; + const { ctx } = harness((command, params) => { + if (command === "bash_drain_completions") + return { success: true, bg_completions: [], pending_matches: [current] }; + if (command === "bash_ack_completions") { + acknowledgements.push(params); + if (failAck) { + failAck = false; + return { success: false, message: "ack response lost" }; + } + return { success: true, acked_matches: params.matches ?? [] }; + } + return { success: true }; + }); + const sent: string[] = []; + const client = makeClient( + mock(async (input: unknown) => { + sent.push((input as { body: { parts: { text: string }[] } }).body.parts[0].text); + }), + ); + const context = { ctx, directory: "/tmp/project", sessionID: "s1", client }; + const deliver = async () => { + if (route === "in-turn") { + const output = { output: "tool" }; + await appendInTurnBgCompletions(context, output); + return output.output; + } + const wanted = sent.length + 1; + await handleSubcBgEventsNudge(context); + await waitUntil(() => sent.length === wanted && acknowledgements.length >= wanted); + return sent.at(-1)!; + }; + const old = await deliver(); + expect(old).toContain("a stalled connection dropped 2 of them, and they cannot be recovered."); + expect(old).toContain("At least 1 match is missing."); + current = { + ...first, + match_offset: 1, + batch: { + ...first.batch, + version: 1, + lines: [], + retry_gap: { + ...first.batch.retry_gap, + dropped_notifications: 3, + known_missing_occurrences: 2, + }, + dropped_lines: 0, + skipped_long_lines: 0, + }, + }; + const next = await deliver(); + expect(next).toContain("a stalled connection dropped 1 of them, and they cannot be recovered."); + expect(next).not.toContain("dropped 3 of them"); + expect(next).toContain("At least 1 match is missing."); + expect(acknowledgements.at(-1)).toEqual({ + session_id: "s1", + matches: [{ task_id: "task-batch", watch_id: "watch-batch", match_offset: 1 }], + }); + }); +} diff --git a/packages/pi-plugin/src/__tests__/bg-notifications.test.ts b/packages/pi-plugin/src/__tests__/bg-notifications.test.ts index c394b25ff..7625b316a 100644 --- a/packages/pi-plugin/src/__tests__/bg-notifications.test.ts +++ b/packages/pi-plugin/src/__tests__/bg-notifications.test.ts @@ -1440,6 +1440,58 @@ function sleep(ms: number): Promise { return new Promise((resolve) => setTimeout(resolve, ms)); } +test.each([ + [ + 2, + 1, + 'Some watch notices were lost: a stalled connection dropped 2 of them, and they cannot be recovered. At least 1 match is missing. Read the task output with bash_status({ task_id: "bash-gap", stderr_offset: 16 }).', + ], + [ + 1, + 0, + 'Some watch notices were lost: a stalled connection dropped 1 of them, and they cannot be recovered. Some matches may be missing. Read the task output with bash_status({ task_id: "bash-gap", stderr_offset: 16 }).', + ], + [ + 1, + 1, + 'Some watch notices were lost: a stalled connection dropped 1 of them, and they cannot be recovered. At least 1 match is missing. Read the task output with bash_status({ task_id: "bash-gap", stderr_offset: 16 }).', + ], + [ + 2, + 3, + 'Some watch notices were lost: a stalled connection dropped 2 of them, and they cannot be recovered. At least 3 matches are missing. Read the task output with bash_status({ task_id: "bash-gap", stderr_offset: 16 }).', + ], +] as const)("retry gap-only reminder pins wording (%i notices, %i missing)", (notices, missing, expected) => { + const reminder = formatPatternMatchReminder([ + { + task_id: "bash-gap", + session_id: "s1", + watch_id: "watch-gap", + match_text: "watch delivery gap", + match_offset: 9, + context: "", + once: false, + batch: { + lines: [], + dropped_lines: 0, + skipped_long_lines: 0, + first_dropped_offset: null, + first_dropped_stream: null, + retry_gap: { + reason: "retry_overflow", + dropped_notifications: notices, + known_missing_occurrences: missing, + first_offset: 16, + last_offset: 24, + stream: "stderr", + }, + }, + }, + ]); + expect(reminder).toContain("[BG BASH NOTIFY]"); + expect(reminder).toContain(expected); +}); + test("each_line reminder keeps bounded rows and drop recovery hint", () => { const reminder = formatPatternMatchReminder([ { @@ -1762,3 +1814,101 @@ for (const route of ["in-turn", "wake"] as const) { }); }); } + +for (const route of ["in-turn", "wake"] as const) { + test(`admitted retry gap-only snapshots wake once per new version after a failed ack (${route})`, async () => { + const first = { + task_id: "task-batch", + session_id: "s1", + watch_id: "watch-batch", + match_offset: 0, + match_text: "matching lines", + context: "batch", + once: false, + reason: "pattern_match", + batch: { + pending_id: "pending-a", + version: 0, + snapshot: true, + lines: [], + retry_gap: { + reason: "retry_overflow" as const, + dropped_notifications: 2, + known_missing_occurrences: 1, + first_offset: 10, + }, + dropped_lines: 0, + skipped_long_lines: 0, + first_dropped_offset: 20, + first_dropped_stream: "stdout", + }, + }; + let current = first; + let failAck = true; + const acknowledgements: Record[] = []; + const { ctx } = harness((command, params) => { + if (command === "bash_drain_completions") + return { success: true, bg_completions: [], pending_matches: [current] }; + if (command === "bash_ack_completions") { + acknowledgements.push(params); + if (failAck) { + failAck = false; + return { success: false, message: "ack response lost" }; + } + return { success: true, acked_matches: params.matches ?? [] }; + } + return { success: true }; + }); + const sent: string[] = []; + const runtime = { + sendUserMessage: (text: string) => { + sent.push(text); + }, + }; + const context = { ctx, directory: "/tmp/project", sessionID: "s1", runtime }; + const deliver = async () => { + if (route === "in-turn") { + const content = await appendToolResultBgCompletions(context, [ + { type: "text", text: "tool" }, + ]); + return ( + content + ?.filter((item) => item.type === "text") + .map((item) => item.text) + .join("\n") ?? "" + ); + } + const wanted = sent.length + 1; + await handleSubcBgEventsNudge(context); + await waitUntil(() => sent.length === wanted && acknowledgements.length >= wanted); + return sent.at(-1)!; + }; + const old = await deliver(); + expect(old).toContain("a stalled connection dropped 2 of them, and they cannot be recovered."); + expect(old).toContain("At least 1 match is missing."); + current = { + ...first, + match_offset: 1, + batch: { + ...first.batch, + version: 1, + lines: [], + retry_gap: { + ...first.batch.retry_gap, + dropped_notifications: 3, + known_missing_occurrences: 2, + }, + dropped_lines: 0, + skipped_long_lines: 0, + }, + }; + const next = await deliver(); + expect(next).toContain("a stalled connection dropped 1 of them, and they cannot be recovered."); + expect(next).not.toContain("dropped 3 of them"); + expect(next).toContain("At least 1 match is missing."); + expect(acknowledgements.at(-1)).toEqual({ + session_id: "s1", + matches: [{ task_id: "task-batch", watch_id: "watch-batch", match_offset: 1 }], + }); + }); +}