feat: surface ecc2 daemon auto-merge activity

This commit is contained in:
Affaan Mustafa
2026-04-08 15:27:16 -07:00
parent d8c8178f92
commit dada133784
4 changed files with 267 additions and 7 deletions

View File

@@ -342,13 +342,33 @@ where
}
async fn maybe_auto_merge_ready_worktrees(db: &StateStore, cfg: &Config) -> Result<usize> {
maybe_auto_merge_ready_worktrees_with(cfg, || manager::merge_ready_worktrees(db, true)).await
maybe_auto_merge_ready_worktrees_with_recorder(
cfg,
|| manager::merge_ready_worktrees(db, true),
|merged, active, conflicted, dirty, failed| {
db.record_daemon_auto_merge_pass(merged, active, conflicted, dirty, failed)
},
)
.await
}
async fn maybe_auto_merge_ready_worktrees_with<F, Fut>(cfg: &Config, merge: F) -> Result<usize>
where
F: Fn() -> Fut,
Fut: Future<Output = Result<manager::WorktreeBulkMergeOutcome>>,
{
maybe_auto_merge_ready_worktrees_with_recorder(cfg, merge, |_, _, _, _, _| Ok(())).await
}
async fn maybe_auto_merge_ready_worktrees_with_recorder<F, Fut, R>(
cfg: &Config,
merge: F,
mut record: R,
) -> Result<usize>
where
F: Fn() -> Fut,
Fut: Future<Output = Result<manager::WorktreeBulkMergeOutcome>>,
R: FnMut(usize, usize, usize, usize, usize) -> Result<()>,
{
if !cfg.auto_merge_ready_worktrees {
return Ok(0);
@@ -356,22 +376,33 @@ where
let outcome = merge().await?;
let merged = outcome.merged.len();
let active = outcome.active_with_worktree_ids.len();
let conflicted = outcome.conflicted_session_ids.len();
let dirty = outcome.dirty_worktree_ids.len();
let failed = outcome.failures.len();
record(merged, active, conflicted, dirty, failed)?;
if merged > 0 {
tracing::info!("Auto-merged {merged} ready worktree(s)");
}
if !outcome.conflicted_session_ids.is_empty() {
if conflicted > 0 {
tracing::warn!(
"Skipped {} conflicted worktree(s) during auto-merge",
outcome.conflicted_session_ids.len()
conflicted
);
}
if !outcome.dirty_worktree_ids.is_empty() {
if dirty > 0 {
tracing::warn!(
"Skipped {} dirty worktree(s) during auto-merge",
outcome.dirty_worktree_ids.len()
dirty
);
}
if active > 0 {
tracing::info!("Skipped {active} active worktree(s) during auto-merge");
}
if failed > 0 {
tracing::warn!("Auto-merge failed for {failed} worktree(s)");
}
Ok(merged)
}
@@ -735,6 +766,12 @@ mod tests {
last_rebalance_at: None,
last_rebalance_rerouted: 0,
last_rebalance_leads: 0,
last_auto_merge_at: None,
last_auto_merge_merged: 0,
last_auto_merge_active_skipped: 0,
last_auto_merge_conflicted_skipped: 0,
last_auto_merge_dirty_skipped: 0,
last_auto_merge_failed: 0,
};
let order = std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));
let dispatch_order = order.clone();
@@ -792,6 +829,12 @@ mod tests {
last_rebalance_at: None,
last_rebalance_rerouted: 0,
last_rebalance_leads: 0,
last_auto_merge_at: None,
last_auto_merge_merged: 0,
last_auto_merge_active_skipped: 0,
last_auto_merge_conflicted_skipped: 0,
last_auto_merge_dirty_skipped: 0,
last_auto_merge_failed: 0,
};
let recorded = std::sync::Arc::new(std::sync::Mutex::new(None));
let recorded_clone = recorded.clone();
@@ -841,6 +884,12 @@ mod tests {
last_rebalance_at: Some(now - chrono::Duration::seconds(1)),
last_rebalance_rerouted: 0,
last_rebalance_leads: 1,
last_auto_merge_at: None,
last_auto_merge_merged: 0,
last_auto_merge_active_skipped: 0,
last_auto_merge_conflicted_skipped: 0,
last_auto_merge_dirty_skipped: 0,
last_auto_merge_failed: 0,
};
let calls = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
let calls_clone = calls.clone();
@@ -891,6 +940,12 @@ mod tests {
last_rebalance_at: Some(now - chrono::Duration::seconds(1)),
last_rebalance_rerouted: 0,
last_rebalance_leads: 1,
last_auto_merge_at: None,
last_auto_merge_merged: 0,
last_auto_merge_active_skipped: 0,
last_auto_merge_conflicted_skipped: 0,
last_auto_merge_dirty_skipped: 0,
last_auto_merge_failed: 0,
};
let calls = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
let calls_clone = calls.clone();
@@ -941,6 +996,12 @@ mod tests {
last_rebalance_at: Some(now),
last_rebalance_rerouted: 1,
last_rebalance_leads: 1,
last_auto_merge_at: None,
last_auto_merge_merged: 0,
last_auto_merge_active_skipped: 0,
last_auto_merge_conflicted_skipped: 0,
last_auto_merge_dirty_skipped: 0,
last_auto_merge_failed: 0,
};
let rebalance_calls = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
let rebalance_calls_clone = rebalance_calls.clone();

View File

@@ -1555,6 +1555,19 @@ impl fmt::Display for CoordinationStatus {
}
}
if let Some(last_auto_merge_at) = self.daemon_activity.last_auto_merge_at.as_ref() {
writeln!(
f,
"Last daemon auto-merge: {} merged / {} active / {} conflicted / {} dirty / {} failed @ {}",
self.daemon_activity.last_auto_merge_merged,
self.daemon_activity.last_auto_merge_active_skipped,
self.daemon_activity.last_auto_merge_conflicted_skipped,
self.daemon_activity.last_auto_merge_dirty_skipped,
self.daemon_activity.last_auto_merge_failed,
last_auto_merge_at.to_rfc3339()
)?;
}
Ok(())
}
}
@@ -1656,6 +1669,12 @@ mod tests {
last_rebalance_at: Some(now - Duration::seconds(2)),
last_rebalance_rerouted: 0,
last_rebalance_leads: 1,
last_auto_merge_at: Some(now - Duration::seconds(1)),
last_auto_merge_merged: 1,
last_auto_merge_active_skipped: 1,
last_auto_merge_conflicted_skipped: 0,
last_auto_merge_dirty_skipped: 0,
last_auto_merge_failed: 0,
}
}
@@ -3100,6 +3119,11 @@ mod tests {
assert!(rendered.contains("Last daemon dispatch: 3 routed / 1 deferred across 2 lead(s)"));
assert!(rendered.contains("Last daemon recovery dispatch: 2 handoff(s) across 1 lead(s)"));
assert!(rendered.contains("Last daemon rebalance: 0 handoff(s) across 1 lead(s)"));
assert!(
rendered.contains(
"Last daemon auto-merge: 1 merged / 1 active / 0 conflicted / 0 dirty / 0 failed"
)
);
}
#[test]

View File

@@ -27,6 +27,12 @@ pub struct DaemonActivity {
pub last_rebalance_at: Option<chrono::DateTime<chrono::Utc>>,
pub last_rebalance_rerouted: usize,
pub last_rebalance_leads: usize,
pub last_auto_merge_at: Option<chrono::DateTime<chrono::Utc>>,
pub last_auto_merge_merged: usize,
pub last_auto_merge_active_skipped: usize,
pub last_auto_merge_conflicted_skipped: usize,
pub last_auto_merge_dirty_skipped: usize,
pub last_auto_merge_failed: usize,
}
impl DaemonActivity {
@@ -162,7 +168,13 @@ impl StateStore {
last_recovery_dispatch_leads INTEGER NOT NULL DEFAULT 0,
last_rebalance_at TEXT,
last_rebalance_rerouted INTEGER NOT NULL DEFAULT 0,
last_rebalance_leads INTEGER NOT NULL DEFAULT 0
last_rebalance_leads INTEGER NOT NULL DEFAULT 0,
last_auto_merge_at TEXT,
last_auto_merge_merged INTEGER NOT NULL DEFAULT 0,
last_auto_merge_active_skipped INTEGER NOT NULL DEFAULT 0,
last_auto_merge_conflicted_skipped INTEGER NOT NULL DEFAULT 0,
last_auto_merge_dirty_skipped INTEGER NOT NULL DEFAULT 0,
last_auto_merge_failed INTEGER NOT NULL DEFAULT 0
);
CREATE INDEX IF NOT EXISTS idx_sessions_state ON sessions(state);
@@ -241,6 +253,60 @@ impl StateStore {
.context("Failed to add chronic_saturation_streak column to daemon_activity table")?;
}
if !self.has_column("daemon_activity", "last_auto_merge_at")? {
self.conn
.execute(
"ALTER TABLE daemon_activity ADD COLUMN last_auto_merge_at TEXT",
[],
)
.context("Failed to add last_auto_merge_at column to daemon_activity table")?;
}
if !self.has_column("daemon_activity", "last_auto_merge_merged")? {
self.conn
.execute(
"ALTER TABLE daemon_activity ADD COLUMN last_auto_merge_merged INTEGER NOT NULL DEFAULT 0",
[],
)
.context("Failed to add last_auto_merge_merged column to daemon_activity table")?;
}
if !self.has_column("daemon_activity", "last_auto_merge_active_skipped")? {
self.conn
.execute(
"ALTER TABLE daemon_activity ADD COLUMN last_auto_merge_active_skipped INTEGER NOT NULL DEFAULT 0",
[],
)
.context("Failed to add last_auto_merge_active_skipped column to daemon_activity table")?;
}
if !self.has_column("daemon_activity", "last_auto_merge_conflicted_skipped")? {
self.conn
.execute(
"ALTER TABLE daemon_activity ADD COLUMN last_auto_merge_conflicted_skipped INTEGER NOT NULL DEFAULT 0",
[],
)
.context("Failed to add last_auto_merge_conflicted_skipped column to daemon_activity table")?;
}
if !self.has_column("daemon_activity", "last_auto_merge_dirty_skipped")? {
self.conn
.execute(
"ALTER TABLE daemon_activity ADD COLUMN last_auto_merge_dirty_skipped INTEGER NOT NULL DEFAULT 0",
[],
)
.context("Failed to add last_auto_merge_dirty_skipped column to daemon_activity table")?;
}
if !self.has_column("daemon_activity", "last_auto_merge_failed")? {
self.conn
.execute(
"ALTER TABLE daemon_activity ADD COLUMN last_auto_merge_failed INTEGER NOT NULL DEFAULT 0",
[],
)
.context("Failed to add last_auto_merge_failed column to daemon_activity table")?;
}
Ok(())
}
@@ -643,7 +709,10 @@ impl StateStore {
"SELECT last_dispatch_at, last_dispatch_routed, last_dispatch_deferred, last_dispatch_leads,
chronic_saturation_streak,
last_recovery_dispatch_at, last_recovery_dispatch_routed, last_recovery_dispatch_leads,
last_rebalance_at, last_rebalance_rerouted, last_rebalance_leads
last_rebalance_at, last_rebalance_rerouted, last_rebalance_leads,
last_auto_merge_at, last_auto_merge_merged, last_auto_merge_active_skipped,
last_auto_merge_conflicted_skipped, last_auto_merge_dirty_skipped,
last_auto_merge_failed
FROM daemon_activity
WHERE id = 1",
[],
@@ -677,6 +746,12 @@ impl StateStore {
last_rebalance_at: parse_ts(row.get(8)?)?,
last_rebalance_rerouted: row.get::<_, i64>(9)? as usize,
last_rebalance_leads: row.get::<_, i64>(10)? as usize,
last_auto_merge_at: parse_ts(row.get(11)?)?,
last_auto_merge_merged: row.get::<_, i64>(12)? as usize,
last_auto_merge_active_skipped: row.get::<_, i64>(13)? as usize,
last_auto_merge_conflicted_skipped: row.get::<_, i64>(14)? as usize,
last_auto_merge_dirty_skipped: row.get::<_, i64>(15)? as usize,
last_auto_merge_failed: row.get::<_, i64>(16)? as usize,
})
},
)
@@ -742,6 +817,36 @@ impl StateStore {
Ok(())
}
pub fn record_daemon_auto_merge_pass(
&self,
merged: usize,
active_skipped: usize,
conflicted_skipped: usize,
dirty_skipped: usize,
failed: usize,
) -> Result<()> {
self.conn.execute(
"UPDATE daemon_activity
SET last_auto_merge_at = ?1,
last_auto_merge_merged = ?2,
last_auto_merge_active_skipped = ?3,
last_auto_merge_conflicted_skipped = ?4,
last_auto_merge_dirty_skipped = ?5,
last_auto_merge_failed = ?6
WHERE id = 1",
rusqlite::params![
chrono::Utc::now().to_rfc3339(),
merged as i64,
active_skipped as i64,
conflicted_skipped as i64,
dirty_skipped as i64,
failed as i64,
],
)?;
Ok(())
}
pub fn delegated_children(&self, session_id: &str, limit: usize) -> Result<Vec<String>> {
let mut stmt = self.conn.prepare(
"SELECT to_session
@@ -1117,6 +1222,7 @@ mod tests {
db.record_daemon_dispatch_pass(4, 1, 2)?;
db.record_daemon_recovery_dispatch_pass(2, 1)?;
db.record_daemon_rebalance_pass(3, 1)?;
db.record_daemon_auto_merge_pass(2, 1, 1, 1, 0)?;
let activity = db.daemon_activity()?;
assert_eq!(activity.last_dispatch_routed, 4);
@@ -1127,9 +1233,15 @@ mod tests {
assert_eq!(activity.last_recovery_dispatch_leads, 1);
assert_eq!(activity.last_rebalance_rerouted, 3);
assert_eq!(activity.last_rebalance_leads, 1);
assert_eq!(activity.last_auto_merge_merged, 2);
assert_eq!(activity.last_auto_merge_active_skipped, 1);
assert_eq!(activity.last_auto_merge_conflicted_skipped, 1);
assert_eq!(activity.last_auto_merge_dirty_skipped, 1);
assert_eq!(activity.last_auto_merge_failed, 0);
assert!(activity.last_dispatch_at.is_some());
assert!(activity.last_recovery_dispatch_at.is_some());
assert!(activity.last_rebalance_at.is_some());
assert!(activity.last_auto_merge_at.is_some());
Ok(())
}
@@ -1156,6 +1268,12 @@ mod tests {
last_rebalance_at: None,
last_rebalance_rerouted: 0,
last_rebalance_leads: 0,
last_auto_merge_at: None,
last_auto_merge_merged: 0,
last_auto_merge_active_skipped: 0,
last_auto_merge_conflicted_skipped: 0,
last_auto_merge_dirty_skipped: 0,
last_auto_merge_failed: 0,
};
assert!(unresolved.prefers_rebalance_first());
assert!(unresolved.dispatch_cooloff_active());

View File

@@ -1912,6 +1912,18 @@ impl Dashboard {
}
}
if let Some(last_auto_merge_at) = self.daemon_activity.last_auto_merge_at.as_ref() {
lines.push(format!(
"Last daemon auto-merge {} merged / {} active / {} conflicted / {} dirty / {} failed @ {}",
self.daemon_activity.last_auto_merge_merged,
self.daemon_activity.last_auto_merge_active_skipped,
self.daemon_activity.last_auto_merge_conflicted_skipped,
self.daemon_activity.last_auto_merge_dirty_skipped,
self.daemon_activity.last_auto_merge_failed,
self.short_timestamp(&last_auto_merge_at.to_rfc3339())
));
}
if let Some(route_preview) = self.selected_route_preview.as_ref() {
lines.push(format!("Next route {route_preview}"));
}
@@ -2774,6 +2786,12 @@ mod tests {
last_rebalance_at: Some(now + chrono::Duration::seconds(2)),
last_rebalance_rerouted: 1,
last_rebalance_leads: 1,
last_auto_merge_at: Some(now + chrono::Duration::seconds(3)),
last_auto_merge_merged: 2,
last_auto_merge_active_skipped: 1,
last_auto_merge_conflicted_skipped: 1,
last_auto_merge_dirty_skipped: 0,
last_auto_merge_failed: 0,
};
let text = dashboard.selected_session_metrics_text();
@@ -2782,6 +2800,9 @@ mod tests {
assert!(text.contains("Last daemon dispatch 4 routed / 2 deferred across 2 lead(s)"));
assert!(text.contains("Last daemon recovery dispatch 1 handoff(s) across 1 lead(s)"));
assert!(text.contains("Last daemon rebalance 1 handoff(s) across 1 lead(s)"));
assert!(
text.contains("Last daemon auto-merge 2 merged / 1 active / 1 conflicted / 0 dirty / 0 failed")
);
}
#[test]
@@ -2809,6 +2830,12 @@ mod tests {
last_rebalance_at: Some(Utc::now()),
last_rebalance_rerouted: 1,
last_rebalance_leads: 1,
last_auto_merge_at: None,
last_auto_merge_merged: 0,
last_auto_merge_active_skipped: 0,
last_auto_merge_conflicted_skipped: 0,
last_auto_merge_dirty_skipped: 0,
last_auto_merge_failed: 0,
};
let text = dashboard.selected_session_metrics_text();
@@ -2840,6 +2867,12 @@ mod tests {
last_rebalance_at: Some(Utc::now()),
last_rebalance_rerouted: 1,
last_rebalance_leads: 1,
last_auto_merge_at: None,
last_auto_merge_merged: 0,
last_auto_merge_active_skipped: 0,
last_auto_merge_conflicted_skipped: 0,
last_auto_merge_dirty_skipped: 0,
last_auto_merge_failed: 0,
};
let text = dashboard.selected_session_metrics_text();
@@ -2873,6 +2906,12 @@ mod tests {
last_rebalance_at: Some(Utc::now()),
last_rebalance_rerouted: 0,
last_rebalance_leads: 1,
last_auto_merge_at: None,
last_auto_merge_merged: 0,
last_auto_merge_active_skipped: 0,
last_auto_merge_conflicted_skipped: 0,
last_auto_merge_dirty_skipped: 0,
last_auto_merge_failed: 0,
};
let text = dashboard.selected_session_metrics_text();
@@ -2907,6 +2946,12 @@ mod tests {
last_rebalance_at: Some(now),
last_rebalance_rerouted: 1,
last_rebalance_leads: 1,
last_auto_merge_at: None,
last_auto_merge_merged: 0,
last_auto_merge_active_skipped: 0,
last_auto_merge_conflicted_skipped: 0,
last_auto_merge_dirty_skipped: 0,
last_auto_merge_failed: 0,
};
let text = dashboard.selected_session_metrics_text();
@@ -2956,6 +3001,12 @@ mod tests {
last_rebalance_at: Some(now),
last_rebalance_rerouted: 1,
last_rebalance_leads: 1,
last_auto_merge_at: None,
last_auto_merge_merged: 0,
last_auto_merge_active_skipped: 0,
last_auto_merge_conflicted_skipped: 0,
last_auto_merge_dirty_skipped: 0,
last_auto_merge_failed: 0,
};
let text = dashboard.selected_session_metrics_text();
@@ -3052,6 +3103,12 @@ mod tests {
last_rebalance_at: Some(now),
last_rebalance_rerouted: 1,
last_rebalance_leads: 1,
last_auto_merge_at: None,
last_auto_merge_merged: 0,
last_auto_merge_active_skipped: 0,
last_auto_merge_conflicted_skipped: 0,
last_auto_merge_dirty_skipped: 0,
last_auto_merge_failed: 0,
};
let text = dashboard.selected_session_metrics_text();