fix(agent-forge): safe concurrent pair execution, sentinel parsing, w…

[sinke237]

Summary

Replace per-pair multi-thread Tokio runtime with a current-thread runtime to prevent thread explosion, add config validation for AGENT_FORGE_MAX_CONCURRENCY, make SENTINEL stdout parsing robust, fix noisy watchdog checks, and improve CLAUDE_PATH validation messaging.

Changes (concise):

• Concurrency runtime: Replaced per-pair tokio::runtime::Runtime::new() inside spawn_blocking with a current-thread runtime (tokio::runtime::Builder::new_current_thread().enable_all().build()) to avoid spawning a multi-threaded runtime per pair and prevent CPU/thread oversubscription.
• Config validation: Added warnings when AGENT_FORGE_MAX_CONCURRENCY is set but invalid (non-numeric or <= 0) instead of silently ignoring it.
• Sentinel parsing: Made read_sentinel_result_payload robust by scanning reversed log lines and attempting to parse JSON (and extracting JSON substrings) until a "result" payload is found. Logs a warning if none found.
• Watchdog check: Replaced modulo-based check with a monotonic last_watchdog_check: Instant to run the watchdog approximately once per 60s and avoid repeated checks within the same second.
• CLAUDE validation: Improved CLAUDE_PATH validation messages to be more actionable and call out that process spawns will fail if the binary is missing/misconfigured.

[Copilot]

Pull request overview

This PR improves the stability and testability of the agent-forge “pair” execution loop by adding a configurable concurrency limiter, avoiding per-pair multi-thread Tokio runtime creation, and making SENTINEL output parsing and watchdog timing more robust.

Changes:

• Add an optional semaphore-based concurrency limit for pair lifecycles via AGENT_FORGE_MAX_CONCURRENCY, plus validation/warnings for invalid values.
• Run each pair lifecycle inside spawn_blocking with a current-thread Tokio runtime to avoid thread explosion.
• Improve robustness of SENTINEL stdout parsing and watchdog check scheduling; add an integration test for concurrency using a simulated pair mode.

Reviewed changes

Copilot reviewed 6 out of 6 changed files in this pull request and generated 4 comments.

Show a summary per file

File	Description
crates/pair-harness/src/process.rs	Improves CLAUDE_PATH error messaging and refactors env injection formatting.
crates/pair-harness/src/pair.rs	Adds watchdog timing change, robust SENTINEL parsing, and introduces simulated pair test hooks.
crates/pair-harness/src/lib.rs	Re-exports new simulated test hook counters.
crates/agent-forge/src/lib.rs	Adds optional semaphore concurrency limiter and executes pair lifecycles in spawn_blocking with a current-thread runtime.
crates/agent-forge/tests/concurrency_integration.rs	Adds an integration test asserting two pairs can run concurrently using simulated pair mode.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[Copilot]

These SIMULATED_* atomics are exported as pub static and (via pair-harness's lib.rs) become part of the crate’s public API even though they are described as test hooks. If this is intended only for integration testing, consider keeping them private and exposing them only under cfg(test)/feature-gated APIs to avoid locking in a public surface area and to prevent external consumers from depending on test-only globals.

Suggested change

	pub static SIMULATED_ACTIVE_PAIRS: AtomicUsize = AtomicUsize::new(0);
	pub static SIMULATED_MAX_ACTIVE_PAIRS: AtomicUsize = AtomicUsize::new(0);
	pub(crate) static SIMULATED_ACTIVE_PAIRS: AtomicUsize = AtomicUsize::new(0);
	pub(crate) static SIMULATED_MAX_ACTIVE_PAIRS: AtomicUsize = AtomicUsize::new(0);

[Copilot]

Re-exporting SIMULATED_ACTIVE_PAIRS / SIMULATED_MAX_ACTIVE_PAIRS from the crate root makes the test-only counters part of pair-harness’s public API. If the intent is only to support internal/integration tests, consider gating these re-exports behind cfg(test) or a cargo feature so production consumers don’t see (or rely on) them.

Suggested change

	pub use worktree::WorktreeManager;
	pub use worktree::WorktreeManager;
	#[cfg(test)]

[Copilot]

This test mutates process-global environment variables (AGENT_FLOW_TEST_PAIR_DELAY_MS, AGENT_FORGE_MAX_CONCURRENCY) without restoring prior values on panic, and without isolating against other concurrently running tests. This can make the suite flaky. Consider capturing the previous values and restoring them with a small RAII guard (Drop), and resetting SIMULATED_MAX_ACTIVE_PAIRS to 0 at the start of the test so it can’t pass due to a leftover max from a prior run.

[Copilot]

The AGENT_FLOW_TEST_PAIR_DELAY_MS test hook is compiled into the production ForgeSentinelPair::run path and causes the lifecycle to short-circuit (including returning a synthetic PairOutcome::PrOpened). If this env var is accidentally set in non-test runs, it will silently bypass real provisioning/process spawning and produce incorrect behavior. Consider gating this behind cfg(test) or a dedicated cargo feature (e.g. test-hooks), and/or moving the simulation hook into test-only code so it cannot be enabled in production builds.

Suggested change

	// Test hook: if AGENT_FLOW_TEST_PAIR_DELAY_MS is set, simulate a pair run
	// by sleeping for the configured milliseconds and updating counters.
	if let Ok(val) = std::env::var("AGENT_FLOW_TEST_PAIR_DELAY_MS") {
	if let Ok(ms) = val.parse::<u64>() {
	let cur = SIMULATED_ACTIVE_PAIRS.fetch_add(1, Ordering::SeqCst) + 1;
	// update max observed
	loop {
	let prev = SIMULATED_MAX_ACTIVE_PAIRS.load(Ordering::SeqCst);
	if cur > prev {
	if SIMULATED_MAX_ACTIVE_PAIRS
	.compare_exchange(prev, cur, Ordering::SeqCst, Ordering::SeqCst)
	.is_ok()
	{
	break;
	}
	} else {
	break;
	}
	}
	std::thread::sleep(std::time::Duration::from_millis(ms));
	SIMULATED_ACTIVE_PAIRS.fetch_sub(1, Ordering::SeqCst);
	return Ok(PairOutcome::PrOpened {
	pr_url: format!("http://localhost/{}/pr", ticket.id),
	pr_number: 1,
	branch: format!("forge-{}/{}", self.config.pair_id, ticket.id),
	});
	#[cfg(test)]
	{
	// Test hook: if AGENT_FLOW_TEST_PAIR_DELAY_MS is set, simulate a pair run
	// by sleeping for the configured milliseconds and updating counters.
	if let Ok(val) = std::env::var("AGENT_FLOW_TEST_PAIR_DELAY_MS") {
	if let Ok(ms) = val.parse::<u64>() {
	let cur = SIMULATED_ACTIVE_PAIRS.fetch_add(1, Ordering::SeqCst) + 1;
	// update max observed
	loop {
	let prev = SIMULATED_MAX_ACTIVE_PAIRS.load(Ordering::SeqCst);
	if cur > prev {
	if SIMULATED_MAX_ACTIVE_PAIRS
	.compare_exchange(prev, cur, Ordering::SeqCst, Ordering::SeqCst)
	.is_ok()
	{
	break;
	}
	} else {
	break;
	}
	}
	std::thread::sleep(std::time::Duration::from_millis(ms));
	SIMULATED_ACTIVE_PAIRS.fetch_sub(1, Ordering::SeqCst);
	return Ok(PairOutcome::PrOpened {
	pr_url: format!("http://localhost/{}/pr", ticket.id),
	pr_number: 1,
	branch: format!("forge-{}/{}", self.config.pair_id, ticket.id),
	});
	}

[Christiantyemele]

@sinke237 glad to review when conflicts resolve