use crate::CliError;
use crate::config_parse::checkout_lock_timeout;
use crate::parallel::retry_after_from_reason;
use crate::pipeline_gate::{abort, finish_workflow, loop_back_to_code, run_gate, transition};
use crate::pipeline_launch::launch_stage;
use devflow_core::config::GitFlowConfig;
use devflow_core::gates::{GateAction, Gates};
use devflow_core::hooks::{self, HookContext};
use devflow_core::mode;
use devflow_core::prompt::FixType;
use devflow_core::stage::Stage;
use devflow_core::state::State;
use devflow_core::{
agent_result,
agent_result::{AgentStatus, Verdict},
events, lock, workflow,
};
use std::path::{Path, PathBuf};
pub(crate) fn handle_infra_outcome(
project_root: &Path,
state: &mut State,
stage: Stage,
reason: Option<String>,
) -> Result<(), CliError> {
state.infra_failures = state.infra_failures.saturating_add(1);
workflow::save_state(state)?;
gate_or_abort_infra(project_root, state, stage, reason)
}
pub(crate) fn gate_or_abort_infra(
project_root: &Path,
state: &mut State,
stage: Stage,
reason: Option<String>,
) -> Result<(), CliError> {
if state.infra_failures >= mode::MAX_INFRA_FAILURES {
return abort(
project_root,
state,
&format!(
"infrastructure failures reached the ceiling ({} of {}) — aborting rather than gating again",
state.infra_failures,
mode::MAX_INFRA_FAILURES
),
);
}
handle_stage_failure(project_root, state, stage, reason)
}
pub(crate) fn handle_rate_limited_outcome(
project_root: &Path,
state: &mut State,
phase: u32,
stage: Stage,
reason: Option<String>,
) -> Result<(), CliError> {
let retry_after = retry_after_from_reason(reason.as_deref());
let projected_infra_failures = state.infra_failures.saturating_add(1);
if projected_infra_failures >= mode::MAX_INFRA_FAILURES {
return handle_infra_outcome(project_root, state, stage, reason);
}
state.infra_failures = projected_infra_failures;
workflow::save_state(state)?;
let instructions =
devflow_core::ship::build_single_agent_cron_instructions(project_root, phase, &retry_after);
devflow_core::ship::write_cron_instructions(project_root, &instructions)?;
if instructions.hermes_cron.schedule.is_empty() {
return gate_or_abort_infra(
project_root,
state,
stage,
Some(format!(
"rate limited with no parseable retry time ({retry_after}) — auto-resume cron not scheduled; resume manually"
)),
);
}
println!(
"rate limited — wrote {}",
devflow_core::ship::cron_instructions_path(project_root, phase)
.strip_prefix(project_root)
.map(|p| p.display().to_string())
.unwrap_or_else(|_| {
devflow_core::ship::cron_instructions_path(project_root, phase)
.display()
.to_string()
})
);
events::emit(
project_root,
phase,
"rate_limit_resume_scheduled",
serde_json::json!({
"stage": stage.to_string(),
"retry_after": retry_after,
"infra_failures": state.infra_failures,
}),
);
Ok(())
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) enum ValidateOutcome {
Passed,
Failed,
Ambiguous(String),
}
pub(crate) fn classify_validate_outcome(result: &agent_result::AgentResult) -> ValidateOutcome {
let external = result.decided_by_layer == Some(0) && result.status == AgentStatus::Success;
match (external, result.verdict) {
(_, Some(Verdict::Pass)) => ValidateOutcome::Passed,
(true, Some(Verdict::Gaps)) => ValidateOutcome::Ambiguous(
"external verification passed but the agent reported gaps".to_string(),
),
(true, None) => ValidateOutcome::Ambiguous(
"external verification passed but no agent verdict arrived".to_string(),
),
_ => ValidateOutcome::Failed,
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum ValidateResult {
Passed,
Failed,
}
pub(crate) fn handle_validate_outcome(
project_root: &Path,
state: &mut State,
outcome: ValidateOutcome,
) -> Result<(), CliError> {
let result = match outcome {
ValidateOutcome::Ambiguous(detail) => {
let context = format!(
"[never-silent] validate ambiguous: {}",
truncate_reason(&detail)
);
return match run_gate(project_root, state, Stage::Validate, &context)? {
GateAction::Advance => transition(project_root, state, Stage::Ship),
GateAction::LoopBack(_) => {
loop_back_to_code(project_root, state, FixType::GapsOnly)
}
GateAction::Abort(reason) => abort(project_root, state, &reason),
};
}
ValidateOutcome::Passed => ValidateResult::Passed,
ValidateOutcome::Failed => ValidateResult::Failed,
};
if result == ValidateResult::Failed {
state.consecutive_failures = state.consecutive_failures.saturating_add(1);
workflow::save_state(state)?;
}
if state
.mode
.should_gate(Stage::Validate, state.consecutive_failures)
{
let context = match result {
ValidateResult::Passed => "Validation passed — approve to ship?".to_string(),
ValidateResult::Failed => format!(
"Validation failed {} time(s) — human review needed.",
state.consecutive_failures
),
};
return match run_gate(project_root, state, Stage::Validate, &context)? {
GateAction::Advance => transition(project_root, state, Stage::Ship),
GateAction::LoopBack(_) => loop_back_to_code(project_root, state, FixType::GapsOnly),
GateAction::Abort(reason) => abort(project_root, state, &reason),
};
}
match result {
ValidateResult::Passed => transition(project_root, state, Stage::Ship),
ValidateResult::Failed => loop_back_to_code(project_root, state, FixType::GapsOnly),
}
}
pub(crate) fn handle_ship_outcome(project_root: &Path, state: &mut State) -> Result<(), CliError> {
match run_gate(
project_root,
state,
Stage::Ship,
"Ship complete — approve merge?",
)? {
GateAction::Advance => finish_workflow(project_root, state),
GateAction::LoopBack(_) => loop_back_to_code(project_root, state, FixType::GapsOnly),
GateAction::Abort(reason) => abort(project_root, state, &reason),
}
}
pub(crate) fn truncate_reason(reason: &str) -> String {
render_gate_context(reason, 300)
}
pub(crate) fn render_gate_context(context: &str, max_chars: usize) -> String {
const TRUNCATED: &str = "… [truncated; full output in .devflow/]";
let sanitized: String = context
.chars()
.map(|character| {
if character.is_control() {
' '
} else {
character
}
})
.collect();
if sanitized.chars().count() <= max_chars {
return sanitized;
}
let suffix_len = TRUNCATED.chars().count().min(max_chars);
let head_len = max_chars.saturating_sub(suffix_len);
let head: String = sanitized.chars().take(head_len).collect();
let suffix: String = TRUNCATED.chars().take(suffix_len).collect();
format!("{head}{suffix}")
}
pub(crate) fn handle_stage_failure(
project_root: &Path,
state: &mut State,
stage: Stage,
reason: Option<String>,
) -> Result<(), CliError> {
let context = format!(
"[never-silent] stage {stage} failed: {} — human review needed (retry, loop-to-code, or abort)",
truncate_reason(&reason.unwrap_or_else(|| "no details available".into()))
);
match run_gate(project_root, state, stage, &context)? {
GateAction::Advance => {
let _ = Gates::cleanup(project_root, state.phase, stage);
state.gate_pending = false;
launch_stage(state, None, Some(stage))
}
GateAction::LoopBack(_) => {
let _ = Gates::cleanup(project_root, state.phase, stage);
launch_stage(state, None, Some(stage))
}
GateAction::Abort(reason) => abort(project_root, state, &reason),
}
}
pub(crate) fn handle_ship_failure(
project_root: &Path,
state: &mut State,
reason: Option<String>,
) -> Result<(), CliError> {
if is_ship_review_failure(&reason) {
return loop_back_to_code(project_root, state, FixType::AuditFix);
}
handle_stage_failure(project_root, state, Stage::Ship, reason)
}
pub(crate) fn is_ship_review_failure(reason: &Option<String>) -> bool {
reason
.as_deref()
.map(|r| r.trim().to_ascii_lowercase().starts_with("review:"))
.unwrap_or(false)
}
pub(crate) fn hook_context_root(
project_root: &Path,
state: &State,
terminal_batch: bool,
) -> PathBuf {
if terminal_batch {
return project_root.to_path_buf();
}
state
.worktree_path
.as_ref()
.filter(|path| path.exists())
.map(|path| path.to_path_buf())
.unwrap_or_else(|| project_root.to_path_buf())
}
pub(crate) fn run_checkout_hooks(
project_root: &Path,
state: &State,
batch: &[hooks::Hook],
stage: Stage,
) -> bool {
if batch.is_empty() {
return true;
}
let _checkout_lock = match lock::acquire_project_blocking(project_root, checkout_lock_timeout())
{
Ok(guard) => guard,
Err(err) => {
println!(
"warning: could not acquire the checkout lock ({err}) — \
SKIPPING hooks {batch:?} rather than mutating the checkout \
unserialized. Re-run them once the holder finishes."
);
events::emit(
project_root,
state.phase,
"checkout_lock_timeout",
serde_json::json!({ "stage": stage.to_string(), "error": err.to_string() }),
);
for hook in batch {
events::emit(
project_root,
state.phase,
"hook_run",
serde_json::json!({
"hook": format!("{hook:?}"),
"ok": false,
"skipped": "checkout lock timeout",
}),
);
}
return false;
}
};
let git_flow = GitFlowConfig::default();
let mut all_succeeded = true;
let terminal_batch = batch == hooks::hooks_after_ship().as_slice();
let hook_root = hook_context_root(project_root, state, terminal_batch);
let mut ctx = HookContext {
phase: state.phase,
project_root: hook_root.clone(),
stage,
git_flow: git_flow.clone(),
shipped_version: None,
};
for hook in batch {
let outcome = hook.run(&mut ctx);
if let Err(ref err) = outcome {
println!("warning: hook {hook:?} failed: {err}");
all_succeeded = false;
}
events::emit(
project_root,
state.phase,
"hook_run",
serde_json::json!({
"hook": format!("{hook:?}"),
"ok": outcome.is_ok(),
}),
);
if terminal_batch && outcome.is_err() {
break;
}
}
all_succeeded
}
#[cfg(test)]
mod tests {
use super::*;
use crate::pipeline_gate::prepare_loop_back_to_code;
use crate::pipeline_launch::advance;
use crate::test_support::*;
use devflow_core::git::GitFlow;
use devflow_core::mode::Mode;
use devflow_core::prompt;
use devflow_core::state::AgentKind;
#[test]
fn checkout_hooks_skip_instead_of_running_unserialized_on_lock_timeout() {
let _guard = ENV_MUTEX.lock().unwrap();
let dir = tempfile::tempdir().unwrap();
let root = dir.path();
let _held = lock::acquire_project(root).expect("hold checkout lock");
unsafe {
std::env::set_var("DEVFLOW_CHECKOUT_LOCK_TIMEOUT_SECS", "0");
}
let state = State::new(33, AgentKind::Claude, Mode::Auto, root.to_path_buf());
run_checkout_hooks(root, &state, &hooks::hooks_after_ship(), Stage::Ship);
unsafe {
std::env::remove_var("DEVFLOW_CHECKOUT_LOCK_TIMEOUT_SECS");
}
assert!(
!root.join("CHANGELOG.md").exists(),
"hooks must not run while the checkout lock is held elsewhere"
);
let last = devflow_core::events::last_event_for_phase(root, 33)
.expect("skip must be recorded in events.jsonl");
assert_eq!(last["event"], "hook_run");
assert_eq!(last["ok"], false);
assert_eq!(last["skipped"], "checkout lock timeout");
}
#[test]
fn terminal_hook_failure_stops_before_branch_cleanup() {
let dir = tempfile::tempdir().unwrap();
let root = dir.path();
init_repo(root);
let phase = 34;
let branch = "feature/phase-34";
let git = |args: &[&str]| {
let output = devflow_core::test_support::git_command(root)
.args(args)
.output()
.unwrap();
assert!(output.status.success(), "git {args:?} failed");
};
git(&["branch", branch, "develop"]);
std::fs::remove_file(root.join("Cargo.toml")).unwrap();
std::fs::create_dir(root.join("Cargo.toml")).unwrap();
let state = State::new(phase, AgentKind::Claude, Mode::Auto, root.to_path_buf());
let succeeded = run_checkout_hooks(root, &state, &hooks::hooks_after_ship(), Stage::Ship);
assert!(!succeeded);
assert!(
GitFlow::new(root).branch_exists(branch),
"a failed terminal batch must preserve the branch for retry"
);
}
#[test]
fn run_checkout_hooks_keeps_changelog_in_sync_with_tag_when_no_version_file() {
let dir = tempfile::tempdir().unwrap();
let root = dir.path();
init_repo_no_version_file(root);
let phase = 47;
let branch = format!("feature/phase-{phase:02}");
let git = |args: &[&str]| {
let output = devflow_core::test_support::git_command(root)
.args(args)
.output()
.unwrap();
assert!(output.status.success(), "git {args:?} failed");
};
git(&["branch", &branch, "develop"]);
std::fs::write(root.join(".gitignore"), ".devflow/\n").unwrap();
git(&["checkout", &branch]);
std::fs::write(root.join("feature.txt"), "phase work\n").unwrap();
git(&["add", "."]);
git(&["commit", "-q", "-m", "phase work"]);
git(&["checkout", "develop"]);
let state = State::new(phase, AgentKind::Claude, Mode::Auto, root.to_path_buf());
let succeeded = run_checkout_hooks(root, &state, &hooks::hooks_after_ship(), Stage::Ship);
assert!(
succeeded,
"after-ship batch must succeed against a clean repo"
);
let all_tags = devflow_core::test_support::git_command(root)
.arg("tag")
.output()
.unwrap();
let all_tags = String::from_utf8_lossy(&all_tags.stdout);
assert_eq!(all_tags.lines().count(), 1, "expected exactly one tag");
let tag = all_tags.trim().to_string();
let tag_version = tag
.strip_prefix('v')
.expect("tag should be prefixed with v")
.to_string();
let changelog = std::fs::read_to_string(root.join("CHANGELOG.md")).unwrap();
let changelog_version = changelog
.lines()
.find(|l| l.starts_with("## "))
.and_then(|l| l.trim_start_matches("## ").split(' ').next())
.unwrap()
.to_string();
assert_ne!(
changelog_version, "unreleased",
"changelog heading must name the tagged version, not fall back to the literal"
);
assert_eq!(
changelog_version, tag_version,
"changelog heading must match the git tag ({tag}) produced by the same \
run_checkout_hooks call, even with no version file"
);
}
#[test]
fn validate_failure_threshold_forces_gate_then_aborts() {
let dir = tempfile::tempdir().unwrap();
let root = dir.path();
let phase = 22;
let mut state = State::new(phase, AgentKind::Claude, Mode::Auto, root.to_path_buf());
state.stage = Stage::Validate;
state.consecutive_failures = mode::MAX_CONSECUTIVE_FAILURES - 1;
workflow::save_state(&state).unwrap();
let response_path = Gates::response_path(root, phase, Stage::Validate);
std::fs::create_dir_all(response_path.parent().unwrap()).unwrap();
std::fs::write(
&response_path,
r#"{"approved":false,"note":"abort: requirements changed","responded_by":"test"}"#,
)
.unwrap();
handle_validate_outcome(root, &mut state, ValidateOutcome::Failed).unwrap();
assert_eq!(state.consecutive_failures, mode::MAX_CONSECUTIVE_FAILURES);
assert!(
!Gates::gate_path(root, phase, Stage::Validate).exists(),
"forced gate's files must be cleaned up once it resolves to Abort"
);
let err = workflow::load_state(root, phase).unwrap_err();
assert!(matches!(err, workflow::WorkflowError::MissingState(_)));
}
fn drive_validate_advance_and_read_gate_context(
root: &Path,
phase: u32,
consecutive_failures: u32,
verdict_json: Option<&str>,
) -> String {
let mut state = State::new(phase, AgentKind::Claude, Mode::Auto, root.to_path_buf());
state.stage = Stage::Validate;
state.consecutive_failures = consecutive_failures;
workflow::save_state(&state).unwrap();
std::fs::create_dir_all(root.join(".devflow")).unwrap();
let marker = match verdict_json {
Some(verdict) => {
format!(r#"DEVFLOW_RESULT: {{"status":"success","verdict":"{verdict}"}}"#)
}
None => r#"DEVFLOW_RESULT: {"status":"success"}"#.to_string(),
};
std::fs::write(agent_result::stdout_path(root, phase), marker).unwrap();
let gate_path = Gates::gate_path(root, phase, Stage::Validate);
let response_path = Gates::response_path(root, phase, Stage::Validate);
let mut context = String::new();
std::thread::scope(|scope| {
scope.spawn(|| {
advance(root, Some(phase)).unwrap();
});
let mut seen = false;
for _ in 0..150 {
if gate_path.exists() {
seen = true;
break;
}
std::thread::sleep(std::time::Duration::from_millis(20));
}
assert!(
seen,
"advance() must force a Validate gate, not advance silently"
);
context = std::fs::read_to_string(&gate_path).unwrap();
std::fs::create_dir_all(response_path.parent().unwrap()).unwrap();
std::fs::write(
&response_path,
r#"{"approved":false,"note":"abort: test cleanup","responded_by":"test"}"#,
)
.unwrap();
});
context
}
#[test]
fn validate_gaps_does_not_advance_to_ship() {
let dir = tempfile::tempdir().unwrap();
let root = dir.path();
let context = drive_validate_advance_and_read_gate_context(
root,
60,
mode::MAX_CONSECUTIVE_FAILURES - 1,
Some("gaps"),
);
assert!(
context.contains("Validation failed"),
"a gaps verdict must be treated as a failed validation, not a pass: {context}"
);
}
#[test]
fn validate_missing_verdict_does_not_advance() {
let dir = tempfile::tempdir().unwrap();
let root = dir.path();
let context = drive_validate_advance_and_read_gate_context(
root,
61,
mode::MAX_CONSECUTIVE_FAILURES - 1,
None,
);
assert!(
context.contains("Validation failed"),
"a missing verdict must be treated as a failed validation, not a pass: {context}"
);
}
#[test]
fn validate_pass_advances() {
let dir = tempfile::tempdir().unwrap();
let root = dir.path();
let context = drive_validate_advance_and_read_gate_context(
root,
62,
mode::MAX_CONSECUTIVE_FAILURES,
Some("pass"),
);
assert!(
context.contains("Validation passed"),
"an explicit pass verdict must advance to Ship: {context}"
);
}
#[test]
fn external_verify_agreement_advances_to_ship() {
let _guard = ENV_MUTEX.lock().unwrap();
let dir = tempfile::tempdir().unwrap();
let root = dir.path();
let phase = 90;
let mut state = State::new(phase, AgentKind::Claude, Mode::Auto, root.to_path_buf());
state.stage = Stage::Validate;
workflow::save_state(&state).unwrap();
let result = agent_result::AgentResult {
status: AgentStatus::Success,
exit_code: None,
reason: None,
commits: None,
summary: None,
verdict: Some(Verdict::Pass),
decided_by_layer: Some(0),
};
let outcome = classify_validate_outcome(&result);
assert_eq!(outcome, ValidateOutcome::Passed);
let neutral_path_dir = agent_free_git_only_path_dir();
let original_path = std::env::var_os("PATH");
unsafe {
std::env::set_var("PATH", neutral_path_dir.path());
}
let _ = handle_validate_outcome(root, &mut state, outcome);
unsafe {
match &original_path {
Some(path) => std::env::set_var("PATH", path),
None => std::env::remove_var("PATH"),
}
}
assert_eq!(state.stage, Stage::Ship);
assert_eq!(
state.consecutive_failures, 0,
"an agreeing outcome must never touch the failure counter"
);
}
#[test]
fn external_verify_disagreement_gates_immediately() {
let dir = tempfile::tempdir().unwrap();
let root = dir.path();
let phase = 91;
let mut state = State::new(phase, AgentKind::Claude, Mode::Auto, root.to_path_buf());
state.stage = Stage::Validate;
workflow::save_state(&state).unwrap();
let result = agent_result::AgentResult {
status: AgentStatus::Success,
exit_code: None,
reason: None,
commits: None,
summary: None,
verdict: Some(Verdict::Gaps),
decided_by_layer: Some(0),
};
let outcome = classify_validate_outcome(&result);
assert!(matches!(outcome, ValidateOutcome::Ambiguous(_)));
let response_path = Gates::response_path(root, phase, Stage::Validate);
std::fs::create_dir_all(response_path.parent().unwrap()).unwrap();
std::fs::write(
&response_path,
r#"{"approved":false,"note":"abort: test cleanup","responded_by":"test"}"#,
)
.unwrap();
handle_validate_outcome(root, &mut state, outcome).unwrap();
assert_eq!(
state.consecutive_failures, 0,
"an ambiguous outcome must gate on cycle one without touching the counter"
);
assert!(
!Gates::gate_path(root, phase, Stage::Validate).exists(),
"the immediate gate must resolve (and clean up) via the same abort path as any other gate"
);
}
#[test]
fn external_verify_no_verdict_gates_immediately() {
let dir = tempfile::tempdir().unwrap();
let root = dir.path();
let phase = 92;
let mut state = State::new(phase, AgentKind::Claude, Mode::Auto, root.to_path_buf());
state.stage = Stage::Validate;
workflow::save_state(&state).unwrap();
let result = agent_result::AgentResult {
status: AgentStatus::Success,
exit_code: None,
reason: None,
commits: None,
summary: None,
verdict: None,
decided_by_layer: Some(0),
};
let outcome = classify_validate_outcome(&result);
assert!(matches!(outcome, ValidateOutcome::Ambiguous(_)));
let response_path = Gates::response_path(root, phase, Stage::Validate);
std::fs::create_dir_all(response_path.parent().unwrap()).unwrap();
std::fs::write(
&response_path,
r#"{"approved":false,"note":"abort: test cleanup","responded_by":"test"}"#,
)
.unwrap();
handle_validate_outcome(root, &mut state, outcome).unwrap();
assert_eq!(
state.consecutive_failures, 0,
"an ambiguous outcome must gate on cycle one without touching the counter"
);
}
#[test]
fn resource_killed_on_code_bumps_infra_failures_not_consecutive_failures() {
let dir = tempfile::tempdir().unwrap();
let root = dir.path();
let phase = 73;
std::fs::create_dir_all(root.join(".devflow")).unwrap();
std::fs::write(agent_result::exit_code_path(root, phase), "137").unwrap();
let mut state = State::new(phase, AgentKind::Claude, Mode::Auto, root.to_path_buf());
state.stage = Stage::Code;
state.consecutive_failures = 1;
workflow::save_state(&state).unwrap();
let response_path = Gates::response_path(root, phase, Stage::Code);
std::fs::create_dir_all(response_path.parent().unwrap()).unwrap();
std::fs::write(
&response_path,
r#"{"approved":false,"note":"abort: test cleanup","responded_by":"test"}"#,
)
.unwrap();
advance(root, Some(phase)).unwrap();
let err = workflow::load_state(root, phase).unwrap_err();
assert!(matches!(err, workflow::WorkflowError::MissingState(_)));
assert!(!Gates::gate_path(root, phase, Stage::Validate).exists());
}
#[test]
fn resource_killed_on_validate_bumps_infra_not_consecutive_failures() {
let dir = tempfile::tempdir().unwrap();
let root = dir.path();
let phase = 74;
let mut state = State::new(phase, AgentKind::Claude, Mode::Auto, root.to_path_buf());
state.stage = Stage::Validate;
state.consecutive_failures = 2;
workflow::save_state(&state).unwrap();
let response_path = Gates::response_path(root, phase, Stage::Validate);
std::fs::create_dir_all(response_path.parent().unwrap()).unwrap();
std::fs::write(
&response_path,
r#"{"approved":false,"note":"abort: test cleanup","responded_by":"test"}"#,
)
.unwrap();
handle_infra_outcome(
root,
&mut state,
Stage::Validate,
Some("agent process was killed (exit code 137, likely OOM)".into()),
)
.unwrap();
assert_eq!(state.infra_failures, 1);
assert_eq!(
state.consecutive_failures, 2,
"consecutive_failures must be untouched by the infra path"
);
}
#[test]
fn infra_ceiling_aborts_instead_of_gating() {
let dir = tempfile::tempdir().unwrap();
let root = dir.path();
let phase = 75;
let mut state = State::new(phase, AgentKind::Claude, Mode::Auto, root.to_path_buf());
state.stage = Stage::Code;
state.infra_failures = mode::MAX_INFRA_FAILURES - 1;
workflow::save_state(&state).unwrap();
handle_infra_outcome(root, &mut state, Stage::Code, Some("killed".into())).unwrap();
assert_eq!(state.infra_failures, mode::MAX_INFRA_FAILURES);
assert!(
!Gates::gate_path(root, phase, Stage::Code).exists(),
"at the ceiling, the run must abort rather than gate again"
);
let err = workflow::load_state(root, phase).unwrap_err();
assert!(matches!(err, workflow::WorkflowError::MissingState(_)));
}
#[test]
fn consecutive_failures_reaches_ceiling_across_cycles() {
let _guard = ENV_MUTEX.lock().unwrap();
let dir = tempfile::tempdir().unwrap();
let root = dir.path();
let phase = 81;
let mut state = State::new(phase, AgentKind::Claude, Mode::Auto, root.to_path_buf());
state.stage = Stage::Code;
workflow::save_state(&state).unwrap();
let response_path = Gates::response_path(root, phase, Stage::Validate);
std::fs::create_dir_all(response_path.parent().unwrap()).unwrap();
let neutral_path_dir = agent_free_git_only_path_dir();
let original_path = std::env::var_os("PATH");
unsafe {
std::env::set_var("PATH", neutral_path_dir.path());
}
for _ in 0..mode::MAX_CONSECUTIVE_FAILURES {
std::fs::write(
&response_path,
r#"{"approved":false,"note":"abort: test cleanup","responded_by":"test"}"#,
)
.unwrap();
let _ = handle_validate_outcome(root, &mut state, ValidateOutcome::Failed);
state.stage = Stage::Code;
let _ = transition(root, &mut state, Stage::Validate);
}
unsafe {
match &original_path {
Some(path) => std::env::set_var("PATH", path),
None => std::env::remove_var("PATH"),
}
}
assert_eq!(state.consecutive_failures, mode::MAX_CONSECUTIVE_FAILURES);
assert!(
state
.mode
.should_gate(Stage::Validate, state.consecutive_failures),
"reaching the ceiling must force the Auto-mode Validate gate"
);
assert_eq!(
state.infra_failures, 0,
"infra_failures must still reset unconditionally on the same hop the consecutive reset now skips"
);
}
#[test]
fn external_verify_cycles_reach_ceiling_without_unbounded_loop() {
let dir = tempfile::tempdir().unwrap();
let root = dir.path();
arm_a_ambiguous_outcome_gates_on_cycle_one(root, 93);
arm_b_genuine_failures_reach_the_ceiling(root, 94);
}
fn arm_a_ambiguous_outcome_gates_on_cycle_one(root: &Path, phase: u32) {
let mut state = State::new(phase, AgentKind::Claude, Mode::Auto, root.to_path_buf());
state.stage = Stage::Validate;
workflow::save_state(&state).unwrap();
let result = agent_result::AgentResult {
status: AgentStatus::Success,
exit_code: None,
reason: None,
commits: None,
summary: None,
verdict: Some(Verdict::Gaps),
decided_by_layer: Some(0),
};
let outcome = classify_validate_outcome(&result);
assert!(matches!(outcome, ValidateOutcome::Ambiguous(_)));
let response_path = Gates::response_path(root, phase, Stage::Validate);
std::fs::create_dir_all(response_path.parent().unwrap()).unwrap();
std::fs::write(
&response_path,
r#"{"approved":false,"note":"abort: test cleanup","responded_by":"test"}"#,
)
.unwrap();
handle_validate_outcome(root, &mut state, outcome).unwrap();
assert_eq!(
state.consecutive_failures, 0,
"18e's ambiguous gate must fire on cycle one, never touching 18d's counter"
);
}
fn arm_b_genuine_failures_reach_the_ceiling(root: &Path, phase: u32) {
let _guard = ENV_MUTEX.lock().unwrap();
let mut state = State::new(phase, AgentKind::Claude, Mode::Auto, root.to_path_buf());
state.stage = Stage::Code;
workflow::save_state(&state).unwrap();
let response_path = Gates::response_path(root, phase, Stage::Validate);
std::fs::create_dir_all(response_path.parent().unwrap()).unwrap();
let neutral_path_dir = agent_free_git_only_path_dir();
let original_path = std::env::var_os("PATH");
unsafe {
std::env::set_var("PATH", neutral_path_dir.path());
}
for _ in 0..mode::MAX_CONSECUTIVE_FAILURES {
std::fs::write(
&response_path,
r#"{"approved":false,"note":"abort: test cleanup","responded_by":"test"}"#,
)
.unwrap();
let _ = handle_validate_outcome(root, &mut state, ValidateOutcome::Failed);
state.stage = Stage::Code;
let _ = transition(root, &mut state, Stage::Validate);
}
unsafe {
match &original_path {
Some(path) => std::env::set_var("PATH", path),
None => std::env::remove_var("PATH"),
}
}
assert_eq!(state.consecutive_failures, mode::MAX_CONSECUTIVE_FAILURES);
assert!(
state
.mode
.should_gate(Stage::Validate, state.consecutive_failures),
"a genuine repeated failure must still reach the reachable ceiling (18d)"
);
}
#[test]
fn consecutive_failures_increment_saturates() {
let dir = tempfile::tempdir().unwrap();
let root = dir.path();
let phase = 82;
let mut state = State::new(phase, AgentKind::Claude, Mode::Auto, root.to_path_buf());
state.stage = Stage::Validate;
state.consecutive_failures = u32::MAX;
workflow::save_state(&state).unwrap();
let response_path = Gates::response_path(root, phase, Stage::Validate);
std::fs::create_dir_all(response_path.parent().unwrap()).unwrap();
std::fs::write(
&response_path,
r#"{"approved":false,"note":"abort: test cleanup","responded_by":"test"}"#,
)
.unwrap();
handle_validate_outcome(root, &mut state, ValidateOutcome::Failed).unwrap();
assert_eq!(state.consecutive_failures, u32::MAX);
}
#[test]
fn primary_loop_rate_limited_writes_single_agent_cron_instructions() {
let dir = tempfile::tempdir().unwrap();
let root = dir.path();
let phase = 76;
let mut state = State::new(phase, AgentKind::Claude, Mode::Auto, root.to_path_buf());
state.stage = Stage::Code;
workflow::save_state(&state).unwrap();
std::fs::create_dir_all(root.join(".devflow")).unwrap();
std::fs::write(
agent_result::stdout_path(root, phase),
r#"{"type":"result","subtype":"error_rate_limit","retry_after":"2026-06-18T15:45:30Z"}"#,
)
.unwrap();
advance(root, Some(phase)).unwrap();
let instructions = devflow_core::ship::load_cron_instructions(root, phase).unwrap();
assert_eq!(instructions.resume.command, "devflow");
assert_eq!(
instructions.resume.args,
["resume", "--phase", &phase.to_string()]
);
assert!(
instructions
.hermes_cron
.command
.contains(&format!("devflow resume --phase {phase}"))
);
let reloaded = workflow::load_state(root, phase).unwrap();
assert_eq!(reloaded.stage, Stage::Code);
assert!(!reloaded.gate_pending);
assert_eq!(reloaded.infra_failures, 1);
assert_eq!(reloaded.consecutive_failures, 0);
assert!(!Gates::gate_path(root, phase, Stage::Code).exists());
}
#[test]
fn rate_limited_at_infra_ceiling_stops_resuming_and_aborts() {
let dir = tempfile::tempdir().unwrap();
let root = dir.path();
let phase = 77;
let mut state = State::new(phase, AgentKind::Claude, Mode::Auto, root.to_path_buf());
state.stage = Stage::Code;
state.infra_failures = mode::MAX_INFRA_FAILURES - 1;
workflow::save_state(&state).unwrap();
std::fs::create_dir_all(root.join(".devflow")).unwrap();
std::fs::write(
agent_result::stdout_path(root, phase),
r#"{"type":"result","subtype":"error_rate_limit","retry_after":"2026-06-18T15:45:30Z"}"#,
)
.unwrap();
advance(root, Some(phase)).unwrap();
let err = workflow::load_state(root, phase).unwrap_err();
assert!(
matches!(err, workflow::WorkflowError::MissingState(_)),
"the infra ceiling must abort, clearing state"
);
assert!(
devflow_core::ship::load_cron_instructions(root, phase).is_err(),
"must not schedule an auto-resume once the infra ceiling stops resumption"
);
}
#[test]
fn rate_limited_with_unparseable_retry_hint_gates_instead_of_stalling_silently() {
let dir = tempfile::tempdir().unwrap();
let root = dir.path();
let phase = 81;
let mut state = State::new(phase, AgentKind::Claude, Mode::Auto, root.to_path_buf());
state.stage = Stage::Code;
workflow::save_state(&state).unwrap();
std::fs::create_dir_all(root.join(".devflow")).unwrap();
let response_path = Gates::response_path(root, phase, Stage::Code);
std::fs::create_dir_all(response_path.parent().unwrap()).unwrap();
std::fs::write(
&response_path,
r#"{"approved":false,"note":"abort: test cleanup","responded_by":"test"}"#,
)
.unwrap();
handle_rate_limited_outcome(
root,
&mut state,
phase,
Stage::Code,
Some("rate limited until usage limit".into()),
)
.unwrap();
let events =
std::fs::read_to_string(devflow_core::events::events_path(root)).unwrap_or_default();
assert!(
events.contains("gate_fired"),
"an unparseable retry hint must raise a gate, not stall the phase silently: {events}"
);
assert!(
events.contains("notify_fired"),
"the operator must be notified that a manual resume is needed: {events}"
);
assert!(
!events.contains("rate_limit_resume_scheduled"),
"nothing was scheduled — emitting a resume-scheduled event would be a false signal: {events}"
);
let instructions = devflow_core::ship::load_cron_instructions(root, phase).unwrap();
assert!(instructions.hermes_cron.schedule.is_empty());
}
#[test]
fn content_hooks_target_the_worktree_while_terminal_hooks_stay_on_project_root() {
let dir = tempfile::tempdir().unwrap();
let root = dir.path();
let worktree = root.join(".worktrees/phase-70");
std::fs::create_dir_all(&worktree).unwrap();
let mut state = State::new(70, AgentKind::Claude, Mode::Auto, root.to_path_buf());
state.worktree_path = Some(worktree.clone());
assert_eq!(
hook_context_root(root, &state, false),
worktree,
"content hooks must write into the phase's worktree"
);
assert_eq!(
hook_context_root(root, &state, true),
root.to_path_buf(),
"terminal hooks merge/tag/delete against the primary checkout"
);
let mut no_worktree = state.clone();
no_worktree.worktree_path = None;
assert_eq!(hook_context_root(root, &no_worktree, false), root);
let mut missing = state.clone();
missing.worktree_path = Some(root.join(".worktrees/gone"));
assert_eq!(hook_context_root(root, &missing, false), root);
}
#[test]
fn truncate_reason_caps_long_reasons_and_keeps_short_ones() {
assert_eq!(truncate_reason("short reason"), "short reason");
let long = "x".repeat(5000);
let capped = truncate_reason(&long);
assert!(capped.chars().count() <= 300);
assert!(capped.ends_with("[truncated; full output in .devflow/]"));
}
#[test]
fn gate_context_rendering_neutralizes_all_controls_and_obeys_limit() {
let rendered = render_gate_context("line 1\n\u{1b}[2J\tline 2\u{7}", 100);
assert!(!rendered.chars().any(char::is_control));
assert_eq!(rendered, "line 1 [2J line 2 ");
let bounded = render_gate_context(&"x".repeat(500), 100);
assert_eq!(bounded.chars().count(), 100);
assert!(bounded.ends_with("[truncated; full output in .devflow/]"));
}
#[test]
fn ship_agent_failed_fires_gate() {
let dir = tempfile::tempdir().unwrap();
let root = dir.path();
let phase = 40;
let mut state = State::new(phase, AgentKind::Claude, Mode::Auto, root.to_path_buf());
state.stage = Stage::Ship;
workflow::save_state(&state).unwrap();
let gate_path = Gates::gate_path(root, phase, Stage::Ship);
let response_path = Gates::response_path(root, phase, Stage::Ship);
std::thread::scope(|scope| {
scope.spawn(|| {
handle_ship_failure(root, &mut state, Some("agent crashed".into())).unwrap();
});
let mut seen = false;
for _ in 0..150 {
if gate_path.exists() {
seen = true;
break;
}
std::thread::sleep(std::time::Duration::from_millis(20));
}
assert!(
seen,
"handle_ship_failure must write a gate file, not silently return an Err"
);
std::fs::create_dir_all(response_path.parent().unwrap()).unwrap();
std::fs::write(
&response_path,
r#"{"approved":false,"note":"abort: test cleanup","responded_by":"test"}"#,
)
.unwrap();
});
}
#[test]
fn ship_review_failed_loops_to_code() {
let dir = tempfile::tempdir().unwrap();
let root = dir.path();
let phase = 41;
let mut state = State::new(phase, AgentKind::Claude, Mode::Auto, root.to_path_buf());
state.stage = Stage::Ship;
workflow::save_state(&state).unwrap();
let reason = Some("review: please fix naming".to_string());
assert!(is_ship_review_failure(&reason));
prepare_loop_back_to_code(root, &mut state, FixType::AuditFix).unwrap();
assert_eq!(state.stage, Stage::Code);
assert!(!Gates::gate_path(root, phase, Stage::Ship).exists());
assert!(workflow::load_state(root, phase).is_ok());
}
#[test]
fn ship_review_failed_uses_audit_fix() {
assert!(is_ship_review_failure(&Some(
"review: needs changes".into()
)));
assert!(is_ship_review_failure(&Some(" Review: nitpick".into())));
assert!(!is_ship_review_failure(&Some("agent crashed".into())));
assert!(!is_ship_review_failure(&None));
let prompt = prompt::fix_prompt(FixType::AuditFix, 11);
assert!(prompt.contains("/gsd-audit-fix"));
assert!(!prompt.contains("--gaps-only"));
}
#[test]
fn non_validate_failure_fires_gate_and_hook() {
let _guard = ENV_MUTEX.lock().unwrap();
let dir = tempfile::tempdir().unwrap();
let root = dir.path();
let sentinel = root.join("notify-sentinel");
unsafe {
std::env::set_var(
"DEVFLOW_GATE_NOTIFY_CMD",
format!("touch {}", sentinel.display()),
);
}
let phase = 42;
let mut state = State::new(phase, AgentKind::Claude, Mode::Auto, root.to_path_buf());
state.stage = Stage::Code;
workflow::save_state(&state).unwrap();
assert!(
!state
.mode
.should_gate(Stage::Code, state.consecutive_failures)
);
let response_path = Gates::response_path(root, phase, Stage::Code);
std::fs::create_dir_all(response_path.parent().unwrap()).unwrap();
std::fs::write(
&response_path,
r#"{"approved":false,"note":"abort: test cleanup","responded_by":"test"}"#,
)
.unwrap();
let result =
handle_stage_failure(root, &mut state, Stage::Code, Some("build failed".into()));
unsafe {
std::env::remove_var("DEVFLOW_GATE_NOTIFY_CMD");
}
result.unwrap();
assert!(
sentinel.exists(),
"handle_stage_failure must fire the configured notify hook, not silently skip it"
);
}
#[test]
fn stage_failure_retry_cleans_stale_response() {
let dir = tempfile::tempdir().unwrap();
let root = dir.path();
let phase = 43;
let mut state = State::new(phase, AgentKind::Claude, Mode::Auto, root.to_path_buf());
state.stage = Stage::Code;
workflow::save_state(&state).unwrap();
let response_path = Gates::response_path(root, phase, Stage::Code);
std::fs::create_dir_all(response_path.parent().unwrap()).unwrap();
std::fs::write(
&response_path,
r#"{"approved":false,"note":"abort: test cleanup","responded_by":"test"}"#,
)
.unwrap();
handle_stage_failure(root, &mut state, Stage::Code, Some("first failure".into())).unwrap();
assert!(!Gates::gate_path(root, phase, Stage::Code).exists());
assert!(!Gates::response_path(root, phase, Stage::Code).exists());
assert!(!Gates::ack_path(root, phase, Stage::Code).exists());
Gates::write_gate(root, phase, Stage::Code, "re-fired gate").unwrap();
let started = std::time::Instant::now();
let got = Gates::poll_response(root, phase, Stage::Code, 1);
assert!(
got.is_none(),
"poll_response must not instantly resolve from a stale response after cleanup"
);
assert!(started.elapsed() >= std::time::Duration::from_secs(1));
}
}