From f9c1cd6bb3724c8333ca9ea0e08d1c8a30d0b01e Mon Sep 17 00:00:00 2001 From: ualtinok <94532+ualtinok@users.noreply.github.com> Date: Wed, 28 Oct 2026 08:09:02 +0100 Subject: [PATCH 1/9] mason: retain retry-overflow watch gaps and expose delivery health Co-authored-by: alfonso-aft <289616620+alfonso-aft@users.noreply.github.com> --- .../release-notes/next/watch-gap-notice.md | 5 + crates/aft/src/bash_background/registry.rs | 371 ++++++++++++++++++ crates/aft/src/bash_background/watch_lines.rs | 19 + crates/aft/src/db/bash_watches.rs | 104 ++++- crates/aft/src/db/mod.rs | 9 +- crates/aft/src/subc/health.rs | 14 +- crates/aft/src/subc/mod.rs | 6 +- crates/aft/src/subc/push.rs | 74 +++- crates/aft/src/subc/watch_delivery.rs | 244 ++++++++++++ crates/aft/src/subc_format.rs | 21 + .../src/__tests__/bash-watch-progress.test.ts | 23 ++ packages/aft-bridge/src/bash-watch-lines.ts | 10 + .../aft-bridge/src/bash-watch-progress.ts | 30 +- packages/aft-bridge/src/bridge.ts | 8 + .../src/__tests__/bg-notifications.test.ts | 33 ++ .../src/__tests__/bg-notifications.test.ts | 33 ++ 16 files changed, 980 insertions(+), 24 deletions(-) create mode 100644 .alfonso/release-notes/next/watch-gap-notice.md create mode 100644 crates/aft/src/subc/watch_delivery.rs 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 0000000000..8ca1ee2cb0 --- /dev/null +++ b/.alfonso/release-notes/next/watch-gap-notice.md @@ -0,0 +1,5 @@ +## Watch delivery gaps are no longer silent + +When a stalled connection overflows the reliable-notification retry buffer, AFT retains a watch delivery-gap warning if it cannot prove that the evicted match notice is recoverable. The warning survives restart, uses the watch's exact-version acknowledgement, and points to retained task output. A warning alone does not consume a once-watch. + +Dispatch health exposes transport-copy evictions separately from outstanding watch gaps and known missing occurrences, and remains degraded until outstanding gaps are acknowledged. Recoverable watch snapshots and task completions are not counted as missing matches. diff --git a/crates/aft/src/bash_background/registry.rs b/crates/aft/src/bash_background/registry.rs index 292ab736e8..a9bf6b71c6 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) { @@ -4138,6 +4141,193 @@ 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())?; + if crate::db::bash_watches::delivery_ack_version( + &conn, + harness, + &frame.session_id, + &frame.task_id, + &frame.watch_id, + ) + .map_err(|e| e.to_string())? + .is_some_and(|v| v >= 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 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 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() + && row.pending_match + && candidate.first_version == candidate.last_version + && (row.match_offset == Some(candidate.first_version as i64) + || previous.as_ref().is_some_and(|b| { + b.lines + .iter() + .any(|line| line.offset == candidate.first_version) + })); + 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.evicted_through >= candidate.last_version) + { + return Ok(()); + } + // A superseded scalar has one identifiable occurrence. Overlapping + // snapshots and uncertain persistence never imply a lost-line count. + let known = u64::from( + found.is_some() + && frame.batch.is_none() + && 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: "stdout".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.first_offset.is_none() { + gap.first_offset = candidate.first_offset; + gap.stream = candidate.stream.clone(); + } + gap.last_offset = candidate.last_offset; + gap.evicted_through = 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())?); + crate::db::bash_watches::upsert_bash_pattern_watch(&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, + ); + notice.batch = Some(batch); + sender(PushFrame::BashPatternMatch(notice)); + } + Ok(()) + } + fn persist_watch_match( &self, session_id: &str, @@ -4166,6 +4356,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() @@ -4342,6 +4552,15 @@ 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, + ) + .map_err(|error| error.to_string())?; if removed { crate::db::bash_watches::delete_bash_pattern_watch( &conn, &harness, session_id, task_id, watch_id, @@ -4349,6 +4568,14 @@ impl BgTaskRegistry { .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; @@ -27468,6 +27695,150 @@ 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, + }; + 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("The dropped notices cannot be recovered.")); + 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)); + } + + #[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()); + } + 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 472d1583e8..4b23a9dbc4 100644 --- a/crates/aft/src/bash_background/watch_lines.rs +++ b/crates/aft/src/bash_background/watch_lines.rs @@ -60,8 +60,24 @@ 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, + pub first_offset: Option, + pub last_offset: Option, + pub stream: Option, + #[serde(default)] + pub evicted_through: 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 +101,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::(); diff --git a/crates/aft/src/db/bash_watches.rs b/crates/aft/src/db/bash_watches.rs index fd0db51bb5..3a639972f4 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( @@ -273,3 +278,94 @@ 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(); + +pub(crate) fn refresh_watch_gap_health(conn: &Connection) { + let key = conn + .path() + .filter(|path| !path.is_empty()) + .map(str::to_owned) + .unwrap_or_else(|| format!("memory:{:p}", 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")?; + 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 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| r.get::<_, i64>(0).map(|v| v.max(0) as u64)).optional() +} + +pub(crate) fn acknowledge_delivery( + conn: &Connection, + harness: &str, + session: &str, + task: &str, + watch: &str, + version: u64, +) -> rusqlite::Result<()> { + conn.execute("INSERT INTO bash_watch_delivery_acks(harness, session_id, task_id, watch_id, version, acknowledged_at) VALUES(?1,?2,?3,?4,?5,unixepoch()) ON CONFLICT(harness, session_id, task_id, watch_id) DO UPDATE SET version=MAX(version,excluded.version), acknowledged_at=excluded.acknowledged_at", params![harness, session, task, watch, version as i64])?; + // Retry queues are process-local. A month-old acknowledgement cannot still + // have a transport copy waiting in the daemon that wrote it. + conn.execute( + "DELETE FROM bash_watch_delivery_acks WHERE acknowledged_at < unixepoch() - 2592000", + [], + )?; + Ok(()) +} diff --git a/crates/aft/src/db/mod.rs b/crates/aft/src/db/mod.rs index 4ff8e40dd9..f5cc89caa2 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. @@ -1024,6 +1025,11 @@ 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 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, + PRIMARY KEY (harness, session_id, task_id, watch_id) + )"), _ => Ok(()), } } @@ -1131,6 +1137,7 @@ mod tests { "schema_version", "bash_tasks", "bash_pattern_watches", + "bash_watch_delivery_acks", "compression_events", "backups", "harness_state", diff --git a/crates/aft/src/subc/health.rs b/crates/aft/src/subc/health.rs index 395b399f4a..895ab54582 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() diff --git a/crates/aft/src/subc/mod.rs b/crates/aft/src/subc/mod.rs index 286e23950e..1ccb8f21e3 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. @@ -4030,7 +4031,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() @@ -5115,6 +5117,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 @@ -5484,6 +5487,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 45640bed44..630a1a2d6b 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 0000000000..2cc152135e --- /dev/null +++ b/crates/aft/src/subc/watch_delivery.rs @@ -0,0 +1,244 @@ +//! 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); + +#[derive(Clone)] +pub(crate) struct Candidate { + 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, +} + +impl Candidate { + fn new(mut frame: BashPatternMatchFrame) -> Self { + let first_offset = frame + .batch + .as_ref() + .and_then(|b| b.lines.first()) + .map(|l| l.offset) + .or(Some(frame.match_offset)); + let last_offset = frame + .batch + .as_ref() + .and_then(|b| b.lines.last()) + .map(|l| l.end_offset) + .or(first_offset); + let stream = frame + .batch + .as_ref() + .and_then(|b| b.lines.first()) + .map(|l| l.stream.clone()); + let version = frame.match_offset; + frame.match_text.clear(); + frame.context.clear(); + if let Some(batch) = &mut frame.batch { + batch.lines.clear(); + } + Self { + frame, + notifications: 1, + first_version: version, + last_version: version, + first_offset, + last_offset, + stream, + same_generation: true, + } + } + + fn merge(&mut self, next: Self) { + 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 (gaps, missing, uncertain) = if self.executor.lock().unwrap().upgrade().is_some() { + crate::db::bash_watches::watch_gap_health() + } else { + (0, 0, false) + }; + let pending = PENDING_CHECKS.load(Ordering::Relaxed); + 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, + }) + } + + 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())) + .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(); + }); + } + }); + } +} + +#[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 17ebd1cab2..7dd80e94cf 100644 --- a/crates/aft/src/subc_format.rs +++ b/crates/aft/src/subc_format.rs @@ -5502,6 +5502,27 @@ 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 notices = if gap.dropped_notifications == 1 { + "notice" + } else { + "notices" + }; + let known = if gap.known_missing_occurrences > 0 { + format!( + " Known missing occurrences: {}.", + gap.known_missing_occurrences + ) + } else { + String::new() + }; + let cursor = if gap.stream.as_deref() == Some("stderr") { + "stderr_offset" + } else { + "output_offset" + }; + text.push_str(&format!("Watch delivery gap: a stalled connection dropped {} watch {notices}; matches may be missing. The dropped notices cannot be recovered.{known} Inspect retained 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() } 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 36ed47996a..e7c1141394 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,29 @@ 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("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 307237e624..06fbbdf985 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,15 @@ 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`; + lines.push( + `Watch delivery gap: a stalled connection dropped ${gap.dropped_notifications} watch ${gap.dropped_notifications === 1 ? "notice" : "notices"}; matches may be missing. The dropped notices cannot be recovered.${gap.known_missing_occurrences > 0 ? ` Known missing occurrences: ${gap.known_missing_occurrences}.` : ""} Inspect retained 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 3c5f69b110..ba2e3e8cc6 100644 --- a/packages/aft-bridge/src/bash-watch-progress.ts +++ b/packages/aft-bridge/src/bash-watch-progress.ts @@ -14,6 +14,8 @@ export class WatchBatchProgress { private ends: Record = { stdout: 0, stderr: 0, pty: 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, }; } @@ -61,6 +77,8 @@ export class WatchBatchProgress { this.ends = { stdout: 0, stderr: 0, pty: 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 564a85350e..21d7e85cd0 100644 --- a/packages/aft-bridge/src/bridge.ts +++ b/packages/aft-bridge/src/bridge.ts @@ -501,6 +501,14 @@ 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; + }; 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 0d16a67463..298936de18 100644 --- a/packages/opencode-plugin/src/__tests__/bg-notifications.test.ts +++ b/packages/opencode-plugin/src/__tests__/bg-notifications.test.ts @@ -2706,6 +2706,39 @@ function preAdmissionRejection(message: string) { }; } +test("retry gap-only reminder reports unrecoverable notices and an output offset", () => { + 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: 2, + known_missing_occurrences: 1, + first_offset: 16, + last_offset: 24, + stream: "stderr", + }, + }, + }, + ]); + expect(reminder).toContain("[BG BASH NOTIFY]"); + expect(reminder).toContain( + 'Watch delivery gap: a stalled connection dropped 2 watch notices; matches may be missing. The dropped notices cannot be recovered. Known missing occurrences: 1. Inspect retained task output with bash_status({ taskId: "bash-gap", stderrOffset: 16 }).', + ); +}); + test("each_line reminder keeps bounded rows and drop recovery hint", () => { const reminder = formatPatternMatchReminder([ { diff --git a/packages/pi-plugin/src/__tests__/bg-notifications.test.ts b/packages/pi-plugin/src/__tests__/bg-notifications.test.ts index 5b1c4cbbf4..c539e20990 100644 --- a/packages/pi-plugin/src/__tests__/bg-notifications.test.ts +++ b/packages/pi-plugin/src/__tests__/bg-notifications.test.ts @@ -1439,6 +1439,39 @@ function sleep(ms: number): Promise { return new Promise((resolve) => setTimeout(resolve, ms)); } +test("retry gap-only reminder reports unrecoverable notices and an output offset", () => { + 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: 2, + known_missing_occurrences: 1, + first_offset: 16, + last_offset: 24, + stream: "stderr", + }, + }, + }, + ]); + expect(reminder).toContain("[BG BASH NOTIFY]"); + expect(reminder).toContain( + 'Watch delivery gap: a stalled connection dropped 2 watch notices; matches may be missing. The dropped notices cannot be recovered. Known missing occurrences: 1. Inspect retained task output with bash_status({ task_id: "bash-gap", stderr_offset: 16 }).', + ); +}); + test("each_line reminder keeps bounded rows and drop recovery hint", () => { const reminder = formatPatternMatchReminder([ { From 97db4fbf2632fee02ea39161406992e6188533f3 Mon Sep 17 00:00:00 2001 From: ualtinok <94532+ualtinok@users.noreply.github.com> Date: Wed, 28 Oct 2026 08:23:37 +0100 Subject: [PATCH 2/9] mason: fence gap acknowledgements and cover gap-only host wakes Co-authored-by: alfonso-aft <289616620+alfonso-aft@users.noreply.github.com> --- crates/aft/src/bash_background/registry.rs | 199 +++++++++++++++--- crates/aft/src/bash_background/watch_lines.rs | 15 +- crates/aft/src/db/bash_watches.rs | 22 +- crates/aft/src/db/mod.rs | 4 +- crates/aft/src/subc/health.rs | 74 +++++++ crates/aft/src/subc/watch_delivery.rs | 112 ++++++---- packages/aft-bridge/src/bridge.ts | 4 + .../src/__tests__/bg-notifications.test.ts | 92 ++++++++ .../src/__tests__/bg-notifications.test.ts | 98 +++++++++ 9 files changed, 546 insertions(+), 74 deletions(-) diff --git a/crates/aft/src/bash_background/registry.rs b/crates/aft/src/bash_background/registry.rs index a9bf6b71c6..2f3585a844 100644 --- a/crates/aft/src/bash_background/registry.rs +++ b/crates/aft/src/bash_background/registry.rs @@ -3905,6 +3905,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() @@ -4156,15 +4163,23 @@ impl BgTaskRegistry { .clone() .ok_or_else(|| "watch persistence unavailable".to_string())?; let conn = pool.lock().map_err(|e| e.to_string())?; - if crate::db::bash_watches::delivery_ack_version( + 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())? - .is_some_and(|v| v >= candidate.last_version) + .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(()); } @@ -4203,10 +4218,11 @@ impl BgTaskRegistry { .match_batch .as_deref() .and_then(|b| serde_json::from_str::(b).ok()); - if previous - .as_ref() - .and_then(|b| b.acknowledged_version) - .is_some_and(|v| v >= candidate.last_version) + if frame.batch.is_some() + && previous + .as_ref() + .and_then(|b| b.acknowledged_version) + .is_some_and(|v| v >= candidate.last_version) { return Ok(()); } @@ -4220,13 +4236,22 @@ impl BgTaskRegistry { }) }); 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_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| line.offset == candidate.first_version) + 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(()); @@ -4236,11 +4261,10 @@ impl BgTaskRegistry { } else { LineBatch::default() }; - if batch - .retry_gap - .as_ref() - .is_some_and(|g| g.evicted_through >= candidate.last_version) - { + 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 @@ -4248,6 +4272,14 @@ impl BgTaskRegistry { let known = u64::from( found.is_some() && 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 @@ -4258,7 +4290,7 @@ impl BgTaskRegistry { if let Some(text) = &row.match_text { let offset = row.match_offset.unwrap_or_default().max(0) as u64; batch.lines.push(WatchLine { - stream: "stdout".to_string(), + stream: "output".to_string(), offset, end_offset: offset.saturating_add(1), text: text.clone(), @@ -4275,12 +4307,27 @@ impl BgTaskRegistry { .dropped_notifications .saturating_add(candidate.notifications); gap.known_missing_occurrences = gap.known_missing_occurrences.saturating_add(known); - if gap.first_offset.is_none() { + 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.last_offset = candidate.last_offset; - gap.evicted_through = candidate.last_version; + 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) @@ -4320,7 +4367,7 @@ impl BgTaskRegistry { row.match_text.unwrap_or_default(), version, row.match_context.unwrap_or_default(), - row.once && !row.scanning, + row.once && !row.scanning && batch.has_matches(), ); notice.batch = Some(batch); sender(PushFrame::BashPatternMatch(notice)); @@ -4559,6 +4606,13 @@ impl BgTaskRegistry { 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 { @@ -4619,19 +4673,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, @@ -12750,17 +12809,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, @@ -27746,6 +27807,13 @@ mod tests { assert!(pending .context .contains("The dropped notices 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() @@ -27780,6 +27848,31 @@ mod tests { .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] @@ -27839,6 +27932,46 @@ mod tests { 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")); + } + 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 4b23a9dbc4..fc6d016711 100644 --- a/crates/aft/src/bash_background/watch_lines.rs +++ b/crates/aft/src/bash_background/watch_lines.rs @@ -65,11 +65,18 @@ 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)] @@ -142,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 3a639972f4..560e572346 100644 --- a/crates/aft/src/db/bash_watches.rs +++ b/crates/aft/src/db/bash_watches.rs @@ -348,8 +348,8 @@ pub(crate) fn delivery_ack_version( session: &str, task: &str, watch: &str, -) -> rusqlite::Result> { - conn.query_row("SELECT version 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| r.get::<_, i64>(0).map(|v| v.max(0) as u64)).optional() +) -> 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( @@ -359,8 +359,13 @@ pub(crate) fn acknowledge_delivery( task: &str, watch: &str, version: u64, + gap: Option<&crate::bash_background::watch_lines::RetryGap>, ) -> rusqlite::Result<()> { - conn.execute("INSERT INTO bash_watch_delivery_acks(harness, session_id, task_id, watch_id, version, acknowledged_at) VALUES(?1,?2,?3,?4,?5,unixepoch()) ON CONFLICT(harness, session_id, task_id, watch_id) DO UPDATE SET version=MAX(version,excluded.version), acknowledged_at=excluded.acknowledged_at", params![harness, session, task, watch, version as i64])?; + let epoch = gap + .map(|gap| gap.eviction_epoch.as_str()) + .unwrap_or_default(); + 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 excluded.eviction_epoch != '' THEN excluded.evicted_through ELSE evicted_through END", params![harness, session, task, watch, version as i64, epoch, through as i64])?; // Retry queues are process-local. A month-old acknowledgement cannot still // have a transport copy waiting in the daemon that wrote it. conn.execute( @@ -369,3 +374,14 @@ pub(crate) fn acknowledge_delivery( )?; Ok(()) } + +#[cfg(test)] +pub(crate) fn watch_gap_health_for_path(path: &std::path::Path) -> GapCensus { + WATCH_GAP_CENSUS + .get_or_init(Default::default) + .lock() + .unwrap() + .get(&path.to_string_lossy().into_owned()) + .copied() + .unwrap_or_default() +} diff --git a/crates/aft/src/db/mod.rs b/crates/aft/src/db/mod.rs index f5cc89caa2..8953a41dd9 100644 --- a/crates/aft/src/db/mod.rs +++ b/crates/aft/src/db/mod.rs @@ -996,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='bash_watch_delivery_acks'", [], |row| row.get::<_, u32>(0)).map(|count| count == 1), _ => Ok(false), } } @@ -1025,9 +1026,10 @@ 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 bash_watch_delivery_acks ( + 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) )"), _ => Ok(()), diff --git a/crates/aft/src/subc/health.rs b/crates/aft/src/subc/health.rs index 895ab54582..09c1cbb7df 100644 --- a/crates/aft/src/subc/health.rs +++ b/crates/aft/src/subc/health.rs @@ -9569,3 +9569,77 @@ mod observability_retention_tests { assert!(events.keys().all(|key| key.session != "inactive")); } } + +#[cfg(test)] +mod watch_gap_tests { + use super::*; + + #[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 mut 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_bash_pattern_watch(&conn, &row).unwrap(); + let (gaps, missing, uncertain) = watch_gap_health_for_path(&path); + 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(); + row.pending_match = false; + upsert_bash_pattern_watch(&conn, &row).unwrap(); + let (gaps, missing, uncertain) = watch_gap_health_for_path(&path); + 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/watch_delivery.rs b/crates/aft/src/subc/watch_delivery.rs index 2cc152135e..ddc9a490c7 100644 --- a/crates/aft/src/subc/watch_delivery.rs +++ b/crates/aft/src/subc/watch_delivery.rs @@ -4,9 +4,14 @@ 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(); #[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, @@ -15,34 +20,52 @@ pub(crate) struct Candidate { pub last_offset: Option, pub stream: Option, pub same_generation: bool, + pub same_occurrence: bool, } impl Candidate { fn new(mut frame: BashPatternMatchFrame) -> Self { - let first_offset = frame - .batch - .as_ref() - .and_then(|b| b.lines.first()) - .map(|l| l.offset) - .or(Some(frame.match_offset)); - let last_offset = frame - .batch - .as_ref() - .and_then(|b| b.lines.last()) - .map(|l| l.end_offset) - .or(first_offset); - let stream = frame - .batch - .as_ref() - .and_then(|b| b.lines.first()) - .map(|l| l.stream.clone()); + 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.clear(); - frame.context.clear(); + frame.match_text = String::new(); + frame.context = String::new(); if let Some(batch) = &mut frame.batch { - batch.lines.clear(); + batch.lines = Vec::new(); } Self { + eviction_epoch: EVICTION_EPOCH + .get_or_init(|| uuid::Uuid::new_v4().to_string()) + .clone(), + last_eviction: NEXT_EVICTION.fetch_add(1, Ordering::Relaxed), + fingerprint, frame, notifications: 1, first_version: version, @@ -51,10 +74,15 @@ impl Candidate { 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); @@ -129,21 +157,7 @@ impl WatchDelivery { (0, 0, false) }; let pending = PENDING_CHECKS.load(Ordering::Relaxed); - 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, - }) + health_from_census(gaps, missing, uncertain, pending) } pub(super) fn kick(self: &Arc) { @@ -175,7 +189,10 @@ impl WatchDelivery { .lock() .unwrap() .entry(key) - .and_modify(|new| new.merge(candidate.clone())) + .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); @@ -193,6 +210,29 @@ impl WatchDelivery { } } +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(); diff --git a/packages/aft-bridge/src/bridge.ts b/packages/aft-bridge/src/bridge.ts index 21d7e85cd0..973feda58f 100644 --- a/packages/aft-bridge/src/bridge.ts +++ b/packages/aft-bridge/src/bridge.ts @@ -508,7 +508,11 @@ export interface BashWatchLineBatch { 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 298936de18..76a1482ef7 100644 --- a/packages/opencode-plugin/src/__tests__/bg-notifications.test.ts +++ b/packages/opencode-plugin/src/__tests__/bg-notifications.test.ts @@ -2941,3 +2941,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 watch notices; matches may be missing."); + expect(old).toContain("Known missing occurrences: 1."); + 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 watch notice; matches may be missing."); + expect(next).not.toContain("dropped 3 watch notices"); + expect(next).toContain("Known missing occurrences: 1."); + 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 c539e20990..8633a802a1 100644 --- a/packages/pi-plugin/src/__tests__/bg-notifications.test.ts +++ b/packages/pi-plugin/src/__tests__/bg-notifications.test.ts @@ -1689,3 +1689,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 watch notices; matches may be missing."); + expect(old).toContain("Known missing occurrences: 1."); + 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 watch notice; matches may be missing."); + expect(next).not.toContain("dropped 3 watch notices"); + expect(next).toContain("Known missing occurrences: 1."); + expect(acknowledgements.at(-1)).toEqual({ + session_id: "s1", + matches: [{ task_id: "task-batch", watch_id: "watch-batch", match_offset: 1 }], + }); + }); +} From f6c5e88a8787c0f64bd736625f501d1a02b7e6de Mon Sep 17 00:00:00 2001 From: ualtinok <94532+ualtinok@users.noreply.github.com> Date: Wed, 28 Oct 2026 08:24:57 +0100 Subject: [PATCH 3/9] mason: identify eviction epochs without a new dependency Co-authored-by: alfonso-aft <289616620+alfonso-aft@users.noreply.github.com> --- crates/aft/src/subc/watch_delivery.rs | 11 ++++++++++- 1 file changed, 10 insertions(+), 1 deletion(-) diff --git a/crates/aft/src/subc/watch_delivery.rs b/crates/aft/src/subc/watch_delivery.rs index ddc9a490c7..8ffe789e3a 100644 --- a/crates/aft/src/subc/watch_delivery.rs +++ b/crates/aft/src/subc/watch_delivery.rs @@ -62,7 +62,16 @@ impl Candidate { } Self { eviction_epoch: EVICTION_EPOCH - .get_or_init(|| uuid::Uuid::new_v4().to_string()) + .get_or_init(|| { + format!( + "{}:{}", + std::process::id(), + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_nanos() + ) + }) .clone(), last_eviction: NEXT_EVICTION.fetch_add(1, Ordering::Relaxed), fingerprint, From 4624b775f28439cadec2a062ac38819a99900b1d Mon Sep 17 00:00:00 2001 From: ualtinok <94532+ualtinok@users.noreply.github.com> Date: Wed, 28 Oct 2026 08:47:12 +0100 Subject: [PATCH 4/9] mason: retain session gaps after task removal and fence old daemon acks Co-authored-by: alfonso-aft <289616620+alfonso-aft@users.noreply.github.com> --- crates/aft/src/bash_background/registry.rs | 132 ++++++++++++++++-- crates/aft/src/db/bash_watches.rs | 119 ++++++++++++++-- crates/aft/src/db/mod.rs | 7 +- crates/aft/src/subc/health.rs | 8 +- crates/aft/src/subc/watch_delivery.rs | 35 +++-- .../src/__tests__/bash-watch-progress.test.ts | 13 ++ .../aft-bridge/src/bash-watch-progress.ts | 8 +- 7 files changed, 275 insertions(+), 47 deletions(-) diff --git a/crates/aft/src/bash_background/registry.rs b/crates/aft/src/bash_background/registry.rs index 2f3585a844..8b6a5a684b 100644 --- a/crates/aft/src/bash_background/registry.rs +++ b/crates/aft/src/bash_background/registry.rs @@ -3858,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}" @@ -4191,6 +4203,16 @@ impl BgTaskRegistry { &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() @@ -4270,7 +4292,7 @@ impl BgTaskRegistry { // A superseded scalar has one identifiable occurrence. Overlapping // snapshots and uncertain persistence never imply a lost-line count. let known = u64::from( - found.is_some() + registered && frame.batch.is_none() && candidate.same_occurrence && acknowledged @@ -4350,8 +4372,12 @@ impl BgTaskRegistry { &batch, )); row.match_batch = Some(serde_json::to_string(&batch).map_err(|e| e.to_string())?); - crate::db::bash_watches::upsert_bash_pattern_watch(&conn, &row) - .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 @@ -4586,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 @@ -4616,10 +4651,17 @@ impl BgTaskRegistry { ) .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 @@ -4642,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); @@ -10757,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 } @@ -27972,6 +28026,60 @@ mod tests { .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/db/bash_watches.rs b/crates/aft/src/db/bash_watches.rs index 560e572346..c43c111ee2 100644 --- a/crates/aft/src/db/bash_watches.rs +++ b/crates/aft/src/db/bash_watches.rs @@ -192,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), ) @@ -205,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), ) @@ -293,7 +293,7 @@ pub(crate) fn refresh_watch_gap_health(conn: &Connection) { .map(str::to_owned) .unwrap_or_else(|| format!("memory:{:p}", 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")?; + 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; @@ -363,14 +363,14 @@ pub(crate) fn acknowledge_delivery( ) -> rusqlite::Result<()> { let epoch = gap .map(|gap| gap.eviction_epoch.as_str()) - .unwrap_or_default(); + .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 excluded.eviction_epoch != '' THEN excluded.evicted_through ELSE evicted_through END", params![harness, session, task, watch, version as i64, epoch, through as i64])?; - // Retry queues are process-local. A month-old acknowledgement cannot still - // have a transport copy waiting in the daemon that wrote it. + 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", - [], + "DELETE FROM bash_watch_delivery_acks WHERE acknowledged_at < unixepoch() - 2592000 AND eviction_epoch != ?1", + [crate::subc::watch_delivery::eviction_epoch()], )?; Ok(()) } @@ -385,3 +385,100 @@ pub(crate) fn watch_gap_health_for_path(path: &std::path::Path) -> GapCensus { .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 8953a41dd9..e998d1e1ee 100644 --- a/crates/aft/src/db/mod.rs +++ b/crates/aft/src/db/mod.rs @@ -996,7 +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='bash_watch_delivery_acks'", [], |row| row.get::<_, u32>(0)).map(|count| count == 1), + 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), } } @@ -1031,6 +1031,10 @@ fn apply_migration_statements(conn: &Connection, version: u32) -> rusqlite::Resu 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(()), } @@ -1140,6 +1144,7 @@ mod tests { "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 09c1cbb7df..e714647437 100644 --- a/crates/aft/src/subc/health.rs +++ b/crates/aft/src/subc/health.rs @@ -9572,7 +9572,6 @@ mod observability_retention_tests { #[cfg(test)] mod watch_gap_tests { - use super::*; #[test] fn retry_overflow_health_keeps_evictions_separate_and_clears_after_ack() { @@ -9591,7 +9590,7 @@ mod watch_gap_tests { version: Some(5), ..Default::default() }; - let mut row = BashPatternWatchRow { + let row = BashPatternWatchRow { harness: "opencode".into(), session_id: "session".into(), task_id: "task".into(), @@ -9611,7 +9610,7 @@ mod watch_gap_tests { match_offset: Some(5), match_context: None, }; - upsert_bash_pattern_watch(&conn, &row).unwrap(); + upsert_delivery_gap(&conn, &row).unwrap(); let (gaps, missing, uncertain) = watch_gap_health_for_path(&path); assert_eq!((gaps, missing, uncertain), (1, 1, false)); let health = super::super::watch_delivery::health_from_census(gaps, missing, uncertain, 0); @@ -9630,8 +9629,7 @@ mod watch_gap_tests { batch.retry_gap.as_ref(), ) .unwrap(); - row.pending_match = false; - upsert_bash_pattern_watch(&conn, &row).unwrap(); + delete_delivery_gap(&conn, "opencode", "session", "task", "watch").unwrap(); let (gaps, missing, uncertain) = watch_gap_health_for_path(&path); let cleared = super::super::watch_delivery::health_from_census(gaps, missing, uncertain, 0); assert_eq!(cleared["status"], "ok"); diff --git a/crates/aft/src/subc/watch_delivery.rs b/crates/aft/src/subc/watch_delivery.rs index 8ffe789e3a..1b8b708bff 100644 --- a/crates/aft/src/subc/watch_delivery.rs +++ b/crates/aft/src/subc/watch_delivery.rs @@ -7,6 +7,19 @@ 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, @@ -61,18 +74,7 @@ impl Candidate { batch.lines = Vec::new(); } Self { - eviction_epoch: EVICTION_EPOCH - .get_or_init(|| { - format!( - "{}:{}", - std::process::id(), - std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .unwrap_or_default() - .as_nanos() - ) - }) - .clone(), + eviction_epoch: eviction_epoch().to_string(), last_eviction: NEXT_EVICTION.fetch_add(1, Ordering::Relaxed), fingerprint, frame, @@ -160,12 +162,17 @@ impl WatchDelivery { } pub(super) fn health(&self) -> serde_json::Value { - let (gaps, missing, uncertain) = if self.executor.lock().unwrap().upgrade().is_some() { + 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 = PENDING_CHECKS.load(Ordering::Relaxed); + let pending = if attached { + PENDING_CHECKS.load(Ordering::Relaxed) + } else { + 0 + }; health_from_census(gaps, missing, uncertain, pending) } 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 e7c1141394..057d24794a 100644 --- a/packages/aft-bridge/src/__tests__/bash-watch-progress.test.ts +++ b/packages/aft-bridge/src/__tests__/bash-watch-progress.test.ts @@ -61,6 +61,19 @@ test("retry gap-only batches are admitted once while newer gap versions remain v 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-progress.ts b/packages/aft-bridge/src/bash-watch-progress.ts index ba2e3e8cc6..ca5df986d7 100644 --- a/packages/aft-bridge/src/bash-watch-progress.ts +++ b/packages/aft-bridge/src/bash-watch-progress.ts @@ -1,17 +1,17 @@ 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; @@ -74,7 +74,7 @@ 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; From c4da174f49eccb7c3f3c4fb4eebc45b2d20f97c9 Mon Sep 17 00:00:00 2001 From: ualtinok <94532+ualtinok@users.noreply.github.com> Date: Wed, 28 Oct 2026 14:53:29 +0100 Subject: [PATCH 5/9] mason: pin approved watch-gap wording after the speech migrations Use the approved plain-language notice in Rust, OpenCode and Pi, including unknown counts and singular/plural missing-match sentences. Replace the prior wording goldens intentionally to pin that approved copy. Keep speech-authority migrations 22 and 23 unchanged, place watch-gap tables and tombstones at migration 24, and test a populated schema-23 upgrade. Earlier migration tests still verify their speech data while advancing to the current schema. Co-authored-by: alfonso-aft <289616620+alfonso-aft@users.noreply.github.com> --- crates/aft/src/bash_background/registry.rs | 5 +- crates/aft/src/db/mod.rs | 52 +++++++++++++++++-- crates/aft/src/subc_format.rs | 40 +++++++++----- packages/aft-bridge/src/bash-watch-lines.ts | 8 ++- .../src/__tests__/bg-notifications.test.ts | 41 +++++++++++---- .../src/__tests__/bg-notifications.test.ts | 41 +++++++++++---- 6 files changed, 144 insertions(+), 43 deletions(-) diff --git a/crates/aft/src/bash_background/registry.rs b/crates/aft/src/bash_background/registry.rs index 8b6a5a684b..2bbd40d7f0 100644 --- a/crates/aft/src/bash_background/registry.rs +++ b/crates/aft/src/bash_background/registry.rs @@ -27830,6 +27830,7 @@ mod tests { 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(); @@ -27858,9 +27859,7 @@ mod tests { assert_eq!(gap.dropped_notifications, 1); assert_eq!(gap.known_missing_occurrences, 1); assert_eq!(gap.first_offset, Some(10)); - assert!(pending - .context - .contains("The dropped notices cannot be recovered.")); + 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 diff --git a/crates/aft/src/db/mod.rs b/crates/aft/src/db/mod.rs index e998d1e1ee..7b0a2e0ec9 100644 --- a/crates/aft/src/db/mod.rs +++ b/crates/aft/src/db/mod.rs @@ -1082,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", @@ -1101,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] @@ -1118,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'", @@ -1136,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; diff --git a/crates/aft/src/subc_format.rs b/crates/aft/src/subc_format.rs index 7dd80e94cf..ed65523046 100644 --- a/crates/aft/src/subc_format.rs +++ b/crates/aft/src/subc_format.rs @@ -5503,25 +5503,17 @@ 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 notices = if gap.dropped_notifications == 1 { - "notice" - } else { - "notices" - }; - let known = if gap.known_missing_occurrences > 0 { - format!( - " Known missing occurrences: {}.", - gap.known_missing_occurrences - ) - } else { - String::new() + 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!("Watch delivery gap: a stalled connection dropped {} watch {notices}; matches may be missing. The dropped notices cannot be recovered.{known} Inspect retained 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.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() } @@ -5529,7 +5521,27 @@ pub fn format_watch_line_batch( #[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/packages/aft-bridge/src/bash-watch-lines.ts b/packages/aft-bridge/src/bash-watch-lines.ts index 06fbbdf985..076594e85d 100644 --- a/packages/aft-bridge/src/bash-watch-lines.ts +++ b/packages/aft-bridge/src/bash-watch-lines.ts @@ -88,8 +88,14 @@ export function formatWatchLineBatch( 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( - `Watch delivery gap: a stalled connection dropped ${gap.dropped_notifications} watch ${gap.dropped_notifications === 1 ? "notice" : "notices"}; matches may be missing. The dropped notices cannot be recovered.${gap.known_missing_occurrences > 0 ? ` Known missing occurrences: ${gap.known_missing_occurrences}.` : ""} Inspect retained task output with bash_status({ ${taskKey}: ${JSON.stringify(taskId)}, ${offsetKey}: ${gap.first_offset ?? 0} }).`, + `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/opencode-plugin/src/__tests__/bg-notifications.test.ts b/packages/opencode-plugin/src/__tests__/bg-notifications.test.ts index 76a1482ef7..ba13e8011e 100644 --- a/packages/opencode-plugin/src/__tests__/bg-notifications.test.ts +++ b/packages/opencode-plugin/src/__tests__/bg-notifications.test.ts @@ -2706,7 +2706,28 @@ function preAdmissionRejection(message: string) { }; } -test("retry gap-only reminder reports unrecoverable notices and an output offset", () => { +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", @@ -2724,8 +2745,8 @@ test("retry gap-only reminder reports unrecoverable notices and an output offset first_dropped_stream: null, retry_gap: { reason: "retry_overflow", - dropped_notifications: 2, - known_missing_occurrences: 1, + dropped_notifications: notices, + known_missing_occurrences: missing, first_offset: 16, last_offset: 24, stream: "stderr", @@ -2734,9 +2755,7 @@ test("retry gap-only reminder reports unrecoverable notices and an output offset }, ]); expect(reminder).toContain("[BG BASH NOTIFY]"); - expect(reminder).toContain( - 'Watch delivery gap: a stalled connection dropped 2 watch notices; matches may be missing. The dropped notices cannot be recovered. Known missing occurrences: 1. Inspect retained task output with bash_status({ taskId: "bash-gap", stderrOffset: 16 }).', - ); + expect(reminder).toContain(expected); }); test("each_line reminder keeps bounded rows and drop recovery hint", () => { @@ -3005,8 +3024,8 @@ for (const route of ["in-turn", "wake"] as const) { return sent.at(-1)!; }; const old = await deliver(); - expect(old).toContain("a stalled connection dropped 2 watch notices; matches may be missing."); - expect(old).toContain("Known missing occurrences: 1."); + 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, @@ -3024,9 +3043,9 @@ for (const route of ["in-turn", "wake"] as const) { }, }; const next = await deliver(); - expect(next).toContain("a stalled connection dropped 1 watch notice; matches may be missing."); - expect(next).not.toContain("dropped 3 watch notices"); - expect(next).toContain("Known missing occurrences: 1."); + 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 8633a802a1..b2f05ca608 100644 --- a/packages/pi-plugin/src/__tests__/bg-notifications.test.ts +++ b/packages/pi-plugin/src/__tests__/bg-notifications.test.ts @@ -1439,7 +1439,28 @@ function sleep(ms: number): Promise { return new Promise((resolve) => setTimeout(resolve, ms)); } -test("retry gap-only reminder reports unrecoverable notices and an output offset", () => { +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", @@ -1457,8 +1478,8 @@ test("retry gap-only reminder reports unrecoverable notices and an output offset first_dropped_stream: null, retry_gap: { reason: "retry_overflow", - dropped_notifications: 2, - known_missing_occurrences: 1, + dropped_notifications: notices, + known_missing_occurrences: missing, first_offset: 16, last_offset: 24, stream: "stderr", @@ -1467,9 +1488,7 @@ test("retry gap-only reminder reports unrecoverable notices and an output offset }, ]); expect(reminder).toContain("[BG BASH NOTIFY]"); - expect(reminder).toContain( - 'Watch delivery gap: a stalled connection dropped 2 watch notices; matches may be missing. The dropped notices cannot be recovered. Known missing occurrences: 1. Inspect retained task output with bash_status({ task_id: "bash-gap", stderr_offset: 16 }).', - ); + expect(reminder).toContain(expected); }); test("each_line reminder keeps bounded rows and drop recovery hint", () => { @@ -1759,8 +1778,8 @@ for (const route of ["in-turn", "wake"] as const) { return sent.at(-1)!; }; const old = await deliver(); - expect(old).toContain("a stalled connection dropped 2 watch notices; matches may be missing."); - expect(old).toContain("Known missing occurrences: 1."); + 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, @@ -1778,9 +1797,9 @@ for (const route of ["in-turn", "wake"] as const) { }, }; const next = await deliver(); - expect(next).toContain("a stalled connection dropped 1 watch notice; matches may be missing."); - expect(next).not.toContain("dropped 3 watch notices"); - expect(next).toContain("Known missing occurrences: 1."); + 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 }], From 45d1a9844aa209bec9fc86b745e66348ff1cf124 Mon Sep 17 00:00:00 2001 From: ualtinok <94532+ualtinok@users.noreply.github.com> Date: Wed, 28 Oct 2026 15:35:47 +0100 Subject: [PATCH 6/9] One-line release note and search descriptor for the watch-gap notice Co-authored-by: alfonso-aft <289616620+alfonso-aft@users.noreply.github.com> --- .alfonso/release-notes/next/watch-gap-notice.md | 6 +----- .../slice-descriptors/land-watch-gap-notice.json | 12 ++++++++++++ 2 files changed, 13 insertions(+), 5 deletions(-) create mode 100644 benchmarks/aft-search/slice-descriptors/land-watch-gap-notice.json diff --git a/.alfonso/release-notes/next/watch-gap-notice.md b/.alfonso/release-notes/next/watch-gap-notice.md index 8ca1ee2cb0..75a1b3016c 100644 --- a/.alfonso/release-notes/next/watch-gap-notice.md +++ b/.alfonso/release-notes/next/watch-gap-notice.md @@ -1,5 +1 @@ -## Watch delivery gaps are no longer silent - -When a stalled connection overflows the reliable-notification retry buffer, AFT retains a watch delivery-gap warning if it cannot prove that the evicted match notice is recoverable. The warning survives restart, uses the watch's exact-version acknowledgement, and points to retained task output. A warning alone does not consume a once-watch. - -Dispatch health exposes transport-copy evictions separately from outstanding watch gaps and known missing occurrences, and remains degraded until outstanding gaps are acknowledged. Recoverable watch snapshots and task completions are not counted as missing matches. +- 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 0000000000..7208dc6841 --- /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." +} From 20a8fffb78e983b1d4b86a80b066b3e2b9c96bb9 Mon Sep 17 00:00:00 2001 From: ualtinok <94532+ualtinok@users.noreply.github.com> Date: Wed, 28 Oct 2026 21:49:24 +0100 Subject: [PATCH 7/9] Watch-gap health test helper looks up SQLite's resolved path The gap census is keyed by SQLite's file name, which resolves symlinks. On macOS the test's temp folder is /var/..., stored as /private/var/..., so the helper found nothing. Canonicalize before the lookup. Co-authored-by: alfonso-aft <289616620+alfonso-aft@users.noreply.github.com> --- crates/aft/src/db/bash_watches.rs | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/crates/aft/src/db/bash_watches.rs b/crates/aft/src/db/bash_watches.rs index c43c111ee2..1268ec2660 100644 --- a/crates/aft/src/db/bash_watches.rs +++ b/crates/aft/src/db/bash_watches.rs @@ -377,6 +377,10 @@ pub(crate) fn acknowledge_delivery( #[cfg(test)] pub(crate) fn watch_gap_health_for_path(path: &std::path::Path) -> GapCensus { + // The census is keyed by SQLite's own file name, which resolves symlinks + // (macOS temp folders live under /var -> /private/var), so look up the + // same resolved spelling. + let path = std::fs::canonicalize(path).unwrap_or_else(|_| path.to_path_buf()); WATCH_GAP_CENSUS .get_or_init(Default::default) .lock() From 86e83f94a51f4279fd673c8948e81f392f20366a Mon Sep 17 00:00:00 2001 From: ualtinok <94532+ualtinok@users.noreply.github.com> Date: Wed, 28 Oct 2026 22:21:45 +0100 Subject: [PATCH 8/9] Scope-end refusal test waits 10 s for the terminal reply refused_shell_start_fails_fast_with_the_terminal_reason allowed 2 s for the refusal; a loaded macOS runner took longer. 10 s still sits well inside the 15 s startup timeout, so the test still proves the refusal arrives as its own terminal reply. Co-authored-by: alfonso-aft <289616620+alfonso-aft@users.noreply.github.com> --- crates/aft/tests/scope_end_kill.rs | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/crates/aft/tests/scope_end_kill.rs b/crates/aft/tests/scope_end_kill.rs index a3eda70da0..9d02504977 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"); } From 2c114b92957f08b71139885d4b334a09f38c590b Mon Sep 17 00:00:00 2001 From: ualtinok <94532+ualtinok@users.noreply.github.com> Date: Wed, 28 Oct 2026 22:50:12 +0100 Subject: [PATCH 9/9] Watch-gap health test looks up the census by connection The test spelled the database path itself and canonicalized it to match SQLite's resolved name on macOS, but on Windows canonicalize adds the \\?\ verbatim prefix SQLite never reports, so the lookup missed and the test read an empty census. Share the production key function instead. Co-authored-by: alfonso-aft <289616620+alfonso-aft@users.noreply.github.com> --- crates/aft/src/db/bash_watches.rs | 25 +++++++++++++++---------- crates/aft/src/subc/health.rs | 4 ++-- 2 files changed, 17 insertions(+), 12 deletions(-) diff --git a/crates/aft/src/db/bash_watches.rs b/crates/aft/src/db/bash_watches.rs index 1268ec2660..6f8b576897 100644 --- a/crates/aft/src/db/bash_watches.rs +++ b/crates/aft/src/db/bash_watches.rs @@ -286,12 +286,18 @@ static WATCH_GAP_CENSUS: std::sync::OnceLock< std::sync::Mutex>, > = std::sync::OnceLock::new(); -pub(crate) fn refresh_watch_gap_health(conn: &Connection) { - let key = conn - .path() +/// 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)); + .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))?; @@ -375,17 +381,16 @@ pub(crate) fn acknowledge_delivery( 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_path(path: &std::path::Path) -> GapCensus { - // The census is keyed by SQLite's own file name, which resolves symlinks - // (macOS temp folders live under /var -> /private/var), so look up the - // same resolved spelling. - let path = std::fs::canonicalize(path).unwrap_or_else(|_| path.to_path_buf()); +pub(crate) fn watch_gap_health_for(conn: &Connection) -> GapCensus { WATCH_GAP_CENSUS .get_or_init(Default::default) .lock() .unwrap() - .get(&path.to_string_lossy().into_owned()) + .get(&census_key(conn)) .copied() .unwrap_or_default() } diff --git a/crates/aft/src/subc/health.rs b/crates/aft/src/subc/health.rs index e714647437..4cf4fee049 100644 --- a/crates/aft/src/subc/health.rs +++ b/crates/aft/src/subc/health.rs @@ -9611,7 +9611,7 @@ mod watch_gap_tests { match_context: None, }; upsert_delivery_gap(&conn, &row).unwrap(); - let (gaps, missing, uncertain) = watch_gap_health_for_path(&path); + 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"); @@ -9630,7 +9630,7 @@ mod watch_gap_tests { ) .unwrap(); delete_delivery_gap(&conn, "opencode", "session", "task", "watch").unwrap(); - let (gaps, missing, uncertain) = watch_gap_health_for_path(&path); + 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());