crashlabs/huskies
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 26 Wiki Activity

storkit: merge 418_refactor_split_pool_auto_assign_rs_into_submodules

Browse Source
This commit is contained in:
dave
2026-03-28 08:44:27 +00:00
parent 4ea4be1462
commit d375c4b1d3
7 changed files with 1901 additions and 1813 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/src/agents/pool/auto_assign/reconcile.rs
+527
View File
@@ -0,0 +1,527 @@
//! Startup reconciliation: detect stories with committed work and advance the pipeline.
use std::path::Path;
use tokio::sync::broadcast;
use crate::worktree;
use super::super::super::ReconciliationEvent;
use super::super::{AgentPool, find_active_story_stage};
impl AgentPool {
/// Reconcile stories whose agent work was committed while the server was offline.
///
/// On server startup the in-memory agent pool is empty, so any story that an agent
/// completed during a previous session is stuck: the worktree has committed work but
/// the pipeline never advanced. This method detects those stories, re-runs the
/// acceptance gates, and advances the pipeline stage so that `auto_assign_available_work`
/// (called immediately after) picks up the right next-stage agents.
///
/// Algorithm:
/// 1. List all worktree directories under `{project_root}/.storkit/worktrees/`.
/// 2. For each worktree, check whether its feature branch has commits ahead of the
/// base branch (`master` / `main`).
/// 3. If committed work is found AND the story is in `2_current/` or `3_qa/`:
/// - Run acceptance gates (uncommitted-change check + clippy + tests).
/// - On pass + `2_current/`: move the story to `3_qa/`.
/// - On pass + `3_qa/`: run the coverage gate; if that also passes move to `4_merge/`.
/// - On failure: leave the story where it is so `auto_assign_available_work` can
/// start a fresh agent to retry.
/// 4. Stories in `4_merge/` are left for `auto_assign_available_work` to handle via a
/// fresh mergemaster (squash-merge must be re-executed by the mergemaster agent).
pub async fn reconcile_on_startup(
&self,
project_root: &Path,
progress_tx: &broadcast::Sender<ReconciliationEvent>,
) {
let worktrees = match worktree::list_worktrees(project_root) {
Ok(wt) => wt,
Err(e) => {
eprintln!("[startup:reconcile] Failed to list worktrees: {e}");
let _ = progress_tx.send(ReconciliationEvent {
story_id: String::new(),
status: "done".to_string(),
message: format!("Reconciliation failed: {e}"),
});
return;
}
};
for wt_entry in &worktrees {
let story_id = &wt_entry.story_id;
let wt_path = wt_entry.path.clone();
// Determine which active stage the story is in.
let stage_dir = match find_active_story_stage(project_root, story_id) {
Some(s) => s,
None => continue, // Not in any active stage (backlog/archived or unknown).
};
// 4_merge/ is left for auto_assign to handle with a fresh mergemaster.
if stage_dir == "4_merge" {
continue;
}
let _ = progress_tx.send(ReconciliationEvent {
story_id: story_id.clone(),
status: "checking".to_string(),
message: format!("Checking for committed work in {stage_dir}/"),
});
// Check whether the worktree has commits ahead of the base branch.
let wt_path_for_check = wt_path.clone();
let has_work = tokio::task::spawn_blocking(move || {
crate::agents::gates::worktree_has_committed_work(&wt_path_for_check)
})
.await
.unwrap_or(false);
if !has_work {
eprintln!(
"[startup:reconcile] No committed work for '{story_id}' in {stage_dir}/; skipping."
);
let _ = progress_tx.send(ReconciliationEvent {
story_id: story_id.clone(),
status: "skipped".to_string(),
message: "No committed work found; skipping.".to_string(),
});
continue;
}
eprintln!(
"[startup:reconcile] Found committed work for '{story_id}' in {stage_dir}/. Running acceptance gates."
);
let _ = progress_tx.send(ReconciliationEvent {
story_id: story_id.clone(),
status: "gates_running".to_string(),
message: "Running acceptance gates…".to_string(),
});
// Run acceptance gates on the worktree.
let wt_path_for_gates = wt_path.clone();
let gates_result = tokio::task::spawn_blocking(move || {
crate::agents::gates::check_uncommitted_changes(&wt_path_for_gates)?;
crate::agents::gates::run_acceptance_gates(&wt_path_for_gates)
})
.await;
let (gates_passed, gate_output) = match gates_result {
Ok(Ok(pair)) => pair,
Ok(Err(e)) => {
eprintln!("[startup:reconcile] Gate check error for '{story_id}': {e}");
let _ = progress_tx.send(ReconciliationEvent {
story_id: story_id.clone(),
status: "failed".to_string(),
message: format!("Gate error: {e}"),
});
continue;
}
Err(e) => {
eprintln!("[startup:reconcile] Gate check task panicked for '{story_id}': {e}");
let _ = progress_tx.send(ReconciliationEvent {
story_id: story_id.clone(),
status: "failed".to_string(),
message: format!("Gate task panicked: {e}"),
});
continue;
}
};
if !gates_passed {
eprintln!(
"[startup:reconcile] Gates failed for '{story_id}': {gate_output}\n\
Leaving in {stage_dir}/ for auto-assign to restart the agent."
);
let _ = progress_tx.send(ReconciliationEvent {
story_id: story_id.clone(),
status: "failed".to_string(),
message: "Gates failed; will be retried by auto-assign.".to_string(),
});
continue;
}
eprintln!("[startup:reconcile] Gates passed for '{story_id}' (stage: {stage_dir}/).");
if stage_dir == "2_current" {
// Coder stage — determine qa mode to decide next step.
let qa_mode = {
let item_type = crate::agents::lifecycle::item_type_from_id(story_id);
if item_type == "spike" {
crate::io::story_metadata::QaMode::Human
} else {
let default_qa = crate::config::ProjectConfig::load(project_root)
.unwrap_or_default()
.default_qa_mode();
let story_path = project_root
.join(".storkit/work/2_current")
.join(format!("{story_id}.md"));
crate::io::story_metadata::resolve_qa_mode(&story_path, default_qa)
}
};
match qa_mode {
crate::io::story_metadata::QaMode::Server => {
if let Err(e) =
crate::agents::move_story_to_merge(project_root, story_id)
{
eprintln!("[startup:reconcile] Failed to move '{story_id}' to 4_merge/: {e}");
let _ = progress_tx.send(ReconciliationEvent {
story_id: story_id.clone(),
status: "failed".to_string(),
message: format!("Failed to advance to merge: {e}"),
});
} else {
eprintln!("[startup:reconcile] Moved '{story_id}' → 4_merge/ (qa: server).");
let _ = progress_tx.send(ReconciliationEvent {
story_id: story_id.clone(),
status: "advanced".to_string(),
message: "Gates passed — moved to merge (qa: server).".to_string(),
});
}
}
crate::io::story_metadata::QaMode::Agent => {
if let Err(e) =
crate::agents::move_story_to_qa(project_root, story_id)
{
eprintln!("[startup:reconcile] Failed to move '{story_id}' to 3_qa/: {e}");
let _ = progress_tx.send(ReconciliationEvent {
story_id: story_id.clone(),
status: "failed".to_string(),
message: format!("Failed to advance to QA: {e}"),
});
} else {
eprintln!("[startup:reconcile] Moved '{story_id}' → 3_qa/.");
let _ = progress_tx.send(ReconciliationEvent {
story_id: story_id.clone(),
status: "advanced".to_string(),
message: "Gates passed — moved to QA.".to_string(),
});
}
}
crate::io::story_metadata::QaMode::Human => {
if let Err(e) =
crate::agents::move_story_to_qa(project_root, story_id)
{
eprintln!("[startup:reconcile] Failed to move '{story_id}' to 3_qa/: {e}");
let _ = progress_tx.send(ReconciliationEvent {
story_id: story_id.clone(),
status: "failed".to_string(),
message: format!("Failed to advance to QA: {e}"),
});
} else {
let story_path = project_root
.join(".storkit/work/3_qa")
.join(format!("{story_id}.md"));
if let Err(e) =
crate::io::story_metadata::write_review_hold(&story_path)
{
eprintln!(
"[startup:reconcile] Failed to set review_hold on '{story_id}': {e}"
);
}
eprintln!("[startup:reconcile] Moved '{story_id}' → 3_qa/ (qa: human — holding for review).");
let _ = progress_tx.send(ReconciliationEvent {
story_id: story_id.clone(),
status: "review_hold".to_string(),
message: "Gates passed — holding for human review.".to_string(),
});
}
}
}
} else if stage_dir == "3_qa" {
// QA stage → run coverage gate before advancing to merge.
let wt_path_for_cov = wt_path.clone();
let coverage_result = tokio::task::spawn_blocking(move || {
crate::agents::gates::run_coverage_gate(&wt_path_for_cov)
})
.await;
let (coverage_passed, coverage_output) = match coverage_result {
Ok(Ok(pair)) => pair,
Ok(Err(e)) => {
eprintln!("[startup:reconcile] Coverage gate error for '{story_id}': {e}");
let _ = progress_tx.send(ReconciliationEvent {
story_id: story_id.clone(),
status: "failed".to_string(),
message: format!("Coverage gate error: {e}"),
});
continue;
}
Err(e) => {
eprintln!(
"[startup:reconcile] Coverage gate panicked for '{story_id}': {e}"
);
let _ = progress_tx.send(ReconciliationEvent {
story_id: story_id.clone(),
status: "failed".to_string(),
message: format!("Coverage gate panicked: {e}"),
});
continue;
}
};
if coverage_passed {
// Check whether this item needs human review before merging.
let needs_human_review = {
let item_type = crate::agents::lifecycle::item_type_from_id(story_id);
if item_type == "spike" {
true
} else {
let story_path = project_root
.join(".storkit/work/3_qa")
.join(format!("{story_id}.md"));
let default_qa = crate::config::ProjectConfig::load(project_root)
.unwrap_or_default()
.default_qa_mode();
matches!(
crate::io::story_metadata::resolve_qa_mode(&story_path, default_qa),
crate::io::story_metadata::QaMode::Human
)
}
};
if needs_human_review {
let story_path = project_root
.join(".storkit/work/3_qa")
.join(format!("{story_id}.md"));
if let Err(e) =
crate::io::story_metadata::write_review_hold(&story_path)
{
eprintln!(
"[startup:reconcile] Failed to set review_hold on '{story_id}': {e}"
);
}
eprintln!(
"[startup:reconcile] '{story_id}' passed QA — holding for human review."
);
let _ = progress_tx.send(ReconciliationEvent {
story_id: story_id.clone(),
status: "review_hold".to_string(),
message: "Passed QA — waiting for human review.".to_string(),
});
} else if let Err(e) =
crate::agents::move_story_to_merge(project_root, story_id)
{
eprintln!(
"[startup:reconcile] Failed to move '{story_id}' to 4_merge/: {e}"
);
let _ = progress_tx.send(ReconciliationEvent {
story_id: story_id.clone(),
status: "failed".to_string(),
message: format!("Failed to advance to merge: {e}"),
});
} else {
eprintln!("[startup:reconcile] Moved '{story_id}' → 4_merge/.");
let _ = progress_tx.send(ReconciliationEvent {
story_id: story_id.clone(),
status: "advanced".to_string(),
message: "Gates passed — moved to merge.".to_string(),
});
}
} else {
eprintln!(
"[startup:reconcile] Coverage gate failed for '{story_id}': {coverage_output}\n\
Leaving in 3_qa/ for auto-assign to restart the QA agent."
);
let _ = progress_tx.send(ReconciliationEvent {
story_id: story_id.clone(),
status: "failed".to_string(),
message: "Coverage gate failed; will be retried.".to_string(),
});
}
}
}
// Signal that reconciliation is complete.
let _ = progress_tx.send(ReconciliationEvent {
story_id: String::new(),
status: "done".to_string(),
message: "Startup reconciliation complete.".to_string(),
});
}
}
// ── Tests ──────────────────────────────────────────────────────────────────
#[cfg(test)]
mod tests {
use std::process::Command;
use tokio::sync::broadcast;
use super::super::super::AgentPool;
use crate::agents::ReconciliationEvent;
fn init_git_repo(repo: &std::path::Path) {
Command::new("git")
.args(["init"])
.current_dir(repo)
.output()
.unwrap();
Command::new("git")
.args(["config", "user.email", "test@test.com"])
.current_dir(repo)
.output()
.unwrap();
Command::new("git")
.args(["config", "user.name", "Test"])
.current_dir(repo)
.output()
.unwrap();
// Create initial commit so master branch exists.
std::fs::write(repo.join("README.md"), "# test\n").unwrap();
Command::new("git")
.args(["add", "."])
.current_dir(repo)
.output()
.unwrap();
Command::new("git")
.args(["commit", "-m", "initial"])
.current_dir(repo)
.output()
.unwrap();
}
#[tokio::test]
async fn reconcile_on_startup_noop_when_no_worktrees() {
let tmp = tempfile::tempdir().unwrap();
let pool = AgentPool::new_test(3001);
let (tx, _rx) = broadcast::channel(16);
// Should not panic; no worktrees to reconcile.
pool.reconcile_on_startup(tmp.path(), &tx).await;
}
#[tokio::test]
async fn reconcile_on_startup_emits_done_event() {
let tmp = tempfile::tempdir().unwrap();
let pool = AgentPool::new_test(3001);
let (tx, mut rx) = broadcast::channel::<ReconciliationEvent>(16);
pool.reconcile_on_startup(tmp.path(), &tx).await;
// Collect all events; the last must be "done".
let mut events: Vec<ReconciliationEvent> = Vec::new();
while let Ok(evt) = rx.try_recv() {
events.push(evt);
}
assert!(
events.iter().any(|e| e.status == "done"),
"reconcile_on_startup must emit a 'done' event; got: {:?}",
events.iter().map(|e| &e.status).collect::<Vec<_>>()
);
}
#[tokio::test]
async fn reconcile_on_startup_skips_story_without_committed_work() {
use std::fs;
let tmp = tempfile::tempdir().unwrap();
let root = tmp.path();
// Set up story in 2_current/.
let current = root.join(".storkit/work/2_current");
fs::create_dir_all(&current).unwrap();
fs::write(current.join("60_story_test.md"), "test").unwrap();
// Create a worktree directory that is a fresh git repo with no commits
// ahead of its own base branch (simulates a worktree where no work was done).
let wt_dir = root.join(".storkit/worktrees/60_story_test");
fs::create_dir_all(&wt_dir).unwrap();
init_git_repo(&wt_dir);
let pool = AgentPool::new_test(3001);
let (tx, _rx) = broadcast::channel(16);
pool.reconcile_on_startup(root, &tx).await;
// Story should still be in 2_current/ — nothing was reconciled.
assert!(
current.join("60_story_test.md").exists(),
"story should stay in 2_current/ when worktree has no committed work"
);
}
#[tokio::test]
async fn reconcile_on_startup_runs_gates_on_worktree_with_committed_work() {
use std::fs;
let tmp = tempfile::tempdir().unwrap();
let root = tmp.path();
// Set up a git repo for the project root.
init_git_repo(root);
// Set up story in 2_current/ and commit it so the project root is clean.
let current = root.join(".storkit/work/2_current");
fs::create_dir_all(&current).unwrap();
fs::write(current.join("61_story_test.md"), "test").unwrap();
Command::new("git")
.args(["add", "."])
.current_dir(root)
.output()
.unwrap();
Command::new("git")
.args([
"-c",
"user.email=test@test.com",
"-c",
"user.name=Test",
"commit",
"-m",
"add story",
])
.current_dir(root)
.output()
.unwrap();
// Create a real git worktree for the story.
let wt_dir = root.join(".storkit/worktrees/61_story_test");
fs::create_dir_all(wt_dir.parent().unwrap()).unwrap();
Command::new("git")
.args([
"worktree",
"add",
&wt_dir.to_string_lossy(),
"-b",
"feature/story-61_story_test",
])
.current_dir(root)
.output()
.unwrap();
// Add a commit to the feature branch (simulates coder completing work).
fs::write(wt_dir.join("implementation.txt"), "done").unwrap();
Command::new("git")
.args(["add", "."])
.current_dir(&wt_dir)
.output()
.unwrap();
Command::new("git")
.args([
"-c",
"user.email=test@test.com",
"-c",
"user.name=Test",
"commit",
"-m",
"implement story",
])
.current_dir(&wt_dir)
.output()
.unwrap();
assert!(
crate::agents::gates::worktree_has_committed_work(&wt_dir),
"test setup: worktree should have committed work"
);
let pool = AgentPool::new_test(3001);
let (tx, _rx) = broadcast::channel(16);
pool.reconcile_on_startup(root, &tx).await;
// In the test env, cargo clippy will fail (no Cargo.toml) so gates fail
// and the story stays in 2_current/. The important assertion is that
// reconcile ran without panicking and the story is in a consistent state.
let in_current = current.join("61_story_test.md").exists();
let in_qa = root.join(".storkit/work/3_qa/61_story_test.md").exists();
assert!(
in_current || in_qa,
"story should be in 2_current/ or 3_qa/ after reconciliation"
);
}
}