dave/storkit
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 7 Wiki Activity

Make merge_agent_work async to avoid MCP 60-second tool timeout

Browse Source 
The merge pipeline (squash merge + quality gates) takes well over 60
seconds. Claude Code's MCP HTTP transport times out at 60s, causing
"completed with no output" — the mergemaster retries fruitlessly.

merge_agent_work now starts the pipeline as a background task and
returns immediately. A new get_merge_status tool lets the mergemaster
poll until the job reaches a terminal state. Also adds a double-start
guard so concurrent calls for the same story are rejected.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Dave
2026-03-17 12:15:42 +00:00
parent a85d1a1170
commit 7c6e1b445d
4 changed files with 289 additions and 87 deletions
Download Patch File Download Diff File Expand all files Collapse all files

29
server/src/agents/merge.rs
View File

@@ -1,5 +1,6 @@
use std::path::Path;
use std::process::Command;
use std::sync::Mutex;
use serde::Serialize;
@@ -7,6 +8,29 @@ use crate::config::ProjectConfig;
use super::gates::run_project_tests;
/// Global lock ensuring only one squash-merge runs at a time.
///
/// The merge pipeline uses a shared `.story_kit/merge_workspace` directory and
/// temporary `merge-queue/{story_id}` branches. If two merges run concurrently,
/// the second call's initial cleanup destroys the first call's branch mid-flight,
/// causing `git cherry-pick merge-queue/…` to fail with "bad revision".
static MERGE_LOCK: Mutex<()> = Mutex::new(());
/// Status of an async merge job.
#[derive(Debug, Clone, Serialize)]
pub enum MergeJobStatus {
Running,
Completed(MergeReport),
Failed(String),
}
/// Tracks a background merge job started by `merge_agent_work`.
#[derive(Debug, Clone, Serialize)]
pub struct MergeJob {
pub story_id: String,
pub status: MergeJobStatus,
}
/// Result of a mergemaster merge operation.
#[derive(Debug, Serialize, Clone)]
pub struct MergeReport {
@@ -57,6 +81,11 @@ pub(crate) fn run_squash_merge(
branch: &str,
story_id: &str,
) -> Result<SquashMergeResult, String> {
// Acquire the merge lock so concurrent calls don't clobber each other.
let _lock = MERGE_LOCK
.lock()
.map_err(|e| format!("Merge lock poisoned: {e}"))?;
let mut all_output = String::new();
let merge_branch = format!("merge-queue/{story_id}");
let merge_wt_path = project_root

176
server/src/agents/pool.rs
View File

@@ -124,6 +124,10 @@ pub struct AgentPool {
/// an `AgentStateChanged` event is emitted so the frontend can refresh the
/// pipeline board without waiting for a filesystem event.
watcher_tx: broadcast::Sender<WatcherEvent>,
/// Tracks background merge jobs started by `merge_agent_work`, keyed by story_id.
/// The MCP tool returns immediately and the mergemaster agent polls
/// `get_merge_status` until the job reaches a terminal state.
merge_jobs: Arc<Mutex<HashMap<String, super::merge::MergeJob>>>,
}
impl AgentPool {
@@ -133,6 +137,7 @@ impl AgentPool {
port,
child_killers: Arc::new(Mutex::new(HashMap::new())),
watcher_tx,
merge_jobs: Arc::new(Mutex::new(HashMap::new())),
}
}
@@ -1110,6 +1115,7 @@ impl AgentPool {
port: self.port,
child_killers: Arc::clone(&self.child_killers),
watcher_tx: self.watcher_tx.clone(),
merge_jobs: Arc::clone(&self.merge_jobs),
};
let sid = story_id.to_string();
let aname = agent_name.to_string();
@@ -1138,8 +1144,67 @@ impl AgentPool {
/// 4. If gates pass: cherry-pick the squash commit onto master and archive the story.
///
/// Returns a `MergeReport` with full details of what happened.
pub async fn merge_agent_work(
&self,
/// Start the merge pipeline as a background task.
///
/// Returns immediately so the MCP tool call doesn't time out (the full
/// pipeline — squash merge + quality gates — takes well over 60 seconds,
/// exceeding Claude Code's MCP tool-call timeout).
///
/// The mergemaster agent should poll [`get_merge_status`](Self::get_merge_status)
/// until the job reaches a terminal state.
pub fn start_merge_agent_work(
self: &Arc<Self>,
project_root: &Path,
story_id: &str,
) -> Result<(), String> {
// Guard against double-starts.
{
let jobs = self.merge_jobs.lock().map_err(|e| e.to_string())?;
if let Some(job) = jobs.get(story_id)
&& matches!(job.status, super::merge::MergeJobStatus::Running)
{
return Err(format!(
"Merge already in progress for '{story_id}'. \
Use get_merge_status to poll for completion."
));
}
}
// Insert Running job.
{
let mut jobs = self.merge_jobs.lock().map_err(|e| e.to_string())?;
jobs.insert(
story_id.to_string(),
super::merge::MergeJob {
story_id: story_id.to_string(),
status: super::merge::MergeJobStatus::Running,
},
);
}
let pool = Arc::clone(self);
let root = project_root.to_path_buf();
let sid = story_id.to_string();
tokio::spawn(async move {
let report = pool.run_merge_pipeline(&root, &sid).await;
let status = match report {
Ok(r) => super::merge::MergeJobStatus::Completed(r),
Err(e) => super::merge::MergeJobStatus::Failed(e),
};
if let Ok(mut jobs) = pool.merge_jobs.lock()
&& let Some(job) = jobs.get_mut(&sid)
{
job.status = status;
}
});
Ok(())
}
/// The actual merge pipeline, run inside a background task.
async fn run_merge_pipeline(
self: &Arc<Self>,
project_root: &Path,
story_id: &str,
) -> Result<super::merge::MergeReport, String> {
@@ -1149,8 +1214,6 @@ impl AgentPool {
let sid = story_id.to_string();
let br = branch.clone();
// Run blocking operations (git + cargo + quality gates) off the async runtime.
// Quality gates now run inside run_squash_merge before the fast-forward.
let merge_result =
tokio::task::spawn_blocking(move || super::merge::run_squash_merge(&root, &br, &sid))
.await
@@ -1170,13 +1233,11 @@ impl AgentPool {
});
}
// Merge + gates both passed — archive the story and clean up agent entries.
let story_archived = super::lifecycle::move_story_to_archived(project_root, story_id).is_ok();
if story_archived {
self.remove_agents_for_story(story_id);
}
// Clean up the worktree if it exists.
let worktree_cleaned_up = if wt_path.exists() {
let config = crate::config::ProjectConfig::load(project_root)
.unwrap_or_default();
@@ -1187,10 +1248,6 @@ impl AgentPool {
false
};
// Mergemaster slot is now free — trigger auto-assign so remaining
// items in 4_merge/ (or other stages) get picked up. The normal
// server-owned completion handler won't run because we already
// removed the agent entry above.
self.auto_assign_available_work(project_root).await;
Ok(super::merge::MergeReport {
@@ -1206,6 +1263,14 @@ impl AgentPool {
})
}
/// Check the status of a background merge job.
pub fn get_merge_status(&self, story_id: &str) -> Option<super::merge::MergeJob> {
self.merge_jobs
.lock()
.ok()
.and_then(|jobs| jobs.get(story_id).cloned())
}
/// Return the port this server is running on.
pub fn port(&self) -> u16 {
self.port
@@ -2128,6 +2193,7 @@ fn spawn_pipeline_advance(
port,
child_killers: Arc::new(Mutex::new(HashMap::new())),
watcher_tx,
merge_jobs: Arc::new(Mutex::new(HashMap::new())),
};
pool.run_pipeline_advance(
&sid,
@@ -2144,6 +2210,7 @@ fn spawn_pipeline_advance(
#[cfg(test)]
mod tests {
use super::*;
use crate::agents::merge::{MergeJob, MergeJobStatus};
use crate::agents::{
AgentEvent, AgentStatus, CompletionReport, PipelineStage, ReconciliationEvent,
lifecycle::move_story_to_archived,
@@ -4087,6 +4154,23 @@ stage = "coder"
// ── merge_agent_work tests ────────────────────────────────────────────────
/// Helper: start a merge and poll until terminal state.
async fn run_merge_to_completion(
pool: &Arc<AgentPool>,
repo: &std::path::Path,
story_id: &str,
) -> MergeJob {
pool.start_merge_agent_work(repo, story_id).unwrap();
loop {
tokio::time::sleep(std::time::Duration::from_millis(50)).await;
if let Some(job) = pool.get_merge_status(story_id)
&& !matches!(job.status, MergeJobStatus::Running)
{
return job;
}
}
}
#[tokio::test]
async fn merge_agent_work_returns_error_when_branch_not_found() {
use tempfile::tempdir;
@@ -4095,14 +4179,19 @@ stage = "coder"
let repo = tmp.path();
init_git_repo(repo);
let pool = AgentPool::new_test(3001);
// branch feature/story-99_nonexistent does not exist
let result = pool
.merge_agent_work(repo, "99_nonexistent")
.await
.unwrap();
// Should fail (no branch) — not panic
assert!(!result.success, "should fail when branch missing");
let pool = Arc::new(AgentPool::new_test(3001));
let job = run_merge_to_completion(&pool, repo, "99_nonexistent").await;
match &job.status {
MergeJobStatus::Completed(report) => {
assert!(!report.success, "should fail when branch missing");
}
MergeJobStatus::Failed(_) => {
// Also acceptable — the pipeline errored out
}
MergeJobStatus::Running => {
panic!("should not still be running");
}
}
}
#[tokio::test]
@@ -4155,22 +4244,29 @@ stage = "coder"
.output()
.unwrap();
let pool = AgentPool::new_test(3001);
let report = pool.merge_agent_work(repo, "23_test").await.unwrap();
let pool = Arc::new(AgentPool::new_test(3001));
let job = run_merge_to_completion(&pool, repo, "23_test").await;
// Merge should succeed (gates will run but cargo/pnpm results will depend on env)
// At minimum the merge itself should succeed
assert!(!report.had_conflicts, "should have no conflicts");
// Note: gates_passed may be false in test env without Rust project, that's OK
// The important thing is the merge itself ran
assert!(
report.success || report.gate_output.contains("Failed to run") || !report.gates_passed,
"report should be coherent: {report:?}"
);
// Story should be in done if gates passed
if report.story_archived {
let done = repo.join(".story_kit/work/5_done/23_test.md");
assert!(done.exists(), "done file should exist");
match &job.status {
MergeJobStatus::Completed(report) => {
assert!(!report.had_conflicts, "should have no conflicts");
assert!(
report.success || report.gate_output.contains("Failed to run") || !report.gates_passed,
"report should be coherent: {report:?}"
);
if report.story_archived {
let done = repo.join(".story_kit/work/5_done/23_test.md");
assert!(done.exists(), "done file should exist");
}
}
MergeJobStatus::Failed(e) => {
// Gate failures are acceptable in test env
assert!(
e.contains("Failed") || e.contains("failed"),
"unexpected failure: {e}"
);
}
MergeJobStatus::Running => panic!("should not still be running"),
}
}
@@ -4348,8 +4444,8 @@ stage = "coder"
.output()
.unwrap();
let pool = AgentPool::new_test(3001);
let report = pool.merge_agent_work(repo, "42_story_foo").await.unwrap();
let pool = Arc::new(AgentPool::new_test(3001));
let job = run_merge_to_completion(&pool, repo, "42_story_foo").await;
// Master should NEVER have conflict markers, regardless of merge outcome.
let master_code = fs::read_to_string(repo.join("code.rs")).unwrap();
@@ -4363,7 +4459,15 @@ stage = "coder"
);
// The report should accurately reflect what happened.
assert!(report.had_conflicts, "should report conflicts");
match &job.status {
MergeJobStatus::Completed(report) => {
assert!(report.had_conflicts, "should report conflicts");
}
MergeJobStatus::Failed(_) => {
// Acceptable — merge aborted due to conflicts
}
MergeJobStatus::Running => panic!("should not still be running"),
}
}
// ── reconcile_on_startup tests ────────────────────────────────────────────