zsp/everything-claude-code
1
0
Fork 0
mirror of https://github.com/affaan-m/everything-claude-code.git synced 2026-04-09 10:53:34 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat: default ecc2 worktrees through policy

Browse Source
This commit is contained in:
Affaan Mustafa
2026-04-08 15:58:31 -07:00
parent 86cbe3d616
commit b3f781a648
8 changed files with 576 additions and 326 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
ecc2/src/config/mod.rs
View File

@@ -31,6 +31,7 @@ pub struct Config {
pub default_agent: String,
pub auto_dispatch_unread_handoffs: bool,
pub auto_dispatch_limit_per_session: usize,
pub auto_create_worktrees: bool,
pub auto_merge_ready_worktrees: bool,
pub cost_budget_usd: f64,
pub token_budget: u64,
@@ -58,6 +59,7 @@ impl Default for Config {
default_agent: "claude".to_string(),
auto_dispatch_unread_handoffs: false,
auto_dispatch_limit_per_session: 5,
auto_create_worktrees: true,
auto_merge_ready_worktrees: false,
cost_budget_usd: 10.0,
token_budget: 500_000,
@@ -156,6 +158,7 @@ theme = "Dark"
config.auto_dispatch_limit_per_session,
defaults.auto_dispatch_limit_per_session
);
assert_eq!(config.auto_create_worktrees, defaults.auto_create_worktrees);
assert_eq!(
config.auto_merge_ready_worktrees,
defaults.auto_merge_ready_worktrees
@@ -185,6 +188,7 @@ theme = "Dark"
let mut config = Config::default();
config.auto_dispatch_unread_handoffs = true;
config.auto_dispatch_limit_per_session = 9;
config.auto_create_worktrees = false;
config.auto_merge_ready_worktrees = true;
config.save_to_path(&path).unwrap();
@@ -193,6 +197,7 @@ theme = "Dark"
assert!(loaded.auto_dispatch_unread_handoffs);
assert_eq!(loaded.auto_dispatch_limit_per_session, 9);
assert!(!loaded.auto_create_worktrees);
assert!(loaded.auto_merge_ready_worktrees);
let _ = std::fs::remove_file(path);

290
ecc2/src/main.rs
View File

@@ -18,6 +18,28 @@ struct Cli {
command: Option<Commands>,
}
#[derive(clap::Args, Debug, Clone, Default)]
struct WorktreePolicyArgs {
/// Create a dedicated worktree
#[arg(short = 'w', long = "worktree", action = clap::ArgAction::SetTrue, overrides_with = "no_worktree")]
worktree: bool,
/// Skip dedicated worktree creation
#[arg(long = "no-worktree", action = clap::ArgAction::SetTrue, overrides_with = "worktree")]
no_worktree: bool,
}
impl WorktreePolicyArgs {
fn resolve(&self, cfg: &config::Config) -> bool {
if self.worktree {
true
} else if self.no_worktree {
false
} else {
cfg.auto_create_worktrees
}
}
}
#[derive(clap::Subcommand, Debug)]
enum Commands {
/// Launch the TUI dashboard
@@ -30,9 +52,8 @@ enum Commands {
/// Agent type (claude, codex, custom)
#[arg(short, long, default_value = "claude")]
agent: String,
/// Create a dedicated worktree for this session
#[arg(short, long)]
worktree: bool,
#[command(flatten)]
worktree: WorktreePolicyArgs,
/// Source session to delegate from
#[arg(long)]
from_session: Option<String>,
@@ -47,9 +68,8 @@ enum Commands {
/// Agent type (claude, codex, custom)
#[arg(short, long, default_value = "claude")]
agent: String,
/// Create a dedicated worktree for the delegated session
#[arg(short, long, default_value_t = true)]
worktree: bool,
#[command(flatten)]
worktree: WorktreePolicyArgs,
},
/// Route work to an existing delegate when possible, otherwise spawn a new one
Assign {
@@ -61,9 +81,8 @@ enum Commands {
/// Agent type (claude, codex, custom)
#[arg(short, long, default_value = "claude")]
agent: String,
/// Create a dedicated worktree if a new delegate must be spawned
#[arg(short, long, default_value_t = true)]
worktree: bool,
#[command(flatten)]
worktree: WorktreePolicyArgs,
},
/// Route unread task handoffs from a lead session inbox through the assignment policy
DrainInbox {
@@ -72,9 +91,8 @@ enum Commands {
/// Agent type for routed delegates
#[arg(short, long, default_value = "claude")]
agent: String,
/// Create a dedicated worktree if new delegates must be spawned
#[arg(short, long, default_value_t = true)]
worktree: bool,
#[command(flatten)]
worktree: WorktreePolicyArgs,
/// Maximum unread task handoffs to route
#[arg(long, default_value_t = 5)]
limit: usize,
@@ -84,9 +102,8 @@ enum Commands {
/// Agent type for routed delegates
#[arg(short, long, default_value = "claude")]
agent: String,
/// Create a dedicated worktree if new delegates must be spawned
#[arg(short, long, default_value_t = true)]
worktree: bool,
#[command(flatten)]
worktree: WorktreePolicyArgs,
/// Maximum lead sessions to sweep in one pass
#[arg(long, default_value_t = 10)]
lead_limit: usize,
@@ -96,9 +113,8 @@ enum Commands {
/// Agent type for routed delegates
#[arg(short, long, default_value = "claude")]
agent: String,
/// Create a dedicated worktree if new delegates must be spawned
#[arg(short, long, default_value_t = true)]
worktree: bool,
#[command(flatten)]
worktree: WorktreePolicyArgs,
/// Maximum lead sessions to sweep in one pass
#[arg(long, default_value_t = 10)]
lead_limit: usize,
@@ -129,9 +145,8 @@ enum Commands {
/// Agent type for routed delegates
#[arg(short, long, default_value = "claude")]
agent: String,
/// Create a dedicated worktree if new delegates must be spawned
#[arg(short, long, default_value_t = true)]
worktree: bool,
#[command(flatten)]
worktree: WorktreePolicyArgs,
/// Maximum lead sessions to sweep in one pass
#[arg(long, default_value_t = 10)]
lead_limit: usize,
@@ -150,9 +165,8 @@ enum Commands {
/// Agent type for routed delegates
#[arg(short, long, default_value = "claude")]
agent: String,
/// Create a dedicated worktree if new delegates must be spawned
#[arg(short, long, default_value_t = true)]
worktree: bool,
#[command(flatten)]
worktree: WorktreePolicyArgs,
/// Maximum lead sessions to sweep in one pass
#[arg(long, default_value_t = 10)]
lead_limit: usize,
@@ -164,9 +178,8 @@ enum Commands {
/// Agent type for routed delegates
#[arg(short, long, default_value = "claude")]
agent: String,
/// Create a dedicated worktree if new delegates must be spawned
#[arg(short, long, default_value_t = true)]
worktree: bool,
#[command(flatten)]
worktree: WorktreePolicyArgs,
/// Maximum handoffs to reroute in one pass
#[arg(long, default_value_t = 5)]
limit: usize,
@@ -319,9 +332,10 @@ async fn main() -> Result<()> {
Some(Commands::Start {
task,
agent,
worktree: use_worktree,
worktree,
from_session,
}) => {
let use_worktree = worktree.resolve(&cfg);
let session_id =
session::manager::create_session(&db, &cfg, &task, &agent, use_worktree).await?;
if let Some(from_session) = from_session {
@@ -334,8 +348,9 @@ async fn main() -> Result<()> {
from_session,
task,
agent,
worktree: use_worktree,
worktree,
}) => {
let use_worktree = worktree.resolve(&cfg);
let from_id = resolve_session_id(&db, &from_session)?;
let source = db
.get_session(&from_id)?
@@ -361,18 +376,13 @@ async fn main() -> Result<()> {
from_session,
task,
agent,
worktree: use_worktree,
worktree,
}) => {
let use_worktree = worktree.resolve(&cfg);
let lead_id = resolve_session_id(&db, &from_session)?;
let outcome = session::manager::assign_session(
&db,
&cfg,
&lead_id,
&task,
&agent,
use_worktree,
)
.await?;
let outcome =
session::manager::assign_session(&db, &cfg, &lead_id, &task, &agent, use_worktree)
.await?;
if session::manager::assignment_action_routes_work(outcome.action) {
println!(
"Assignment routed: {} -> {} ({})",
@@ -396,32 +406,28 @@ async fn main() -> Result<()> {
Some(Commands::DrainInbox {
session_id,
agent,
worktree: use_worktree,
worktree,
limit,
}) => {
let use_worktree = worktree.resolve(&cfg);
let lead_id = resolve_session_id(&db, &session_id)?;
let outcomes = session::manager::drain_inbox(
&db,
&cfg,
&lead_id,
&agent,
use_worktree,
limit,
)
.await?;
let outcomes =
session::manager::drain_inbox(&db, &cfg, &lead_id, &agent, use_worktree, limit)
.await?;
if outcomes.is_empty() {
println!("No unread task handoffs for {}", short_session(&lead_id));
} else {
let routed_count = outcomes
.iter()
.filter(|outcome| session::manager::assignment_action_routes_work(outcome.action))
.filter(|outcome| {
session::manager::assignment_action_routes_work(outcome.action)
})
.count();
let deferred_count = outcomes.len().saturating_sub(routed_count);
println!(
"Processed {} inbox task handoff(s) from {} ({} routed, {} deferred)",
outcomes.len(),
short_session(&lead_id)
,
short_session(&lead_id),
routed_count,
deferred_count
);
@@ -445,9 +451,10 @@ async fn main() -> Result<()> {
}
Some(Commands::AutoDispatch {
agent,
worktree: use_worktree,
worktree,
lead_limit,
}) => {
let use_worktree = worktree.resolve(&cfg);
let outcomes = session::manager::auto_dispatch_backlog(
&db,
&cfg,
@@ -459,14 +466,17 @@ async fn main() -> Result<()> {
if outcomes.is_empty() {
println!("No unread task handoff backlog found");
} else {
let total_processed: usize = outcomes.iter().map(|outcome| outcome.routed.len()).sum();
let total_processed: usize =
outcomes.iter().map(|outcome| outcome.routed.len()).sum();
let total_routed: usize = outcomes
.iter()
.map(|outcome| {
outcome
.routed
.iter()
.filter(|item| session::manager::assignment_action_routes_work(item.action))
.filter(|item| {
session::manager::assignment_action_routes_work(item.action)
})
.count()
})
.sum();
@@ -497,18 +507,15 @@ async fn main() -> Result<()> {
}
Some(Commands::CoordinateBacklog {
agent,
worktree: use_worktree,
worktree,
lead_limit,
json,
check,
until_healthy,
max_passes,
}) => {
let pass_budget = if until_healthy {
max_passes.max(1)
} else {
1
};
let use_worktree = worktree.resolve(&cfg);
let pass_budget = if until_healthy { max_passes.max(1) } else { 1 };
let run = run_coordination_loop(
&db,
&cfg,
@@ -542,12 +549,13 @@ async fn main() -> Result<()> {
}
Some(Commands::MaintainCoordination {
agent,
worktree: use_worktree,
worktree,
lead_limit,
json,
check,
max_passes,
}) => {
let use_worktree = worktree.resolve(&cfg);
let initial_status = session::manager::get_coordination_status(&db, &cfg)?;
let run = if matches!(
initial_status.health,
@@ -591,17 +599,13 @@ async fn main() -> Result<()> {
}
Some(Commands::RebalanceAll {
agent,
worktree: use_worktree,
worktree,
lead_limit,
}) => {
let outcomes = session::manager::rebalance_all_teams(
&db,
&cfg,
&agent,
use_worktree,
lead_limit,
)
.await?;
let use_worktree = worktree.resolve(&cfg);
let outcomes =
session::manager::rebalance_all_teams(&db, &cfg, &agent, use_worktree, lead_limit)
.await?;
if outcomes.is_empty() {
println!("No delegate backlog needed global rebalancing");
} else {
@@ -624,9 +628,10 @@ async fn main() -> Result<()> {
Some(Commands::RebalanceTeam {
session_id,
agent,
worktree: use_worktree,
worktree,
limit,
}) => {
let use_worktree = worktree.resolve(&cfg);
let lead_id = resolve_session_id(&db, &session_id)?;
let outcomes = session::manager::rebalance_team_backlog(
&db,
@@ -638,7 +643,10 @@ async fn main() -> Result<()> {
)
.await?;
if outcomes.is_empty() {
println!("No delegate backlog needed rebalancing for {}", short_session(&lead_id));
println!(
"No delegate backlog needed rebalancing for {}",
short_session(&lead_id)
);
} else {
println!(
"Rebalanced {} task handoff(s) for {}",
@@ -779,12 +787,9 @@ async fn main() -> Result<()> {
} else {
let id = session_id.unwrap_or_else(|| "latest".to_string());
let resolved_id = resolve_session_id(&db, &id)?;
let outcome = session::manager::merge_session_worktree(
&db,
&resolved_id,
!keep_worktree,
)
.await?;
let outcome =
session::manager::merge_session_worktree(&db, &resolved_id, !keep_worktree)
.await?;
if json {
println!("{}", serde_json::to_string_pretty(&outcome)?);
} else {
@@ -821,7 +826,11 @@ async fn main() -> Result<()> {
let to = resolve_session_id(&db, &to)?;
let message = build_message(kind, text, context, file)?;
comms::send(&db, &from, &to, &message)?;
println!("Message sent: {} -> {}", short_session(&from), short_session(&to));
println!(
"Message sent: {} -> {}",
short_session(&from),
short_session(&to)
);
}
MessageCommands::Inbox { session_id, limit } => {
let session_id = resolve_session_id(&db, &session_id)?;
@@ -1057,7 +1066,10 @@ struct WorktreeResolutionReport {
resolution_steps: Vec<String>,
}
fn build_worktree_status_report(session: &session::Session, include_patch: bool) -> Result<WorktreeStatusReport> {
fn build_worktree_status_report(
session: &session::Session,
include_patch: bool,
) -> Result<WorktreeStatusReport> {
let Some(worktree) = session.worktree.as_ref() else {
return Ok(WorktreeStatusReport {
session_id: session.id.clone(),
@@ -1117,7 +1129,9 @@ fn build_worktree_status_report(session: &session::Session, include_patch: bool)
})
}
fn build_worktree_resolution_report(session: &session::Session) -> Result<WorktreeResolutionReport> {
fn build_worktree_resolution_report(
session: &session::Session,
) -> Result<WorktreeResolutionReport> {
let Some(worktree) = session.worktree.as_ref() else {
return Ok(WorktreeResolutionReport {
session_id: session.id.clone(),
@@ -1139,11 +1153,17 @@ fn build_worktree_resolution_report(session: &session::Session) -> Result<Worktr
let conflicted = merge_readiness.status == worktree::MergeReadinessStatus::Conflicted;
let resolution_steps = if conflicted {
vec![
format!("Inspect current patch: ecc worktree-status {} --patch", session.id),
format!(
"Inspect current patch: ecc worktree-status {} --patch",
session.id
),
format!("Open worktree: cd {}", worktree.path.display()),
"Resolve conflicts and stage files: git add <paths>".to_string(),
format!("Commit the resolution on {}: git commit", worktree.branch),
format!("Re-check readiness: ecc worktree-status {} --check", session.id),
format!(
"Re-check readiness: ecc worktree-status {} --check",
session.id
),
format!("Merge when clear: ecc merge-worktree {}", session.id),
]
} else {
@@ -1183,7 +1203,8 @@ fn format_worktree_status_human(report: &WorktreeStatusReport) -> String {
if let Some(path) = report.path.as_ref() {
lines.push(format!("Path {path}"));
}
if let (Some(branch), Some(base_branch)) = (report.branch.as_ref(), report.base_branch.as_ref()) {
if let (Some(branch), Some(base_branch)) = (report.branch.as_ref(), report.base_branch.as_ref())
{
lines.push(format!("Branch {branch} (base {base_branch})"));
}
if let Some(diff_summary) = report.diff_summary.as_ref() {
@@ -1237,7 +1258,8 @@ fn format_worktree_resolution_human(report: &WorktreeResolutionReport) -> String
if let Some(path) = report.path.as_ref() {
lines.push(format!("Path {path}"));
}
if let (Some(branch), Some(base_branch)) = (report.branch.as_ref(), report.base_branch.as_ref()) {
if let (Some(branch), Some(base_branch)) = (report.branch.as_ref(), report.base_branch.as_ref())
{
lines.push(format!("Branch {branch} (base {base_branch})"));
}
lines.push(report.summary.clone());
@@ -1295,12 +1317,11 @@ fn format_worktree_merge_human(outcome: &session::manager::WorktreeMergeOutcome)
lines.join("\n")
}
fn format_bulk_worktree_merge_human(outcome: &session::manager::WorktreeBulkMergeOutcome) -> String {
fn format_bulk_worktree_merge_human(
outcome: &session::manager::WorktreeBulkMergeOutcome,
) -> String {
let mut lines = Vec::new();
lines.push(format!(
"Merged {} ready worktree(s)",
outcome.merged.len()
));
lines.push(format!("Merged {} ready worktree(s)", outcome.merged.len()));
for merged in &outcome.merged {
lines.push(format!(
@@ -1427,7 +1448,10 @@ fn summarize_coordinate_backlog(
.map(|rebalance| rebalance.rerouted.len())
.sum();
let message = if total_routed == 0 && total_rerouted == 0 && outcome.remaining_backlog_sessions == 0 {
let message = if total_routed == 0
&& total_rerouted == 0
&& outcome.remaining_backlog_sessions == 0
{
"Backlog already clear".to_string()
} else {
format!(
@@ -1470,11 +1494,7 @@ fn coordination_status_exit_code(status: &session::manager::CoordinationStatus)
}
}
fn send_handoff_message(
db: &session::store::StateStore,
from_id: &str,
to_id: &str,
) -> Result<()> {
fn send_handoff_message(db: &session::store::StateStore, from_id: &str, to_id: &str) -> Result<()> {
let from_session = db
.get_session(from_id)?
.ok_or_else(|| anyhow::anyhow!("Session not found: {from_id}"))?;
@@ -1508,6 +1528,37 @@ fn send_handoff_message(
#[cfg(test)]
mod tests {
use super::*;
use crate::config::Config;
#[test]
fn worktree_policy_defaults_to_config_setting() {
let mut cfg = Config::default();
let policy = WorktreePolicyArgs::default();
assert!(policy.resolve(&cfg));
cfg.auto_create_worktrees = false;
assert!(!policy.resolve(&cfg));
}
#[test]
fn worktree_policy_explicit_flags_override_config_setting() {
let mut cfg = Config::default();
cfg.auto_create_worktrees = false;
assert!(WorktreePolicyArgs {
worktree: true,
no_worktree: false,
}
.resolve(&cfg));
cfg.auto_create_worktrees = true;
assert!(!WorktreePolicyArgs {
worktree: false,
no_worktree: true,
}
.resolve(&cfg));
}
#[test]
fn cli_parses_resume_command() {
@@ -1586,6 +1637,20 @@ mod tests {
}
}
#[test]
fn cli_parses_start_no_worktree_override() {
let cli = Cli::try_parse_from(["ecc", "start", "--task", "Follow up", "--no-worktree"])
.expect("start --no-worktree should parse");
match cli.command {
Some(Commands::Start { worktree, .. }) => {
assert!(!worktree.worktree);
assert!(worktree.no_worktree);
}
_ => panic!("expected start subcommand"),
}
}
#[test]
fn cli_parses_delegate_command() {
let cli = Cli::try_parse_from([
@@ -1614,6 +1679,20 @@ mod tests {
}
}
#[test]
fn cli_parses_delegate_worktree_override() {
let cli = Cli::try_parse_from(["ecc", "delegate", "planner", "--worktree"])
.expect("delegate --worktree should parse");
match cli.command {
Some(Commands::Delegate { worktree, .. }) => {
assert!(worktree.worktree);
assert!(!worktree.no_worktree);
}
_ => panic!("expected delegate subcommand"),
}
}
#[test]
fn cli_parses_team_command() {
let cli = Cli::try_parse_from(["ecc", "team", "planner", "--depth", "3"])
@@ -1704,9 +1783,7 @@ mod tests {
let command = err.command.expect("expected command");
let Commands::WorktreeStatus {
session_id,
all,
..
session_id, all, ..
} = command
else {
panic!("expected worktree-status subcommand");
@@ -1807,8 +1884,9 @@ mod tests {
#[test]
fn cli_parses_worktree_resolution_flags() {
let cli = Cli::try_parse_from(["ecc", "worktree-resolution", "planner", "--json", "--check"])
.expect("worktree-resolution flags should parse");
let cli =
Cli::try_parse_from(["ecc", "worktree-resolution", "planner", "--json", "--check"])
.expect("worktree-resolution flags should parse");
match cli.command {
Some(Commands::WorktreeResolution {
@@ -2280,9 +2358,7 @@ mod tests {
match cli.command {
Some(Commands::AutoDispatch {
agent,
lead_limit,
..
agent, lead_limit, ..
}) => {
assert_eq!(agent, "claude");
assert_eq!(lead_limit, 4);
@@ -2406,9 +2482,7 @@ mod tests {
match cli.command {
Some(Commands::RebalanceAll {
agent,
lead_limit,
..
agent, lead_limit, ..
}) => {
assert_eq!(agent, "claude");
assert_eq!(lead_limit, 6);

5
ecc2/src/session/daemon.rs
View File

@@ -392,10 +392,7 @@ where
);
}
if dirty > 0 {
tracing::warn!(
"Skipped {} dirty worktree(s) during auto-merge",
dirty
);
tracing::warn!("Skipped {} dirty worktree(s) during auto-merge", dirty);
}
if active > 0 {
tracing::info!("Skipped {active} active worktree(s) during auto-merge");

228
ecc2/src/session/manager.rs
View File

@@ -101,7 +101,8 @@ pub async fn drain_inbox(
) -> Result<Vec<InboxDrainOutcome>> {
let repo_root =
std::env::current_dir().context("Failed to resolve current working directory")?;
let runner_program = std::env::current_exe().context("Failed to resolve ECC executable path")?;
let runner_program =
std::env::current_exe().context("Failed to resolve ECC executable path")?;
let lead = resolve_session(db, lead_id)?;
let messages = db.unread_task_handoffs_for_session(&lead.id, limit)?;
let mut outcomes = Vec::new();
@@ -184,7 +185,12 @@ pub async fn rebalance_all_teams(
for session in sessions
.into_iter()
.filter(|session| matches!(session.state, SessionState::Running | SessionState::Pending | SessionState::Idle))
.filter(|session| {
matches!(
session.state,
SessionState::Running | SessionState::Pending | SessionState::Idle
)
})
.take(lead_limit)
{
let rerouted = rebalance_team_backlog(
@@ -245,7 +251,8 @@ pub async fn rebalance_team_backlog(
) -> Result<Vec<RebalanceOutcome>> {
let repo_root =
std::env::current_dir().context("Failed to resolve current working directory")?;
let runner_program = std::env::current_exe().context("Failed to resolve ECC executable path")?;
let runner_program =
std::env::current_exe().context("Failed to resolve ECC executable path")?;
let lead = resolve_session(db, lead_id)?;
let mut outcomes = Vec::new();
@@ -888,7 +895,15 @@ async fn queue_session_in_dir_with_runner_program(
.map(|worktree| worktree.path.as_path())
.unwrap_or(repo_root);
match spawn_session_runner_for_program(task, &session.id, agent_type, working_dir, runner_program).await {
match spawn_session_runner_for_program(
task,
&session.id,
agent_type,
working_dir,
runner_program,
)
.await
{
Ok(()) => Ok(session.id),
Err(error) => {
db.update_state(&session.id, &SessionState::Failed)?;
@@ -989,7 +1004,11 @@ async fn spawn_session_runner(
.await
}
fn direct_delegate_sessions(db: &StateStore, lead_id: &str, agent_type: &str) -> Result<Vec<Session>> {
fn direct_delegate_sessions(
db: &StateStore,
lead_id: &str,
agent_type: &str,
) -> Result<Vec<Session>> {
let mut sessions = Vec::new();
for child_id in db.delegated_children(lead_id, 50)? {
let Some(session) = db.get_session(&child_id)? else {
@@ -1101,12 +1120,7 @@ async fn spawn_session_runner_for_program(
.stdout(Stdio::null())
.stderr(Stdio::null())
.spawn()
.with_context(|| {
format!(
"Failed to spawn ECC runner from {}",
current_exe.display()
)
})?;
.with_context(|| format!("Failed to spawn ECC runner from {}", current_exe.display()))?;
child
.id()
@@ -1114,7 +1128,12 @@ async fn spawn_session_runner_for_program(
Ok(())
}
fn build_agent_command(agent_program: &Path, task: &str, session_id: &str, working_dir: &Path) -> Command {
fn build_agent_command(
agent_program: &Path,
task: &str,
session_id: &str,
working_dir: &Path,
) -> Command {
let mut command = Command::new(agent_program);
command
.arg("--print")
@@ -1414,7 +1433,11 @@ impl fmt::Display for TeamStatus {
writeln!(f, "Branch: {}", worktree.branch)?;
}
let lead_handoff_backlog = self.handoff_backlog.get(&self.root.id).copied().unwrap_or(0);
let lead_handoff_backlog = self
.handoff_backlog
.get(&self.root.id)
.copied()
.unwrap_or(0);
writeln!(f, "Backlog: {}", lead_handoff_backlog)?;
if self.descendants.is_empty() {
@@ -1424,7 +1447,8 @@ impl fmt::Display for TeamStatus {
writeln!(f, "Board:")?;
let mut lanes: BTreeMap<&'static str, Vec<&DelegatedSessionSummary>> = BTreeMap::new();
for summary in &self.descendants {
lanes.entry(session_state_label(&summary.session.state))
lanes
.entry(session_state_label(&summary.session.state))
.or_default()
.push(summary);
}
@@ -1502,18 +1526,11 @@ impl fmt::Display for CoordinationStatus {
}
if self.operator_escalation_required {
writeln!(
f,
"Operator escalation: chronic saturation is not clearing"
)?;
writeln!(f, "Operator escalation: chronic saturation is not clearing")?;
}
if let Some(cleared_at) = self.daemon_activity.chronic_saturation_cleared_at() {
writeln!(
f,
"Chronic saturation cleared: {}",
cleared_at.to_rfc3339()
)?;
writeln!(f, "Chronic saturation cleared: {}", cleared_at.to_rfc3339())?;
}
if let Some(stabilized_at) = stabilized {
@@ -1631,6 +1648,7 @@ mod tests {
default_agent: "claude".to_string(),
auto_dispatch_unread_handoffs: false,
auto_dispatch_limit_per_session: 5,
auto_create_worktrees: true,
auto_merge_ready_worktrees: false,
cost_budget_usd: 10.0,
token_budget: 500_000,
@@ -1685,14 +1703,7 @@ mod tests {
run_git(path, ["config", "user.email", "ecc-tests@example.com"])?;
fs::write(path.join("README.md"), "hello\n")?;
run_git(path, ["add", "README.md"])?;
run_git(
path,
[
"commit",
"-qm",
"init",
],
)?;
run_git(path, ["commit", "-qm", "init"])?;
Ok(())
}
@@ -1885,7 +1896,13 @@ mod tests {
assert!(log.contains("--session-id"));
assert!(log.contains("deadbeef"));
assert!(log.contains("resume previous task"));
assert!(log.contains(tempdir.path().join("resume-working-dir").to_string_lossy().as_ref()));
assert!(log.contains(
tempdir
.path()
.join("resume-working-dir")
.to_string_lossy()
.as_ref()
));
Ok(())
}
@@ -1920,14 +1937,20 @@ mod tests {
.clone()
.context("stopped session worktree missing")?
.path;
assert!(worktree_path.exists(), "worktree should still exist before cleanup");
assert!(
worktree_path.exists(),
"worktree should still exist before cleanup"
);
cleanup_session_worktree(&db, &session_id).await?;
let cleaned = db
.get_session(&session_id)?
.context("cleaned session should still exist")?;
assert!(cleaned.worktree.is_none(), "worktree metadata should be cleared");
assert!(
cleaned.worktree.is_none(),
"worktree metadata should be cleared"
);
assert!(!worktree_path.exists(), "worktree path should be removed");
Ok(())
@@ -2051,12 +2074,18 @@ mod tests {
assert_eq!(outcome.base_branch, worktree.base_branch);
assert!(outcome.cleaned_worktree);
assert!(!outcome.already_up_to_date);
assert_eq!(fs::read_to_string(repo_root.join("feature.txt"))?, "ready to merge\n");
assert_eq!(
fs::read_to_string(repo_root.join("feature.txt"))?,
"ready to merge\n"
);
let merged = db
.get_session(&outcome.session_id)?
.context("merged session should still exist")?;
assert!(merged.worktree.is_none(), "worktree metadata should be cleared");
assert!(
merged.worktree.is_none(),
"worktree metadata should be cleared"
);
assert!(!worktree.path.exists(), "worktree path should be removed");
let branch_output = StdCommand::new("git")
@@ -2065,7 +2094,9 @@ mod tests {
.args(["branch", "--list", &worktree.branch])
.output()?;
assert!(
String::from_utf8_lossy(&branch_output.stdout).trim().is_empty(),
String::from_utf8_lossy(&branch_output.stdout)
.trim()
.is_empty(),
"merged worktree branch should be deleted"
);
@@ -2136,8 +2167,14 @@ mod tests {
assert_eq!(outcome.merged.len(), 1);
assert_eq!(outcome.merged[0].session_id, "merge-ready");
assert_eq!(outcome.active_with_worktree_ids, vec!["active-worktree".to_string()]);
assert_eq!(outcome.dirty_worktree_ids, vec!["dirty-worktree".to_string()]);
assert_eq!(
outcome.active_with_worktree_ids,
vec!["active-worktree".to_string()]
);
assert_eq!(
outcome.dirty_worktree_ids,
vec!["dirty-worktree".to_string()]
);
assert!(outcome.conflicted_session_ids.is_empty());
assert!(outcome.failures.is_empty());
@@ -2145,24 +2182,21 @@ mod tests {
fs::read_to_string(repo_root.join("merged.txt"))?,
"bulk merge\n"
);
assert!(
db.get_session("merge-ready")?
.context("merged session should still exist")?
.worktree
.is_none()
);
assert!(
db.get_session("active-worktree")?
.context("active session should still exist")?
.worktree
.is_some()
);
assert!(
db.get_session("dirty-worktree")?
.context("dirty session should still exist")?
.worktree
.is_some()
);
assert!(db
.get_session("merge-ready")?
.context("merged session should still exist")?
.worktree
.is_none());
assert!(db
.get_session("active-worktree")?
.context("active session should still exist")?
.worktree
.is_some());
assert!(db
.get_session("dirty-worktree")?
.context("dirty session should still exist")?
.worktree
.is_some());
assert!(!merged_worktree.path.exists());
assert!(active_worktree.path.exists());
assert!(dirty_worktree.path.exists());
@@ -2203,7 +2237,10 @@ mod tests {
delete_session(&db, &session_id).await?;
assert!(db.get_session(&session_id)?.is_none(), "session should be deleted");
assert!(
db.get_session(&session_id)?.is_none(),
"session should be deleted"
);
assert!(!worktree_path.exists(), "worktree path should be removed");
Ok(())
@@ -2233,8 +2270,16 @@ mod tests {
let db = StateStore::open(&cfg.db_path)?;
let now = Utc::now();
db.insert_session(&build_session("parent", SessionState::Running, now - Duration::minutes(2)))?;
db.insert_session(&build_session("child", SessionState::Pending, now - Duration::minutes(1)))?;
db.insert_session(&build_session(
"parent",
SessionState::Running,
now - Duration::minutes(2),
))?;
db.insert_session(&build_session(
"child",
SessionState::Pending,
now - Duration::minutes(1),
))?;
db.insert_session(&build_session("sibling", SessionState::Idle, now))?;
db.send_message(
@@ -2270,9 +2315,21 @@ mod tests {
let db = StateStore::open(&tempdir.path().join("state.db"))?;
let now = Utc::now();
db.insert_session(&build_session("lead", SessionState::Running, now - Duration::minutes(3)))?;
db.insert_session(&build_session("worker-a", SessionState::Running, now - Duration::minutes(2)))?;
db.insert_session(&build_session("worker-b", SessionState::Pending, now - Duration::minutes(1)))?;
db.insert_session(&build_session(
"lead",
SessionState::Running,
now - Duration::minutes(3),
))?;
db.insert_session(&build_session(
"worker-a",
SessionState::Running,
now - Duration::minutes(2),
))?;
db.insert_session(&build_session(
"worker-b",
SessionState::Pending,
now - Duration::minutes(1),
))?;
db.insert_session(&build_session("reviewer", SessionState::Completed, now))?;
db.send_message(
@@ -2444,15 +2501,15 @@ mod tests {
let spawned_messages = db.list_messages_for_session(&outcome.session_id, 10)?;
assert!(spawned_messages.iter().any(|message| {
message.msg_type == "task_handoff"
&& message.content.contains("Fresh delegated task")
message.msg_type == "task_handoff" && message.content.contains("Fresh delegated task")
}));
Ok(())
}
#[tokio::test(flavor = "current_thread")]
async fn assign_session_reuses_idle_delegate_when_only_non_handoff_messages_are_unread() -> Result<()> {
async fn assign_session_reuses_idle_delegate_when_only_non_handoff_messages_are_unread(
) -> Result<()> {
let tempdir = TestDir::new("manager-assign-reuse-idle-info-inbox")?;
let repo_root = tempdir.path().join("repo");
init_git_repo(&repo_root)?;
@@ -2512,8 +2569,7 @@ mod tests {
let idle_messages = db.list_messages_for_session("idle-worker", 10)?;
assert!(idle_messages.iter().any(|message| {
message.msg_type == "task_handoff"
&& message.content.contains("Fresh delegated task")
message.msg_type == "task_handoff" && message.content.contains("Fresh delegated task")
}));
Ok(())
@@ -2583,8 +2639,7 @@ mod tests {
let messages = db.list_messages_for_session(&outcome.session_id, 10)?;
assert!(messages.iter().any(|message| {
message.msg_type == "task_handoff"
&& message.content.contains("New delegated task")
message.msg_type == "task_handoff" && message.content.contains("New delegated task")
}));
Ok(())
@@ -2650,8 +2705,7 @@ mod tests {
let busy_messages = db.list_messages_for_session("busy-worker", 10)?;
assert!(!busy_messages.iter().any(|message| {
message.msg_type == "task_handoff"
&& message.content.contains("New delegated task")
message.msg_type == "task_handoff" && message.content.contains("New delegated task")
}));
Ok(())
@@ -2697,8 +2751,7 @@ mod tests {
let messages = db.list_messages_for_session(&outcomes[0].session_id, 10)?;
assert!(messages.iter().any(|message| {
message.msg_type == "task_handoff"
&& message.content.contains("Review auth changes")
message.msg_type == "task_handoff" && message.content.contains("Review auth changes")
}));
Ok(())
@@ -2764,8 +2817,7 @@ mod tests {
let messages = db.list_messages_for_session("busy-worker", 10)?;
assert!(!messages.iter().any(|message| {
message.msg_type == "task_handoff"
&& message.content.contains("Review auth changes")
message.msg_type == "task_handoff" && message.content.contains("Review auth changes")
}));
Ok(())
@@ -3030,8 +3082,7 @@ mod tests {
let worker_b_messages = db.list_messages_for_session("worker-b", 10)?;
assert!(worker_b_messages.iter().any(|message| {
message.msg_type == "task_handoff"
&& message.content.contains("Review auth flow")
message.msg_type == "task_handoff" && message.content.contains("Review auth flow")
}));
Ok(())
@@ -3108,22 +3159,18 @@ mod tests {
};
let rendered = status.to_string();
assert!(
rendered.contains(
"Global handoff backlog: 2 lead(s) / 5 handoff(s) [1 absorbable, 1 saturated]"
)
);
assert!(rendered.contains(
"Global handoff backlog: 2 lead(s) / 5 handoff(s) [1 absorbable, 1 saturated]"
));
assert!(rendered.contains("Auto-dispatch: on @ 4/lead"));
assert!(rendered.contains("Coordination mode: rebalance-first (chronic saturation)"));
assert!(rendered.contains("Chronic saturation streak: 2 cycle(s)"));
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"
)
);
assert!(rendered.contains(
"Last daemon auto-merge: 1 merged / 1 active / 0 conflicted / 0 dirty / 0 failed"
));
}
#[test]
@@ -3174,7 +3221,10 @@ mod tests {
assert_eq!(status.backlog_messages, 3);
assert_eq!(status.absorbable_sessions, 2);
assert_eq!(status.saturated_sessions, 1);
assert_eq!(status.mode, CoordinationMode::RebalanceFirstChronicSaturation);
assert_eq!(
status.mode,
CoordinationMode::RebalanceFirstChronicSaturation
);
assert_eq!(status.health, CoordinationHealth::Saturated);
assert!(!status.operator_escalation_required);
assert_eq!(status.daemon_activity.last_dispatch_routed, 1);

18
ecc2/src/session/runtime.rs
View File

@@ -70,11 +70,7 @@ impl DbWriter {
}
}
fn run_db_writer(
db_path: PathBuf,
session_id: String,
mut rx: mpsc::UnboundedReceiver<DbMessage>,
) {
fn run_db_writer(db_path: PathBuf, session_id: String, mut rx: mpsc::UnboundedReceiver<DbMessage>) {
let (opened, open_error) = match StateStore::open(&db_path) {
Ok(db) => (Some(db), None),
Err(error) => (None, Some(error.to_string())),
@@ -84,7 +80,9 @@ fn run_db_writer(
match message {
DbMessage::UpdateState { state, ack } => {
let result = match opened.as_ref() {
Some(db) => db.update_state(&session_id, &state).map_err(|error| error.to_string()),
Some(db) => db
.update_state(&session_id, &state)
.map_err(|error| error.to_string()),
None => Err(open_error
.clone()
.unwrap_or_else(|| "Failed to open state store".to_string())),
@@ -93,7 +91,9 @@ fn run_db_writer(
}
DbMessage::UpdatePid { pid, ack } => {
let result = match opened.as_ref() {
Some(db) => db.update_pid(&session_id, pid).map_err(|error| error.to_string()),
Some(db) => db
.update_pid(&session_id, pid)
.map_err(|error| error.to_string()),
None => Err(open_error
.clone()
.unwrap_or_else(|| "Failed to open state store".to_string())),
@@ -205,9 +205,7 @@ where
let mut lines = BufReader::new(reader).lines();
while let Some(line) = lines.next_line().await? {
db_writer
.append_output_line(stream, line.clone())
.await?;
db_writer.append_output_line(stream, line.clone()).await?;
output_store.push_line(&session_id, stream, line);
}

1
ecc2/src/tui/app.rs
View File

@@ -50,6 +50,7 @@ pub async fn run(db: StateStore, cfg: Config) -> Result<()> {
(_, KeyCode::Char('m')) => dashboard.merge_selected_worktree().await,
(_, KeyCode::Char('M')) => dashboard.merge_ready_worktrees().await,
(_, KeyCode::Char('p')) => dashboard.toggle_auto_dispatch_policy(),
(_, KeyCode::Char('t')) => dashboard.toggle_auto_worktree_policy(),
(_, KeyCode::Char('w')) => dashboard.toggle_auto_merge_policy(),
(_, KeyCode::Char(',')) => dashboard.adjust_auto_dispatch_limit(-1),
(_, KeyCode::Char('.')) => dashboard.adjust_auto_dispatch_limit(1),

323
ecc2/src/tui/dashboard.rs
View File

@@ -7,12 +7,12 @@ use ratatui::{
use std::collections::HashMap;
use tokio::sync::broadcast;
use super::widgets::{BudgetState, TokenMeter, budget_state, format_currency, format_token_count};
use super::widgets::{budget_state, format_currency, format_token_count, BudgetState, TokenMeter};
use crate::comms;
use crate::config::{Config, PaneLayout};
use crate::observability::ToolLogEntry;
use crate::session::manager;
use crate::session::output::{OUTPUT_BUFFER_LIMIT, OutputEvent, OutputLine, SessionOutputStore};
use crate::session::output::{OutputEvent, OutputLine, SessionOutputStore, OUTPUT_BUFFER_LIMIT};
use crate::session::store::{DaemonActivity, StateStore};
use crate::session::{Session, SessionMessage, SessionState};
use crate::worktree;
@@ -362,7 +362,11 @@ impl Dashboard {
let content = if lines.is_empty() {
"Waiting for session output...".to_string()
} else {
lines.iter().map(|line| line.text.as_str()).collect::<Vec<_>>().join("\n")
lines
.iter()
.map(|line| line.text.as_str())
.collect::<Vec<_>>()
.join("\n")
};
(" Output ", content)
}
@@ -383,18 +387,17 @@ impl Dashboard {
(" Diff ", content)
}
OutputMode::ConflictProtocol => {
let content = self
.selected_conflict_protocol
.clone()
.unwrap_or_else(|| {
"No conflicted worktree available for the selected session."
.to_string()
});
let content = self.selected_conflict_protocol.clone().unwrap_or_else(|| {
"No conflicted worktree available for the selected session.".to_string()
});
(" Conflict Protocol ", content)
}
}
} else {
(" Output ", "No sessions. Press 'n' to start one.".to_string())
(
" Output ",
"No sessions. Press 'n' to start one.".to_string(),
)
};
let paragraph = Paragraph::new(content)
@@ -523,7 +526,7 @@ impl Dashboard {
fn render_status_bar(&self, frame: &mut Frame, area: Rect) {
let text = format!(
" [n]ew session [a]ssign re[b]alance global re[B]alance dra[i]n inbox [g]lobal dispatch coordinate [G]lobal [v]iew diff conflict proto[c]ol [m]erge merge ready [M] auto-merge [w] toggle [p]olicy [,/.] dispatch limit [s]top [u]resume [x]cleanup prune inactive [X] [d]elete [r]efresh [Tab] switch pane [j/k] scroll [+/-] resize [{}] layout [?] help [q]uit ",
" [n]ew session [a]ssign re[b]alance global re[B]alance dra[i]n inbox [g]lobal dispatch coordinate [G]lobal [v]iew diff conflict proto[c]ol [m]erge merge ready [M] auto-worktree [t] auto-merge [w] toggle [p]olicy [,/.] dispatch limit [s]top [u]resume [x]cleanup prune inactive [X] [d]elete [r]efresh [Tab] switch pane [j/k] scroll [+/-] resize [{}] layout [?] help [q]uit ",
self.layout_label()
);
let text = if let Some(note) = self.operator_note.as_ref() {
@@ -578,6 +581,7 @@ impl Dashboard {
" c Show conflict-resolution protocol for selected conflicted worktree",
" m Merge selected ready worktree into base and clean it up",
" M Merge all ready inactive worktrees and clean them up",
" t Toggle default worktree creation for new sessions and delegated work",
" p Toggle daemon auto-dispatch policy and persist config",
" w Toggle daemon auto-merge for ready inactive worktrees",
" ,/. Decrease/increase auto-dispatch limit per lead",
@@ -714,15 +718,22 @@ impl Dashboard {
let task = self.new_session_task();
let agent = self.cfg.default_agent.clone();
let session_id =
match manager::create_session(&self.db, &self.cfg, &task, &agent, true).await {
Ok(session_id) => session_id,
Err(error) => {
tracing::warn!("Failed to create new session from dashboard: {error}");
self.set_operator_note(format!("new session failed: {error}"));
return;
}
};
let session_id = match manager::create_session(
&self.db,
&self.cfg,
&task,
&agent,
self.cfg.auto_create_worktrees,
)
.await
{
Ok(session_id) => session_id,
Err(error) => {
tracing::warn!("Failed to create new session from dashboard: {error}");
self.set_operator_note(format!("new session failed: {error}"));
return;
}
};
if let Some(source_session) = self.sessions.get(self.selected_session) {
let context = format!(
@@ -834,7 +845,7 @@ impl Dashboard {
&source_session.id,
&task,
&agent,
true,
self.cfg.auto_create_worktrees,
)
.await
{
@@ -876,7 +887,7 @@ impl Dashboard {
&self.cfg,
&source_session_id,
&agent,
true,
self.cfg.auto_create_worktrees,
self.cfg.auto_dispatch_limit_per_session,
)
.await
@@ -930,7 +941,7 @@ impl Dashboard {
&self.cfg,
&source_session_id,
&agent,
true,
self.cfg.auto_create_worktrees,
self.cfg.max_parallel_sessions,
)
.await
@@ -975,17 +986,22 @@ impl Dashboard {
let agent = self.cfg.default_agent.clone();
let lead_limit = self.sessions.len().max(1);
let outcomes =
match manager::auto_dispatch_backlog(&self.db, &self.cfg, &agent, true, lead_limit)
.await
{
Ok(outcomes) => outcomes,
Err(error) => {
tracing::warn!("Failed to auto-dispatch backlog from dashboard: {error}");
self.set_operator_note(format!("global auto-dispatch failed: {error}"));
return;
}
};
let outcomes = match manager::auto_dispatch_backlog(
&self.db,
&self.cfg,
&agent,
self.cfg.auto_create_worktrees,
lead_limit,
)
.await
{
Ok(outcomes) => outcomes,
Err(error) => {
tracing::warn!("Failed to auto-dispatch backlog from dashboard: {error}");
self.set_operator_note(format!("global auto-dispatch failed: {error}"));
return;
}
};
let total_processed: usize = outcomes.iter().map(|outcome| outcome.routed.len()).sum();
let total_routed: usize = outcomes
@@ -1029,16 +1045,22 @@ impl Dashboard {
let agent = self.cfg.default_agent.clone();
let lead_limit = self.sessions.len().max(1);
let outcomes =
match manager::rebalance_all_teams(&self.db, &self.cfg, &agent, true, lead_limit).await
{
Ok(outcomes) => outcomes,
Err(error) => {
tracing::warn!("Failed to rebalance teams from dashboard: {error}");
self.set_operator_note(format!("global rebalance failed: {error}"));
return;
}
};
let outcomes = match manager::rebalance_all_teams(
&self.db,
&self.cfg,
&agent,
self.cfg.auto_create_worktrees,
lead_limit,
)
.await
{
Ok(outcomes) => outcomes,
Err(error) => {
tracing::warn!("Failed to rebalance teams from dashboard: {error}");
self.set_operator_note(format!("global rebalance failed: {error}"));
return;
}
};
let total_rerouted: usize = outcomes.iter().map(|outcome| outcome.rerouted.len()).sum();
let selected_session_id = self
@@ -1070,7 +1092,11 @@ impl Dashboard {
let lead_limit = self.sessions.len().max(1);
let outcome = match manager::coordinate_backlog(
&self.db, &self.cfg, &agent, true, lead_limit,
&self.db,
&self.cfg,
&agent,
self.cfg.auto_create_worktrees,
lead_limit,
)
.await
{
@@ -1386,6 +1412,29 @@ impl Dashboard {
}
}
pub fn toggle_auto_worktree_policy(&mut self) {
self.cfg.auto_create_worktrees = !self.cfg.auto_create_worktrees;
match self.cfg.save() {
Ok(()) => {
let state = if self.cfg.auto_create_worktrees {
"enabled"
} else {
"disabled"
};
self.set_operator_note(format!(
"default worktree creation {state} | saved to {}",
crate::config::Config::config_path().display()
));
}
Err(error) => {
self.cfg.auto_create_worktrees = !self.cfg.auto_create_worktrees;
self.set_operator_note(format!(
"failed to persist worktree creation policy: {error}"
));
}
}
}
pub fn adjust_auto_dispatch_limit(&mut self, delta: isize) {
let next =
(self.cfg.auto_dispatch_limit_per_session as isize + delta).clamp(1, 50) as usize;
@@ -1571,10 +1620,13 @@ impl Dashboard {
self.selected_diff_preview = worktree
.and_then(|worktree| worktree::diff_file_preview(worktree, MAX_DIFF_PREVIEW_LINES).ok())
.unwrap_or_default();
self.selected_diff_patch = worktree
.and_then(|worktree| worktree::diff_patch_preview(worktree, MAX_DIFF_PATCH_LINES).ok().flatten());
self.selected_merge_readiness = worktree
.and_then(|worktree| worktree::merge_readiness(worktree).ok());
self.selected_diff_patch = worktree.and_then(|worktree| {
worktree::diff_patch_preview(worktree, MAX_DIFF_PATCH_LINES)
.ok()
.flatten()
});
self.selected_merge_readiness =
worktree.and_then(|worktree| worktree::merge_readiness(worktree).ok());
self.selected_conflict_protocol = session
.zip(worktree)
.zip(self.selected_merge_readiness.as_ref())
@@ -1870,7 +1922,7 @@ impl Dashboard {
}
lines.push(format!(
"Global handoff backlog {} lead(s) / {} handoff(s) | Auto-dispatch {} @ {}/lead | Auto-merge {}",
"Global handoff backlog {} lead(s) / {} handoff(s) | Auto-dispatch {} @ {}/lead | Auto-worktree {} | Auto-merge {}",
self.global_handoff_backlog_leads,
self.global_handoff_backlog_messages,
if self.cfg.auto_dispatch_unread_handoffs {
@@ -1879,6 +1931,11 @@ impl Dashboard {
"off"
},
self.cfg.auto_dispatch_limit_per_session,
if self.cfg.auto_create_worktrees {
"on"
} else {
"off"
},
if self.cfg.auto_merge_ready_worktrees {
"on"
} else {
@@ -2099,10 +2156,7 @@ impl Dashboard {
.is_some();
for session in &self.sessions {
if self
.worktree_health_by_session
.get(&session.id)
.copied()
if self.worktree_health_by_session.get(&session.id).copied()
== Some(worktree::WorktreeHealth::Conflicted)
{
items.push(format!(
@@ -2283,7 +2337,11 @@ impl Dashboard {
fn log_field<'a>(&self, value: &'a str) -> &'a str {
let trimmed = value.trim();
if trimmed.is_empty() { "n/a" } else { trimmed }
if trimmed.is_empty() {
"n/a"
} else {
trimmed
}
}
fn short_timestamp(&self, timestamp: &str) -> String {
@@ -2423,11 +2481,19 @@ fn summary_line(summary: &SessionSummary) -> Line<'static> {
];
if summary.conflicted_worktrees > 0 {
spans.push(summary_span("Conflicts", summary.conflicted_worktrees, Color::Red));
spans.push(summary_span(
"Conflicts",
summary.conflicted_worktrees,
Color::Red,
));
}
if summary.in_progress_worktrees > 0 {
spans.push(summary_span("Worktrees", summary.in_progress_worktrees, Color::Cyan));
spans.push(summary_span(
"Worktrees",
summary.in_progress_worktrees,
Color::Cyan,
));
}
Line::from(spans)
@@ -2462,17 +2528,19 @@ fn attention_queue_line(summary: &SessionSummary, stabilized: bool) -> Line<'sta
]);
}
let mut spans = vec![
Span::styled(
"Attention queue ",
Style::default()
.fg(Color::Yellow)
.add_modifier(Modifier::BOLD),
),
];
let mut spans = vec![Span::styled(
"Attention queue ",
Style::default()
.fg(Color::Yellow)
.add_modifier(Modifier::BOLD),
)];
if summary.conflicted_worktrees > 0 {
spans.push(summary_span("Conflicts", summary.conflicted_worktrees, Color::Red));
spans.push(summary_span(
"Conflicts",
summary.conflicted_worktrees,
Color::Red,
));
}
spans.extend([
@@ -2606,11 +2674,16 @@ fn build_conflict_protocol(
));
lines.push(format!("2. Open worktree: cd {}", worktree.path.display()));
lines.push("3. Resolve conflicts and stage files: git add <paths>".to_string());
lines.push(format!("4. Commit the resolution on {}: git commit", worktree.branch));
lines.push(format!(
"4. Commit the resolution on {}: git commit",
worktree.branch
));
lines.push(format!(
"5. Re-check readiness: ecc worktree-status {session_id} --check"
));
lines.push(format!("6. Merge when clear: ecc merge-worktree {session_id}"));
lines.push(format!(
"6. Merge when clear: ecc merge-worktree {session_id}"
));
Some(lines.join("\n"))
}
@@ -2643,7 +2716,7 @@ fn format_duration(duration_secs: u64) -> String {
mod tests {
use anyhow::{Context, Result};
use chrono::Utc;
use ratatui::{Terminal, backend::TestBackend};
use ratatui::{backend::TestBackend, Terminal};
use std::fs;
use std::path::{Path, PathBuf};
use std::process::Command;
@@ -2862,14 +2935,14 @@ diff --git a/src/next.rs b/src/next.rs
let text = dashboard.selected_session_metrics_text();
assert!(text.contains("Team 3/8 | idle 1 | running 1 | pending 1 | failed 0 | stopped 0"));
assert!(text.contains(
"Global handoff backlog 2 lead(s) / 5 handoff(s) | Auto-dispatch off @ 5/lead | Auto-merge off"
"Global handoff backlog 2 lead(s) / 5 handoff(s) | Auto-dispatch off @ 5/lead | Auto-worktree on | Auto-merge off"
));
assert!(text.contains("Coordination mode dispatch-first"));
assert!(text.contains("Next route reuse idle worker-1"));
}
#[test]
fn selected_session_metrics_text_shows_auto_merge_policy_state() {
fn selected_session_metrics_text_shows_worktree_and_auto_merge_policy_state() {
let mut dashboard = test_dashboard(
vec![sample_session(
"focus-12345678",
@@ -2882,16 +2955,60 @@ diff --git a/src/next.rs b/src/next.rs
0,
);
dashboard.cfg.auto_dispatch_unread_handoffs = true;
dashboard.cfg.auto_create_worktrees = false;
dashboard.cfg.auto_merge_ready_worktrees = true;
dashboard.global_handoff_backlog_leads = 1;
dashboard.global_handoff_backlog_messages = 2;
let text = dashboard.selected_session_metrics_text();
assert!(text.contains(
"Global handoff backlog 1 lead(s) / 2 handoff(s) | Auto-dispatch on @ 5/lead | Auto-merge on"
"Global handoff backlog 1 lead(s) / 2 handoff(s) | Auto-dispatch on @ 5/lead | Auto-worktree off | Auto-merge on"
));
}
#[test]
fn toggle_auto_worktree_policy_persists_config() {
let tempdir = std::env::temp_dir().join(format!("ecc2-worktree-policy-{}", Uuid::new_v4()));
std::fs::create_dir_all(&tempdir).unwrap();
let previous_home = std::env::var_os("HOME");
std::env::set_var("HOME", &tempdir);
let mut dashboard = test_dashboard(
vec![sample_session(
"focus-12345678",
"planner",
SessionState::Running,
Some("ecc/focus"),
512,
42,
)],
0,
);
dashboard.cfg.auto_create_worktrees = true;
dashboard.toggle_auto_worktree_policy();
assert!(!dashboard.cfg.auto_create_worktrees);
let expected_note = format!(
"default worktree creation disabled | saved to {}",
crate::config::Config::config_path().display()
);
assert_eq!(
dashboard.operator_note.as_deref(),
Some(expected_note.as_str())
);
let saved = std::fs::read_to_string(crate::config::Config::config_path()).unwrap();
assert!(saved.contains("auto_create_worktrees = false"));
if let Some(home) = previous_home {
std::env::set_var("HOME", home);
} else {
std::env::remove_var("HOME");
}
let _ = std::fs::remove_dir_all(tempdir);
}
#[test]
fn selected_session_metrics_text_includes_daemon_activity() {
let now = Utc::now();
@@ -2932,9 +3049,9 @@ diff --git a/src/next.rs b/src/next.rs
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")
);
assert!(text.contains(
"Last daemon auto-merge 2 merged / 1 active / 1 conflicted / 0 dirty / 0 failed"
));
}
#[test]
@@ -3013,8 +3130,8 @@ diff --git a/src/next.rs b/src/next.rs
}
#[test]
fn selected_session_metrics_text_recommends_operator_escalation_when_chronic_saturation_is_stuck()
{
fn selected_session_metrics_text_recommends_operator_escalation_when_chronic_saturation_is_stuck(
) {
let mut dashboard = test_dashboard(
vec![sample_session(
"focus-12345678",
@@ -3664,18 +3781,16 @@ diff --git a/src/next.rs b/src/next.rs
dashboard.operator_note.as_deref(),
Some("pruned 1 inactive worktree(s); skipped 1 active session(s)")
);
assert!(
db.get_session("stopped-1")?
.expect("stopped session should exist")
.worktree
.is_none()
);
assert!(
db.get_session("running-1")?
.expect("running session should exist")
.worktree
.is_some()
);
assert!(db
.get_session("stopped-1")?
.expect("stopped session should exist")
.worktree
.is_none());
assert!(db
.get_session("running-1")?
.expect("running session should exist")
.worktree
.is_some());
let _ = std::fs::remove_dir_all(active_path);
let _ = std::fs::remove_dir_all(stopped_path);
@@ -3734,7 +3849,10 @@ diff --git a/src/next.rs b/src/next.rs
.db
.get_session(&session_id)?
.context("merged session should still exist")?;
assert!(session.worktree.is_none(), "worktree metadata should be cleared");
assert!(
session.worktree.is_none(),
"worktree metadata should be cleared"
);
assert!(!worktree.path.exists(), "worktree path should be removed");
assert_eq!(
std::fs::read_to_string(repo_root.join("dashboard.txt"))?,
@@ -3756,8 +3874,12 @@ diff --git a/src/next.rs b/src/next.rs
let db = StateStore::open(&cfg.db_path)?;
let now = Utc::now();
let merged_worktree = worktree::create_for_session_in_repo("merge-ready", &cfg, &repo_root)?;
std::fs::write(merged_worktree.path.join("merged.txt"), "dashboard bulk merge\n")?;
let merged_worktree =
worktree::create_for_session_in_repo("merge-ready", &cfg, &repo_root)?;
std::fs::write(
merged_worktree.path.join("merged.txt"),
"dashboard bulk merge\n",
)?;
Command::new("git")
.arg("-C")
.arg(&merged_worktree.path)
@@ -3805,14 +3927,12 @@ diff --git a/src/next.rs b/src/next.rs
.context("operator note should be set")?;
assert!(note.contains("merged 1 ready worktree(s)"));
assert!(note.contains("skipped 1 active"));
assert!(
dashboard
.db
.get_session("merge-ready")?
.context("merged session should still exist")?
.worktree
.is_none()
);
assert!(dashboard
.db
.get_session("merge-ready")?
.context("merged session should still exist")?
.worktree
.is_none());
assert_eq!(
std::fs::read_to_string(repo_root.join("merged.txt"))?,
"dashboard bulk merge\n"
@@ -4100,6 +4220,7 @@ diff --git a/src/next.rs b/src/next.rs
default_agent: "claude".to_string(),
auto_dispatch_unread_handoffs: false,
auto_dispatch_limit_per_session: 5,
auto_create_worktrees: true,
auto_merge_ready_worktrees: false,
cost_budget_usd: 10.0,
token_budget: 500_000,

32
ecc2/src/worktree/mod.rs
View File

@@ -237,7 +237,12 @@ pub fn merge_readiness(worktree: &WorktreeInfo) -> Result<MergeReadiness> {
let output = Command::new("git")
.arg("-C")
.arg(&worktree.path)
.args(["merge-tree", "--write-tree", &worktree.base_branch, &worktree.branch])
.args([
"merge-tree",
"--write-tree",
&worktree.base_branch,
&worktree.branch,
])
.output()
.context("Failed to generate merge readiness preview")?;
@@ -519,7 +524,8 @@ fn base_checkout_path(worktree: &WorktreeInfo) -> Result<PathBuf> {
if fallback.is_none() && path != worktree.path {
fallback = Some(path.clone());
}
if current_branch.as_deref() == Some(target_branch.as_str()) && path != worktree.path
if current_branch.as_deref() == Some(target_branch.as_str())
&& path != worktree.path
{
return Ok(path);
}
@@ -660,16 +666,12 @@ mod tests {
};
let preview = diff_file_preview(&info, 6)?;
assert!(
preview
.iter()
.any(|line| line.contains("Branch A") && line.contains("src.txt"))
);
assert!(
preview
.iter()
.any(|line| line.contains("Working M") && line.contains("README.md"))
);
assert!(preview
.iter()
.any(|line| line.contains("Branch A") && line.contains("src.txt")));
assert!(preview
.iter()
.any(|line| line.contains("Working M") && line.contains("README.md")));
let _ = Command::new("git")
.arg("-C")
@@ -736,7 +738,8 @@ mod tests {
#[test]
fn merge_readiness_reports_ready_worktree() -> Result<()> {
let root = std::env::temp_dir().join(format!("ecc2-worktree-merge-ready-{}", Uuid::new_v4()));
let root =
std::env::temp_dir().join(format!("ecc2-worktree-merge-ready-{}", Uuid::new_v4()));
let repo = root.join("repo");
fs::create_dir_all(&repo)?;
@@ -787,7 +790,8 @@ mod tests {
#[test]
fn merge_readiness_reports_conflicted_worktree() -> Result<()> {
let root = std::env::temp_dir().join(format!("ecc2-worktree-merge-conflict-{}", Uuid::new_v4()));
let root =
std::env::temp_dir().join(format!("ecc2-worktree-merge-conflict-{}", Uuid::new_v4()));
let repo = root.join("repo");
fs::create_dir_all(&repo)?;