use crate::config;
use crate::parser::AgentResponse;
use crate::store;
use crate::store::TaskStore;
use std::path::Path;
use std::sync::Arc;
use super::{agents, git_ops, response, worktree};
#[allow(clippy::too_many_arguments)]
pub async fn write_result_json(
attempt_dir: &Path,
task_id: &str,
agent_name: &str,
model_name: Option<&str>,
exit_code: i32,
new_attempts: u32,
parse_result: &Result<agents::ParsedResponse, agents::AgentError>,
raw_stdout: &str,
raw_stderr: &str,
) {
let result_json = match parse_result {
Ok(parsed) => {
serde_json::json!({
"outcome": "success",
"agent": agent_name,
"model": model_name.unwrap_or("default"),
"exit_code": exit_code,
"attempt": new_attempts,
"status": parsed.response.status,
"summary": parsed.response.summary,
"input_tokens": parsed.input_tokens,
"output_tokens": parsed.output_tokens,
"duration_ms": parsed.duration_ms,
"files": parsed.response.files,
"accomplished": parsed.response.accomplished,
"remaining": parsed.response.remaining,
"error": parsed.response.error,
"learnings": parsed.response.learnings,
"delegations": parsed.response.delegations.iter()
.map(|d| serde_json::json!({"title": d.title, "body": d.body}))
.collect::<Vec<_>>(),
})
}
Err(agent_err) => {
serde_json::json!({
"outcome": "error",
"agent": agent_name,
"model": model_name.unwrap_or("default"),
"exit_code": exit_code,
"attempt": new_attempts,
"error_class": agents::error_class_name(agent_err),
"error_message": agent_err.to_string(),
"stderr_tail": agents::patterns::safe_tail(raw_stderr, 2000),
"stdout_tail": agents::patterns::safe_tail(raw_stdout, 2000),
})
}
};
if let Err(e) = tokio::fs::write(
attempt_dir.join("result.json"),
serde_json::to_string_pretty(&result_json).unwrap_or_default(),
)
.await
{
tracing::debug!(task_id, ?e, "failed to write result.json");
}
}
#[derive(Default)]
struct DecisionInput<'a> {
agent_status: &'a str,
is_workflow_scope_failure: bool,
is_rebase_conflict_failure: bool,
push_failed: bool,
push_failures: u64,
has_pr: bool,
has_delegations: bool,
has_pushed: bool,
is_pr_base_invalid: bool,
requires_pr: bool,
no_code_reroutes: u64,
max_reroutes: u32,
is_same_agent: bool,
is_completed_status: bool,
is_retryable_blocked: bool,
is_superseded_conflict: bool,
}
fn classify_final_status(input: &DecisionInput<'_>) -> String {
if input.is_workflow_scope_failure || input.is_rebase_conflict_failure {
"blocked".to_string()
} else if input.push_failed {
if input.push_failures >= 3 {
"blocked".to_string()
} else {
"new".to_string()
}
} else if input.agent_status == "done" && input.has_pr {
"needs_review".to_string()
} else if input.agent_status == "done" && !input.has_pr && input.has_delegations {
"blocked".to_string()
} else if input.agent_status == "done"
&& !input.has_pr
&& input.has_pushed
&& input.is_pr_base_invalid
{
"blocked".to_string()
} else if input.agent_status == "done" && !input.has_pr && input.has_pushed {
"routed".to_string()
} else if input.is_superseded_conflict {
"done".to_string()
} else if input.agent_status == "done" && input.requires_pr {
if input.is_same_agent || input.no_code_reroutes >= input.max_reroutes as u64 {
"blocked".to_string()
} else {
"new".to_string()
}
} else if input.agent_status == "done" && !input.has_pushed {
"done".to_string()
} else if input.is_completed_status && input.has_pr {
"needs_review".to_string()
} else if input.is_completed_status && !input.has_pr && input.has_delegations {
"blocked".to_string()
} else if input.is_completed_status
&& !input.has_pr
&& input.has_pushed
&& input.is_pr_base_invalid
{
"blocked".to_string()
} else if input.is_completed_status && !input.has_pr && input.has_pushed {
"routed".to_string()
} else if input.is_completed_status && input.requires_pr {
if input.is_same_agent || input.no_code_reroutes >= input.max_reroutes as u64 {
"blocked".to_string()
} else {
"new".to_string()
}
} else if input.is_completed_status {
"done".to_string()
} else if input.agent_status == "blocked" && input.is_retryable_blocked {
"new".to_string()
} else if crate::status_heuristics::status_looks_like_descriptive_completion(input.agent_status)
{
"done".to_string()
} else {
input.agent_status.to_string()
}
}
struct StoreCtx<'a> {
store: &'a Option<Arc<TaskStore>>,
store_id_opt: Option<i64>,
repo: &'a str,
task_id: &'a str,
}
impl<'a> StoreCtx<'a> {
async fn set(&self, fields: &[(&str, serde_json::Value)]) {
if let Some(store_id) = self.store_id_opt {
let _ = store::store_set_by_id(&self.store.as_ref(), store_id, fields).await;
} else {
store::store_set(self.store, self.repo, self.task_id, fields).await;
}
}
async fn get_task(&self) -> Option<crate::store::Task> {
if let Some(store_id) = self.store_id_opt {
store::opt_store_get_task_by_id(self.store, store_id).await
} else {
store::opt_store_get_task(self.store, self.repo, self.task_id).await
}
}
async fn increment(&self, field: &str) -> anyhow::Result<u64> {
if let Some(store_id) = self.store_id_opt {
store::store_increment_by_id(self.store, store_id, field).await
} else {
store::store_increment(self.store, self.repo, self.task_id, field).await
}
}
async fn append_activity(
&self,
event_type: &str,
agent: Option<&str>,
model: Option<&str>,
details: Option<&serde_json::Value>,
) {
if let Some(store_id) = self.store_id_opt {
if let Some(ref s) = self.store {
if let Err(e) = s
.append_activity(store_id, event_type, None, None, agent, model, details)
.await
{
tracing::warn!(
task_id = self.task_id,
event_type,
error = %e,
"store append_activity failed"
);
}
}
} else {
store::store_log_activity(
self.store,
self.repo,
self.task_id,
event_type,
None,
None,
agent,
model,
details,
)
.await;
}
}
}
#[derive(Default)]
struct GitOpsResult {
has_pr: bool,
has_pushed: bool,
has_commits: bool,
pr_base_invalid: bool,
}
struct PushFailureState {
push_failed: bool,
is_workflow_scope_failure: bool,
is_rebase_conflict_failure: bool,
push_failures: u64,
stored_last_error: String,
}
async fn auto_commit_changes(
ctx: &StoreCtx<'_>,
work_dir: &Path,
task_title: &str,
agent_name: &str,
new_attempts: u32,
) {
if let Err(e) =
git_ops::auto_commit(work_dir, ctx.task_id, task_title, agent_name, new_attempts).await
{
tracing::error!(task_id = ctx.task_id, error = ?e, "auto commit failed");
ctx.set(&[(
"last_error",
serde_json::json!(format!("auto commit failed: {e}")),
)])
.await;
}
}
async fn check_commits_and_clear_stale_errors(
ctx: &StoreCtx<'_>,
work_dir: &Path,
default_branch: &str,
) -> bool {
let has_commits = git_ops::has_commits_ahead(work_dir, default_branch).await;
if !has_commits {
tracing::info!(
task_id = ctx.task_id,
"no commits ahead of default branch, skipping push + PR"
);
let last_err = ctx
.get_task()
.await
.map(|t| t.last_error)
.unwrap_or_default();
if last_err.contains("push failed") {
ctx.set(&[("last_error", serde_json::json!(""))]).await;
}
}
has_commits
}
enum PushResult {
Success,
Failed,
NoCommits, WorktreeMissing, }
fn is_missing_path_error(err: &anyhow::Error) -> bool {
err.chain().any(|cause| {
cause
.downcast_ref::<std::io::Error>()
.is_some_and(|ioe| ioe.kind() == std::io::ErrorKind::NotFound)
}) || err
.to_string()
.to_ascii_lowercase()
.contains("no such file or directory")
}
async fn recover_missing_worktree_for_push(
ctx: &StoreCtx<'_>,
wt: &worktree::WorktreeSetup,
task_title: &str,
agent_name: &str,
model_name: Option<&str>,
) -> anyhow::Result<worktree::WorktreeSetup> {
tracing::warn!(
task_id = ctx.task_id,
worktree = %wt.work_dir.display(),
branch = wt.branch,
"worktree missing during push; attempting one-time recovery"
);
ctx.append_activity(
"push_recover_worktree",
Some(agent_name),
model_name,
Some(&serde_json::json!({
"status": "start",
"worktree": wt.work_dir.display().to_string(),
"branch": wt.branch,
})),
)
.await;
let recovered = worktree::setup_worktree(
ctx.task_id,
task_title,
&wt.main_project_dir,
ctx.store,
ctx.repo,
)
.await?;
if !recovered.work_dir.exists() {
anyhow::bail!(
"recovered worktree path is still missing: {}",
recovered.work_dir.display()
);
}
ctx.append_activity(
"push_recover_worktree",
Some(agent_name),
model_name,
Some(&serde_json::json!({
"status": "ok",
"worktree": recovered.work_dir.display().to_string(),
"branch": recovered.branch,
})),
)
.await;
Ok(recovered)
}
async fn push_after_rebase(
ctx: &StoreCtx<'_>,
work_dir: &Path,
branch: &str,
default_branch: &str,
agent_name: &str,
model_name: Option<&str>,
) -> PushResult {
match git_ops::push_branch(work_dir, branch, default_branch).await {
Ok(_) => {
ctx.append_activity(
"push",
Some(agent_name),
model_name,
Some(&serde_json::json!({
"status": "ok",
"branch": branch,
"default_branch": default_branch,
"reason": "pushed_after_rebase",
})),
)
.await;
ctx.set(&[
("last_error", serde_json::json!("")),
("push_failures", serde_json::json!(0)),
])
.await;
PushResult::Success
}
Err(e) => {
tracing::error!(
task_id = ctx.task_id,
branch = branch,
error = ?e,
"push failed again after rebasing on the remote branch"
);
ctx.append_activity(
"push",
Some(agent_name),
model_name,
Some(&serde_json::json!({
"status": "error",
"branch": branch,
"default_branch": default_branch,
"reason": "pushed_after_rebase",
"error": e.to_string(),
})),
)
.await;
ctx.set(&[(
"last_error",
serde_json::json!(format!("push failed after rebase on remote branch: {e}")),
)])
.await;
PushResult::Failed
}
}
}
async fn push_branch_with_log(
ctx: &StoreCtx<'_>,
work_dir: &Path,
branch: &str,
default_branch: &str,
agent_name: &str,
model_name: Option<&str>,
) -> PushResult {
match git_ops::push_branch(work_dir, branch, default_branch).await {
Ok(_) => {
ctx.append_activity(
"push",
Some(agent_name),
model_name,
Some(&serde_json::json!({
"status": "ok",
"branch": branch,
"default_branch": default_branch,
})),
)
.await;
ctx.set(&[
("last_error", serde_json::json!("")),
("push_failures", serde_json::json!(0)),
])
.await;
PushResult::Success
}
Err(e) => {
let err_str = e.to_string();
tracing::error!(task_id = ctx.task_id, error = ?e, "push failed");
if is_missing_path_error(&e) && !work_dir.exists() {
tracing::warn!(
task_id = ctx.task_id,
worktree = %work_dir.display(),
"push failed because worktree path disappeared; will attempt recovery"
);
ctx.append_activity(
"push",
Some(agent_name),
model_name,
Some(&serde_json::json!({
"status": "retry",
"branch": branch,
"default_branch": default_branch,
"reason": "worktree_missing",
"error": err_str,
})),
)
.await;
return PushResult::WorktreeMissing;
}
let is_diverged = err_str.contains("non-fast-forward")
|| err_str.contains("rejected")
|| err_str.contains("fetch first")
|| err_str.contains("behind");
if is_diverged {
tracing::info!(
task_id = ctx.task_id,
branch = branch,
"attempting fetch and rebase to recover from diverged branch"
);
match git_ops::rebase_on_branch(work_dir, branch).await {
Ok(true) => {
tracing::info!(
task_id = ctx.task_id,
branch = branch,
"rebase succeeded — retrying the push"
);
ctx.append_activity(
"push_rebased",
Some(agent_name),
model_name,
Some(&serde_json::json!({
"status": "rebased",
"branch": branch,
"reason": "diverged_from_remote",
})),
)
.await;
return push_after_rebase(
ctx,
work_dir,
branch,
default_branch,
agent_name,
model_name,
)
.await;
}
Ok(false) => {
tracing::info!(
task_id = ctx.task_id,
branch = branch,
"no local commits after fetch — marking as success"
);
ctx.append_activity(
"push",
Some(agent_name),
model_name,
Some(&serde_json::json!({
"status": "ok",
"branch": branch,
"reason": "no_commits_after_fetch",
})),
)
.await;
ctx.set(&[
("last_error", serde_json::json!("")),
("push_failures", serde_json::json!(0)),
])
.await;
return PushResult::Success;
}
Err(rebase_err) => {
tracing::error!(
task_id = ctx.task_id,
branch = branch,
error = ?rebase_err,
"rebase failed — blocking task for human intervention"
);
ctx.append_activity(
"push_rebased",
Some(agent_name),
model_name,
Some(&serde_json::json!({
"status": "error",
"branch": branch,
"error": rebase_err.to_string(),
})),
)
.await;
ctx.set(&[(
"last_error",
serde_json::json!(format!(
"push failed and rebase failed: {rebase_err} — manual resolution required"
)),
)])
.await;
return PushResult::Failed;
}
}
}
ctx.append_activity(
"push",
Some(agent_name),
model_name,
Some(&serde_json::json!({
"status": "error",
"branch": branch,
"default_branch": default_branch,
"error": err_str,
})),
)
.await;
ctx.set(&[("last_error", serde_json::json!(format!("push failed: {e}")))])
.await;
PushResult::Failed
}
}
}
fn is_local_rate_limit_skip(err_str: &str) -> bool {
err_str.contains("rate-limited, skipping call")
}
async fn create_pr_with_log(
ctx: &StoreCtx<'_>,
wt: &worktree::WorktreeSetup,
task_title: &str,
resp: &AgentResponse,
agent_name: &str,
model_name: Option<&str>,
mut has_pushed: bool,
) -> (bool, bool, bool) {
match git_ops::create_pr_if_needed(
&wt.work_dir,
&wt.branch,
task_title,
&resp.summary,
&resp.accomplished,
&resp.remaining,
&resp.files,
ctx.task_id,
agent_name,
model_name,
ctx.repo,
&wt.default_branch,
)
.await
{
Ok(ref url) => {
match crate::engine::review::parse_pr_number_from_url(url) {
Ok(pr_num) => {
if let Err(e) = store::store_set_result(
ctx.store,
ctx.repo,
ctx.task_id,
&[("pr_number", serde_json::json!(pr_num as i64))],
)
.await
{
tracing::error!(
task_id = ctx.task_id,
pr_url = %url,
err = %e,
"CRITICAL: failed to save pr_number to store — downstream review gate may trigger duplicate gh pr create"
);
}
}
Err(e) => {
tracing::warn!(
task_id = ctx.task_id,
pr_url = %url,
error = %e,
"PR created but failed to parse PR number from URL"
);
}
}
ctx.append_activity(
"pr_create",
Some(agent_name),
model_name,
Some(&serde_json::json!({"status": "created", "url": url})),
)
.await;
(true, has_pushed, false)
}
Err(e) => {
let err_str = format!("{e}");
tracing::error!(task_id = ctx.task_id, error = ?e, "create PR failed");
ctx.append_activity(
"pr_create",
Some(agent_name),
model_name,
Some(&serde_json::json!({
"status": "error",
"branch": wt.branch,
"error": err_str.clone(),
})),
)
.await;
ctx.set(&[(
"last_error",
serde_json::json!(format!("create PR failed: {e}")),
)])
.await;
let mut pr_base_invalid = false;
if is_local_rate_limit_skip(&err_str) {
tracing::warn!(
task_id = ctx.task_id,
"PR creation skipped by local GitHub rate limit — review gate will create the PR"
);
return (true, has_pushed, false);
}
if agents::patterns::detect_auth_error(&err_str).is_some() {
tracing::error!(
task_id = ctx.task_id,
"PR creation failed with GitHub auth error — blocking task"
);
pr_base_invalid = true;
} else if err_str.contains("422") {
if err_str.contains("No commits between") || err_str.contains("head") {
tracing::info!(
task_id = ctx.task_id,
"PR creation returned 422/no-commits/head-invalid — agent made no code changes or branch merged, marking done"
);
has_pushed = false;
} else if git_ops::is_invalid_base_error(&err_str) {
tracing::error!(
task_id = ctx.task_id,
"PR creation failed with invalid base branch — blocking task"
);
pr_base_invalid = true;
}
}
(false, has_pushed, pr_base_invalid)
}
}
}
async fn run_git_ops(
ctx: &StoreCtx<'_>,
wt: &worktree::WorktreeSetup,
task_title: &str,
resp: &AgentResponse,
agent_name: &str,
model_name: Option<&str>,
new_attempts: u32,
) -> GitOpsResult {
auto_commit_changes(ctx, &wt.work_dir, task_title, agent_name, new_attempts).await;
let mut has_commits =
check_commits_and_clear_stale_errors(ctx, &wt.work_dir, &wt.default_branch).await;
let push_result = if !has_commits {
store::store_log_activity(
ctx.store,
ctx.repo,
ctx.task_id,
"push",
None,
None,
Some(agent_name),
model_name,
Some(&serde_json::json!({
"status": "skipped",
"reason": "no_commits_ahead",
"branch": wt.branch,
"default_branch": wt.default_branch,
})),
)
.await;
PushResult::NoCommits
} else {
push_branch_with_log(
ctx,
&wt.work_dir,
&wt.branch,
&wt.default_branch,
agent_name,
model_name,
)
.await
};
let push_result = if matches!(push_result, PushResult::WorktreeMissing) {
match recover_missing_worktree_for_push(ctx, wt, task_title, agent_name, model_name).await {
Ok(recovered) => {
let wt_for_ops = &recovered;
has_commits = check_commits_and_clear_stale_errors(
ctx,
&wt_for_ops.work_dir,
&wt_for_ops.default_branch,
)
.await;
if !has_commits {
PushResult::NoCommits
} else {
push_branch_with_log(
ctx,
&wt_for_ops.work_dir,
&wt_for_ops.branch,
&wt_for_ops.default_branch,
agent_name,
model_name,
)
.await
}
}
Err(e) => {
tracing::error!(
task_id = ctx.task_id,
err = %e,
"worktree recovery before push failed"
);
ctx.append_activity(
"push_recover_worktree",
Some(agent_name),
model_name,
Some(&serde_json::json!({
"status": "error",
"worktree": wt.work_dir.display().to_string(),
"branch": wt.branch,
"error": e.to_string(),
})),
)
.await;
ctx.set(&[(
"last_error",
serde_json::json!(format!(
"push failed: worktree disappeared and recovery failed: {e}"
)),
)])
.await;
PushResult::Failed
}
}
} else {
push_result
};
let mut has_pushed = matches!(push_result, PushResult::Success);
let mut has_pr = false;
if has_commits && matches!(push_result, PushResult::Failed) {
tracing::warn!(
task_id = ctx.task_id,
"skipping PR creation due to push failure"
);
} else if matches!(push_result, PushResult::NoCommits) {
tracing::info!(
task_id = ctx.task_id,
"no commits ahead of default branch, skipping push + PR"
);
} else if !has_pushed {
} else if ctx.repo.is_empty() {
tracing::warn!(
task_id = ctx.task_id,
"skipping PR creation — repo is empty (internal task?)"
);
} else {
let (pr, updated_pushed, pr_base_invalid) = create_pr_with_log(
ctx, wt, task_title, resp, agent_name, model_name, has_pushed,
)
.await;
has_pr = pr;
has_pushed = updated_pushed;
return GitOpsResult {
has_pr,
has_pushed,
has_commits,
pr_base_invalid,
};
}
GitOpsResult {
has_pr,
has_pushed,
has_commits,
pr_base_invalid: false,
}
}
async fn close_superseded_pr(ctx: &StoreCtx<'_>, pr_number: u64) {
if ctx.repo.is_empty() {
return;
}
let gh = match crate::github::http::GhHttp::new() {
Ok(gh) => gh,
Err(e) => {
tracing::warn!(task_id = ctx.task_id, err = %e, "cannot close superseded PR");
return;
}
};
let note = "Closing as superseded: the change is already on the default branch, so this branch has no commits ahead of it.";
if let Err(e) = gh.add_comment(ctx.repo, &pr_number.to_string(), note).await {
tracing::warn!(task_id = ctx.task_id, pr_number, err = %e, "failed to comment on superseded PR");
}
if let Err(e) = gh.close_pr(ctx.repo, pr_number).await {
tracing::warn!(task_id = ctx.task_id, pr_number, err = %e, "failed to close superseded PR");
}
ctx.set(&[(
"last_error",
serde_json::json!("PR closed as superseded: the fix already landed on the default branch"),
)])
.await;
}
fn superseded_conflict_pr(
merge_conflict_retries: i32,
pr_number: Option<i32>,
has_commits: bool,
) -> Option<u64> {
if has_commits || merge_conflict_retries <= 0 {
return None;
}
pr_number.filter(|n| *n > 0).map(|n| n as u64)
}
async fn detect_push_failure_state(
ctx: &StoreCtx<'_>,
has_commits: bool,
has_pushed: bool,
) -> PushFailureState {
let stored_last_error = ctx
.get_task()
.await
.map(|t| t.last_error)
.unwrap_or_default();
let push_failed = !has_pushed && has_commits && stored_last_error.contains("push failed");
let is_workflow_scope_failure = push_failed
&& (stored_last_error.contains("workflow` scope")
|| stored_last_error.contains("workflow' scope")
|| (stored_last_error.contains("refusing to allow")
&& stored_last_error.contains("workflow")));
let is_rebase_conflict_failure =
push_failed && stored_last_error.contains("push failed and rebase failed:");
let push_failures = if push_failed && !is_workflow_scope_failure && !is_rebase_conflict_failure
{
match ctx.increment("push_failures").await {
Ok(v) => v,
Err(e) => {
tracing::warn!(
task_id = ctx.task_id,
err = %e,
"failed to increment push_failures — reading current value from store"
);
ctx.get_task()
.await
.map(|t| (t.push_failures.max(0) as u64) + 1)
.unwrap_or(1) }
}
} else {
0
};
PushFailureState {
push_failed,
is_workflow_scope_failure,
is_rebase_conflict_failure,
push_failures,
stored_last_error,
}
}
async fn detect_no_code_reroutes(
ctx: &StoreCtx<'_>,
is_no_code_reroute: bool,
new_attempts: u32,
max_reroutes: u32,
) -> u64 {
if !is_no_code_reroute {
return 0;
}
tracing::warn!(
task_id = ctx.task_id,
attempts = new_attempts,
max_reroutes,
"agent reported done but produced no code changes on external task requiring PR"
);
match ctx.increment("no_code_reroutes").await {
Ok(v) => v,
Err(e) => {
tracing::warn!(
task_id = ctx.task_id,
err = %e,
"failed to increment no_code_reroutes — reading current value from store"
);
ctx.get_task()
.await
.map(|t| (t.no_code_reroutes.max(0) as u64) + 1)
.unwrap_or(1) }
}
}
fn agent_blocked_reason(resp: &crate::parser::AgentResponse) -> String {
let summary_opt = if !resp.summary.is_empty() {
Some(resp.summary.as_str())
} else {
None
};
let base = resp
.error
.as_deref()
.filter(|s| !s.is_empty())
.or(summary_opt)
.unwrap_or("agent returned blocked status without a reason");
if resp.remaining.is_empty() {
base.to_string()
} else {
format!("{base}. Remaining: {}", resp.remaining.join("; "))
}
}
fn is_denied_push_block(resp: &crate::parser::AgentResponse) -> bool {
let r = agent_blocked_reason(resp).to_lowercase();
r.contains("push")
&& ["denied", "not allowed", "permission", "disallowed"]
.iter()
.any(|p| r.contains(p))
}
fn is_retryable_blocked_reason(reason: &str) -> bool {
let r = reason.to_lowercase();
let permanent_patterns = [
"not found",
"does not exist",
"invalid",
"malformed",
"syntax error",
"unsupported",
"unimplemented",
"manual intervention required",
"human intervention required",
];
if permanent_patterns.iter().any(|p| r.contains(p)) {
return false;
}
let transient_patterns = [
"unavailable",
"temporarily unavailable",
"timeout",
"timed out",
"connection reset",
"connection refused",
"network error",
"rate limit",
"try again",
"service down",
"permission denied",
"worktree lock",
"git lock",
"lockfile permission",
"lock permission",
"resource busy",
"could not resolve host",
"gpg decryption fails",
"gpg decryption failed",
"gpg agent",
"no passphrase available",
"passphrase unavailable",
"pass store",
"passwordstoreprovider",
"agentvaultprovider",
"onepasswordprovider",
"decryption failed in this agent session",
];
transient_patterns.iter().any(|p| r.contains(p))
}
#[allow(clippy::too_many_arguments)]
async fn apply_post_decision_effects(
ctx: &StoreCtx<'_>,
final_status: &str,
resp_status: &str,
push_state: &PushFailureState,
is_no_code_reroute: bool,
has_pr: bool,
has_pushed: bool,
is_pr_base_invalid: bool,
has_delegations: bool,
is_external: bool,
no_code_reroutes: u64,
max_reroutes: u32,
agent_name: &str,
) {
if push_state.is_workflow_scope_failure {
tracing::error!(
task_id = ctx.task_id,
"push failed: token lacks `workflow` OAuth scope — blocking immediately \
(rerouting would not help)"
);
ctx.set(&[(
"last_error",
serde_json::json!(format!(
"push failed: GitHub token lacks `workflow` OAuth scope. \
The agent modified .github/workflows/ files but the token cannot push them. \
Fix: add `workflow` scope to your GitHub token, or use a GitHub App for auth. \
Original error: {}",
push_state.stored_last_error
)),
)])
.await;
} else if push_state.is_rebase_conflict_failure {
tracing::error!(
task_id = ctx.task_id,
"push failed and automatic rebase recovery failed — blocking for human intervention"
);
ctx.set(&[
("agent", serde_json::json!(null)),
("model", serde_json::json!(null)),
])
.await;
} else if push_state.push_failed {
ctx.set(&[
("agent", serde_json::json!(null)),
("model", serde_json::json!(null)),
])
.await;
let push_failures = push_state.push_failures;
if push_failures >= 3 {
tracing::error!(
task_id = ctx.task_id,
push_failures,
"push failed {push_failures} times — blocking for human intervention"
);
} else {
tracing::warn!(
task_id = ctx.task_id,
push_failures,
"agent done but push failed ({push_failures}/3) — rerouting to different agent"
);
}
} else if resp_status == "done" && !has_pr && has_delegations {
tracing::info!(
task_id = ctx.task_id,
"agent reported done with delegations but no PR — setting blocked"
);
} else if (resp_status == "done" || resp_status == "completed")
&& !has_pr
&& has_pushed
&& is_pr_base_invalid
{
tracing::error!(
task_id = ctx.task_id,
"PR creation failed with invalid base branch — blocking for human intervention"
);
ctx.set(&[(
"last_error",
serde_json::json!(
"PR creation failed terminally (invalid base branch or GitHub auth error). The agent's commits were pushed successfully. \
Check the repository default branch and the GitHub token, then create the PR manually or unblock the task."
),
)])
.await;
} else if resp_status == "done" && !has_pr && has_pushed {
tracing::warn!(
task_id = ctx.task_id,
"agent done, commits pushed, but PR creation failed — re-dispatching as routed"
);
} else if is_no_code_reroute {
if final_status == "blocked" {
let prev_no_code_agent = ctx
.get_task()
.await
.map(|t| t.no_code_last_agent.clone())
.unwrap_or_default();
let is_same_agent = !prev_no_code_agent.is_empty() && prev_no_code_agent == agent_name;
let msg = if is_same_agent {
tracing::error!(
task_id = ctx.task_id,
agent = %agent_name,
"blocking same-agent no-code loop: agent {} would be selected again",
agent_name
);
format!(
"agent {} completed without code changes twice — same-agent loop detected, blocking for human review",
agent_name
)
} else {
tracing::error!(
task_id = ctx.task_id,
no_code_reroutes,
max_reroutes,
"blocking no-code reroute: reached max reroute attempts ({}/{}) for no-code-result",
no_code_reroutes, max_reroutes
);
format!(
"agent completed without code changes after {}/{} reroute attempts on external task requiring PR",
no_code_reroutes, max_reroutes
)
};
ctx.set(&[("last_error", serde_json::json!(msg))]).await;
} else {
ctx.set(&[(
"last_error",
serde_json::json!(
"agent completed without code changes on external task requiring PR"
),
)])
.await;
}
ctx.set(&[
("no_code_last_agent", serde_json::json!(agent_name)),
("agent", serde_json::json!(null)),
("model", serde_json::json!(null)),
])
.await;
} else if resp_status == "done" && !has_pr && is_external {
tracing::info!(
task_id = ctx.task_id,
"external task with non-code labels reported done — marking done without PR"
);
} else if resp_status == "done" && !has_pr {
tracing::info!(
task_id = ctx.task_id,
"internal task reported done with no PR — marking done"
);
}
}
async fn store_token_usage(
ctx: &StoreCtx<'_>,
input_tokens: Option<u64>,
output_tokens: Option<u64>,
model_name: Option<&str>,
) {
let (Some(input), Some(output)) = (input_tokens, output_tokens) else {
return;
};
let model = model_name.unwrap_or("haiku");
if let Some(ref st) = ctx.store {
if let Some(store_id) = ctx.store_id_opt {
if let Err(e) = st.store_tokens(store_id, input, output, model).await {
tracing::warn!(task_id = ctx.task_id, ?e, "failed to store token usage");
}
} else if let Ok(Some(store_id)) = st.resolve_task_id(ctx.repo, ctx.task_id).await {
if let Err(e) = st.store_tokens(store_id, input, output, model).await {
tracing::warn!(task_id = ctx.task_id, ?e, "failed to store token usage");
}
}
}
}
#[allow(clippy::too_many_arguments)]
pub async fn handle_success(
task_id: &str,
parsed: agents::ParsedResponse,
wt: &worktree::WorktreeSetup,
task_title: &str,
agent_name: &str,
model_name: Option<&str>,
new_attempts: u32,
repo: &str,
store: &Option<Arc<TaskStore>>,
raw_stdout: &str,
) -> anyhow::Result<(String, bool)> {
let input_tokens = parsed.input_tokens;
let output_tokens = parsed.output_tokens;
let mut resp = parsed.response;
tracing::info!(
task_id,
status = resp.status,
summary = resp.summary,
"agent completed successfully"
);
let push_denied = resp.status == "blocked" && is_denied_push_block(&resp);
let store_id_opt = store::resolve_store_id(store, repo, task_id).await;
let ctx = StoreCtx {
store,
store_id_opt,
repo,
task_id,
};
ctx.set(&[("network_retries", serde_json::json!(0))]).await;
let git = if resp.status == "done"
|| resp.status == "completed"
|| resp.status == "in_progress"
|| resp.status == "needs_review"
|| push_denied
{
run_git_ops(
&ctx,
wt,
task_title,
&resp,
agent_name,
model_name,
new_attempts,
)
.await
} else {
GitOpsResult::default()
};
if push_denied && (git.has_pr || git.has_pushed) {
tracing::info!(
task_id,
"agent blocked on denied push, orch pushed the branch"
);
resp.status = "done".to_string();
}
if !resp.delegations.is_empty() {
ctx.set(&[("delegations", serde_json::json!(resp.delegations))])
.await;
}
let has_delegations = !resp.delegations.is_empty();
let push_state = detect_push_failure_state(&ctx, git.has_commits, git.has_pushed).await;
let is_external = !task_id.starts_with("internal:");
let requires_pr = is_external && !git.has_pr && !resp.status.starts_with("needs_review");
let max_reroutes: u32 = config::get("workflow.max_reroute_attempts")
.ok()
.and_then(|s| s.parse().ok())
.or_else(|| {
config::get("workflow.max_attempts")
.ok()
.and_then(|s| s.parse().ok())
})
.unwrap_or(3);
let is_no_code_reroute = !push_state.is_workflow_scope_failure
&& !push_state.push_failed
&& (resp.status == "done" || resp.status == "completed")
&& !has_delegations
&& !git.has_pushed
&& requires_pr;
let superseded_pr = match (is_no_code_reroute, ctx.get_task().await) {
(true, Some(task)) => {
superseded_conflict_pr(task.merge_conflict_retries, task.pr_number, git.has_commits)
}
_ => None,
};
let is_no_code_reroute = is_no_code_reroute && superseded_pr.is_none();
let no_code_reroutes =
detect_no_code_reroutes(&ctx, is_no_code_reroute, new_attempts, max_reroutes).await;
let blocked_reason = if resp.status == "blocked" {
Some(agent_blocked_reason(&resp))
} else {
None
};
let is_retryable_blocked = blocked_reason
.as_deref()
.map(is_retryable_blocked_reason)
.unwrap_or(false);
let is_same_agent = if is_no_code_reroute {
let prev_no_code_agent = ctx
.get_task()
.await
.map(|t| t.no_code_last_agent.clone())
.unwrap_or_default();
!prev_no_code_agent.is_empty() && prev_no_code_agent == agent_name
} else {
false
};
let final_status_owned = classify_final_status(&DecisionInput {
agent_status: &resp.status,
is_workflow_scope_failure: push_state.is_workflow_scope_failure,
is_rebase_conflict_failure: push_state.is_rebase_conflict_failure,
push_failed: push_state.push_failed,
push_failures: push_state.push_failures,
has_pr: git.has_pr,
has_delegations,
has_pushed: git.has_pushed,
is_pr_base_invalid: git.pr_base_invalid,
requires_pr,
no_code_reroutes,
max_reroutes,
is_same_agent,
is_completed_status: resp.status == "completed",
is_retryable_blocked,
is_superseded_conflict: superseded_pr.is_some(),
});
let final_status = final_status_owned.as_str();
apply_post_decision_effects(
&ctx,
final_status,
&resp.status,
&push_state,
is_no_code_reroute,
git.has_pr,
git.has_pushed,
git.pr_base_invalid,
has_delegations,
is_external,
no_code_reroutes,
max_reroutes,
agent_name,
)
.await;
if resp.status == "blocked" {
let reason = blocked_reason.unwrap_or_else(|| agent_blocked_reason(&resp));
if final_status == "new" {
crate::engine::cooldown::record_agent_failure_with_message(agent_name, &reason).await;
if let Some(m) = model_name {
crate::engine::cooldown::record_model_failure(agent_name, m).await;
}
ctx.set(&[
(
"last_error",
serde_json::json!(format!(
"transient blocker (auto-retry scheduled): {reason}"
)),
),
("agent", serde_json::json!(null)),
("model", serde_json::json!(null)),
])
.await;
tracing::warn!(
task_id = ctx.task_id,
reason = %reason,
"agent returned retryable blocked status — rerouting as new with cooldown"
);
} else if final_status == "blocked" {
ctx.set(&[("last_error", serde_json::json!(reason))]).await;
}
}
if let Some(pr_number) = superseded_pr {
tracing::info!(
task_id = ctx.task_id,
pr_number,
"conflict reroute found the fix already on the default branch — closing superseded PR"
);
close_superseded_pr(&ctx, pr_number).await;
}
ctx.set(&[("summary", serde_json::json!(resp.summary))])
.await;
store_token_usage(
&ctx,
input_tokens.or(resp.input_tokens),
output_tokens.or(resp.output_tokens),
model_name,
)
.await;
response::store_learnings_from_response(
task_id,
new_attempts,
agent_name,
model_name,
&resp,
resp.error.as_deref(),
store,
repo,
raw_stdout,
)
.await;
Ok((final_status.to_string(), push_state.push_failed))
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn local_rate_limit_skip_is_detected() {
assert!(is_local_rate_limit_skip(
"GitHub API error: GitHub REST API rate-limited, skipping call for 236s"
));
assert!(!is_local_rate_limit_skip(
"GitHub API POST failed (502): Bad Gateway"
));
}
fn git(dir: &Path, args: &[&str]) {
let out = std::process::Command::new("git")
.current_dir(dir)
.args(args)
.output()
.unwrap_or_else(|e| panic!("git {args:?} failed to start: {e}"));
assert!(
out.status.success(),
"git {args:?} failed:\nstdout: {}\nstderr: {}",
String::from_utf8_lossy(&out.stdout),
String::from_utf8_lossy(&out.stderr)
);
}
fn init_clone(remote: &Path, dir: &Path) {
git(dir, &["clone", remote.to_str().unwrap(), "."]);
git(dir, &["config", "user.email", "t@t.com"]);
git(dir, &["config", "user.name", "T"]);
git(dir, &["config", "commit.gpgsign", "false"]);
}
#[tokio::test]
async fn push_rejected_after_remote_rebase_is_pushed_not_rerouted() {
let remote = tempfile::tempdir().unwrap();
git(remote.path(), &["init", "--bare", "-b", "main"]);
let work = tempfile::tempdir().unwrap();
init_clone(remote.path(), work.path());
std::fs::write(work.path().join("README.md"), "base").unwrap();
git(work.path(), &["add", "."]);
git(work.path(), &["commit", "-m", "init"]);
git(work.path(), &["push", "origin", "HEAD:main"]);
git(work.path(), &["checkout", "-b", "feat"]);
std::fs::write(work.path().join("fix.txt"), "the fix").unwrap();
git(work.path(), &["add", "."]);
git(work.path(), &["commit", "-m", "apply the fix"]);
git(work.path(), &["push", "-u", "origin", "feat"]);
let other = tempfile::tempdir().unwrap();
init_clone(remote.path(), other.path());
git(other.path(), &["checkout", "feat"]);
git(
other.path(),
&["commit", "--amend", "-m", "apply the fix (rebased)"],
);
git(other.path(), &["push", "--force", "origin", "feat"]);
std::fs::write(work.path().join("more.txt"), "follow-up").unwrap();
git(work.path(), &["add", "."]);
git(work.path(), &["commit", "-m", "follow-up"]);
let store = Arc::new(TaskStore::open_memory().await.unwrap());
let store_id = store
.create_internal("owner/repo", "Push recovery", "", "task", "3783", None)
.await
.unwrap();
let task_id = format!("internal:{store_id}");
let store_opt = Some(Arc::clone(&store));
let ctx = StoreCtx {
store: &store_opt,
store_id_opt: Some(store_id),
repo: "owner/repo",
task_id: &task_id,
};
let result =
push_branch_with_log(&ctx, work.path(), "feat", "main", "claude", Some("opus")).await;
assert!(
matches!(result, PushResult::Success),
"orch must land the push itself after adopting the moved remote branch"
);
let task = store.get(store_id).await.unwrap();
assert!(
task.last_error.is_empty(),
"a recovered push must leave no error behind, got: {}",
task.last_error
);
assert_eq!(task.push_failures, 0, "recovery is not a push failure");
let log = std::process::Command::new("git")
.current_dir(work.path())
.args(["log", "--oneline", "origin/feat"])
.output()
.unwrap();
let log = String::from_utf8_lossy(&log.stdout);
assert!(
log.contains("follow-up"),
"remote must carry the task's commit: {log}"
);
assert!(
log.contains("apply the fix (rebased)"),
"orch must adopt the rebased remote history, not re-push its own: {log}"
);
assert!(
!log.contains("apply the fix\n"),
"the superseded commit must not come back: {log}"
);
}
#[test]
fn missing_path_error_detects_io_not_found() {
let err = anyhow::Error::from(std::io::Error::new(
std::io::ErrorKind::NotFound,
"simulated missing path",
));
assert!(is_missing_path_error(&err));
}
#[test]
fn missing_path_error_detects_message_fallback() {
let err = anyhow::anyhow!("push failed: No such file or directory (os error 2)");
assert!(is_missing_path_error(&err));
}
#[test]
fn superseded_conflict_pr_requires_conflict_reroute_and_no_commits() {
assert_eq!(superseded_conflict_pr(1, Some(42), false), Some(42));
assert_eq!(superseded_conflict_pr(1, Some(42), true), None);
assert_eq!(superseded_conflict_pr(0, Some(42), false), None);
assert_eq!(superseded_conflict_pr(1, None, false), None);
}
#[tokio::test]
async fn conflict_reroute_with_no_commits_closes_pr_and_marks_done() {
fn git(dir: &std::path::Path, args: &[&str]) {
let ok = std::process::Command::new("git")
.args(args)
.current_dir(dir)
.env("GIT_AUTHOR_NAME", "t")
.env("GIT_AUTHOR_EMAIL", "t@example.com")
.env("GIT_COMMITTER_NAME", "t")
.env("GIT_COMMITTER_EMAIL", "t@example.com")
.output()
.expect("git runs")
.status
.success();
assert!(ok, "git {args:?} failed");
}
let dir = tempfile::tempdir().unwrap();
let origin = dir.path().join("origin.git");
std::fs::create_dir(&origin).unwrap();
git(&origin, &["init", "--bare", "-b", "main"]);
let work_dir = dir.path().join("work");
std::fs::create_dir(&work_dir).unwrap();
git(&work_dir, &["init", "-b", "main"]);
git(
&work_dir,
&["remote", "add", "origin", &origin.display().to_string()],
);
std::fs::write(work_dir.join("a.txt"), "a").unwrap();
git(&work_dir, &["add", "-A"]);
git(&work_dir, &["commit", "-m", "init"]);
git(&work_dir, &["push", "-u", "origin", "main"]);
git(&work_dir, &["checkout", "-b", "fix-branch"]);
let wt = worktree::WorktreeSetup {
work_dir: work_dir.clone(),
branch: "fix-branch".to_string(),
default_branch: "main".to_string(),
main_project_dir: work_dir.clone(),
};
let store = Arc::new(TaskStore::open_memory().await.unwrap());
let store_id = store
.create(&crate::store::NewTask {
external_id: Some("3772".to_string()),
repo: "owner/repo".to_string(),
origin: "github".to_string(),
title: "Fix the thing".to_string(),
body: String::new(),
source: "manual".to_string(),
source_id: String::new(),
author: "bot".to_string(),
url: String::new(),
labels: vec![],
parent_id: None,
})
.await
.unwrap();
store
.set_fields(store_id, &[("pr_number", serde_json::json!(42))])
.await
.unwrap();
store
.increment(store_id, "merge_conflict_retries")
.await
.unwrap();
store
.set_fields(
store_id,
&[("no_code_last_agent", serde_json::json!("claude"))],
)
.await
.unwrap();
let parsed = agents::ParsedResponse {
response: AgentResponse {
status: "done".to_string(),
summary: "the fix is already on main".to_string(),
..Default::default()
},
input_tokens: None,
output_tokens: None,
duration_ms: None,
};
let (final_status, push_failed) = handle_success(
"3772",
parsed,
&wt,
"Fix the thing",
"claude",
Some("sonnet"),
1,
"owner/repo",
&Some(Arc::clone(&store)),
"",
)
.await
.unwrap();
assert_eq!(
final_status, "done",
"a superseded conflict reroute must end done, not blocked"
);
assert!(!push_failed);
let task = store.get(store_id).await.unwrap();
assert!(
task.last_error.contains("superseded"),
"the close attempt must be recorded on the task: {}",
task.last_error
);
}
#[test]
fn classify_superseded_conflict_marks_done() {
let status = classify_final_status(&DecisionInput {
agent_status: "done",
requires_pr: true,
is_same_agent: true,
is_superseded_conflict: true,
..Default::default()
});
assert_eq!(status, "done");
}
#[test]
fn classify_workflow_scope_failure_blocks() {
let status = classify_final_status(&DecisionInput {
is_workflow_scope_failure: true,
push_failed: true,
push_failures: 1,
..Default::default()
});
assert_eq!(status, "blocked");
}
#[test]
fn classify_rebase_conflict_failure_blocks() {
let status = classify_final_status(&DecisionInput {
is_rebase_conflict_failure: true,
push_failed: true,
..Default::default()
});
assert_eq!(status, "blocked");
}
#[test]
fn classify_push_failed_under_threshold_reroutes() {
let status = classify_final_status(&DecisionInput {
push_failed: true,
push_failures: 2, ..Default::default()
});
assert_eq!(status, "new");
}
#[test]
fn denied_push_block_detected() {
let resp = |s: &str| crate::parser::AgentResponse {
status: "blocked".to_string(),
summary: s.to_string(),
..Default::default()
};
assert!(is_denied_push_block(&resp(
"Work committed, but `git push -u origin HEAD` was denied"
)));
assert!(!is_denied_push_block(&resp("Missing API key")));
}
#[test]
fn classify_push_failed_at_threshold_blocks() {
let status = classify_final_status(&DecisionInput {
push_failed: true,
push_failures: 3, ..Default::default()
});
assert_eq!(status, "blocked");
}
#[test]
fn classify_done_with_pr_goes_to_needs_review() {
let status = classify_final_status(&DecisionInput {
agent_status: "done",
has_pr: true,
..Default::default()
});
assert_eq!(status, "needs_review");
}
#[test]
fn classify_done_with_delegations_no_pr_blocks() {
let status = classify_final_status(&DecisionInput {
agent_status: "done",
has_delegations: true,
..Default::default()
});
assert_eq!(status, "blocked");
}
#[test]
fn classify_done_pushed_no_pr_reroutes_for_pr_creation() {
let status = classify_final_status(&DecisionInput {
agent_status: "done",
has_pushed: true,
..Default::default()
});
assert_eq!(status, "routed");
}
#[test]
fn classify_done_pushed_pr_base_invalid_blocks() {
let status = classify_final_status(&DecisionInput {
agent_status: "done",
has_pushed: true,
is_pr_base_invalid: true,
..Default::default()
});
assert_eq!(status, "blocked");
}
#[test]
fn classify_completed_pushed_pr_base_invalid_blocks() {
let status = classify_final_status(&DecisionInput {
agent_status: "completed",
is_completed_status: true,
has_pushed: true,
is_pr_base_invalid: true,
..Default::default()
});
assert_eq!(status, "blocked");
}
#[test]
fn classify_done_requires_pr_under_max_reroutes() {
let status = classify_final_status(&DecisionInput {
agent_status: "done",
requires_pr: true,
no_code_reroutes: 1,
max_reroutes: 3,
is_same_agent: false,
..Default::default()
});
assert_eq!(status, "new");
}
#[test]
fn classify_done_requires_pr_same_agent_blocks_immediately() {
let status = classify_final_status(&DecisionInput {
agent_status: "done",
requires_pr: true,
no_code_reroutes: 1,
max_reroutes: 3,
is_same_agent: true, ..Default::default()
});
assert_eq!(status, "blocked");
}
#[test]
fn classify_done_requires_pr_at_max_reroutes_blocks() {
let status = classify_final_status(&DecisionInput {
agent_status: "done",
requires_pr: true,
no_code_reroutes: 3,
max_reroutes: 3,
is_same_agent: false,
..Default::default()
});
assert_eq!(status, "blocked");
}
#[test]
fn classify_done_requires_pr_exceeds_max_reroutes_blocks() {
let status = classify_final_status(&DecisionInput {
agent_status: "done",
requires_pr: true,
no_code_reroutes: 5,
max_reroutes: 3,
is_same_agent: false,
..Default::default()
});
assert_eq!(status, "blocked");
}
#[test]
fn classify_done_external_no_pushed_same_agent_blocks_immediately() {
let status = classify_final_status(&DecisionInput {
agent_status: "done",
requires_pr: true, no_code_reroutes: 0,
max_reroutes: 3,
is_same_agent: true,
..Default::default()
});
assert_eq!(status, "blocked");
}
#[test]
fn classify_done_external_no_pushed_commits_reroutes() {
let status = classify_final_status(&DecisionInput {
agent_status: "done",
has_pushed: false,
requires_pr: true, no_code_reroutes: 0,
max_reroutes: 3,
is_same_agent: false,
..Default::default()
});
assert_eq!(status, "new");
}
#[test]
fn classify_done_internal_task_no_pr_is_done() {
let status = classify_final_status(&DecisionInput {
agent_status: "done",
has_pushed: false,
requires_pr: false,
..Default::default()
});
assert_eq!(status, "done");
}
#[test]
fn classify_non_done_status_passes_through() {
for status_str in &["in_progress", "needs_review", "blocked"] {
let result = classify_final_status(&DecisionInput {
agent_status: status_str,
..Default::default()
});
assert_eq!(
result, *status_str,
"status '{status_str}' should pass through unchanged"
);
}
}
#[test]
fn classify_descriptive_completion_status_normalizes_to_done() {
let status = classify_final_status(&DecisionInput {
agent_status: "Trading scan complete. Nothing to trade.",
..Default::default()
});
assert_eq!(status, "done");
}
#[test]
fn classify_descriptive_completion_with_failure_cues_does_not_normalize() {
let status = classify_final_status(&DecisionInput {
agent_status: "completed but failed to push branch due to error",
..Default::default()
});
assert_eq!(status, "completed but failed to push branch due to error");
}
#[test]
fn classify_push_failed_takes_precedence_over_has_pr() {
let status = classify_final_status(&DecisionInput {
agent_status: "done",
has_pr: true,
push_failed: true,
push_failures: 1,
..Default::default()
});
assert_eq!(status, "new");
}
#[test]
fn classify_workflow_scope_failure_takes_precedence_over_push_reroute() {
let status = classify_final_status(&DecisionInput {
is_workflow_scope_failure: true,
push_failed: true,
push_failures: 1, ..Default::default()
});
assert_eq!(status, "blocked");
}
#[test]
fn classify_completed_with_pr_goes_to_needs_review() {
let status = classify_final_status(&DecisionInput {
agent_status: "completed",
has_pr: true,
is_completed_status: true,
..Default::default()
});
assert_eq!(status, "needs_review");
}
#[test]
fn classify_completed_with_delegations_no_pr_blocks() {
let status = classify_final_status(&DecisionInput {
agent_status: "completed",
has_delegations: true,
is_completed_status: true,
..Default::default()
});
assert_eq!(status, "blocked");
}
#[test]
fn classify_completed_pushed_no_pr_reroutes_for_pr_creation() {
let status = classify_final_status(&DecisionInput {
agent_status: "completed",
has_pushed: true,
is_completed_status: true,
..Default::default()
});
assert_eq!(status, "routed");
}
#[test]
fn classify_completed_requires_pr_under_max_reroutes() {
let status = classify_final_status(&DecisionInput {
agent_status: "completed",
requires_pr: true,
no_code_reroutes: 1,
max_reroutes: 3,
is_completed_status: true,
is_same_agent: false,
..Default::default()
});
assert_eq!(status, "new");
}
#[test]
fn classify_completed_requires_pr_at_max_reroutes_blocks() {
let status = classify_final_status(&DecisionInput {
agent_status: "completed",
requires_pr: true,
no_code_reroutes: 3,
max_reroutes: 3,
is_completed_status: true,
is_same_agent: false,
..Default::default()
});
assert_eq!(status, "blocked");
}
#[test]
fn classify_completed_same_agent_blocks_immediately() {
let status = classify_final_status(&DecisionInput {
agent_status: "completed",
requires_pr: true,
no_code_reroutes: 1,
max_reroutes: 3,
is_completed_status: true,
is_same_agent: true,
..Default::default()
});
assert_eq!(status, "blocked");
}
#[test]
fn classify_completed_internal_task_is_done() {
let status = classify_final_status(&DecisionInput {
agent_status: "completed",
requires_pr: false,
is_completed_status: true,
..Default::default()
});
assert_eq!(status, "done");
}
#[test]
fn agent_blocked_reason_prefers_error_field() {
let resp = crate::parser::AgentResponse {
status: "blocked".to_string(),
error: Some("CDP endpoint not reachable".to_string()),
summary: "Could not complete research".to_string(),
remaining: vec!["Run x-twitter-brave with reachable endpoint".to_string()],
..Default::default()
};
let reason = agent_blocked_reason(&resp);
assert!(
reason.contains("CDP endpoint not reachable"),
"reason: {reason}"
);
assert!(reason.contains("Remaining:"), "reason: {reason}");
assert!(reason.contains("x-twitter-brave"), "reason: {reason}");
}
#[test]
fn agent_blocked_reason_falls_back_to_summary() {
let resp = crate::parser::AgentResponse {
status: "blocked".to_string(),
error: None,
summary: "Could not access required data sources in sandboxed environment".to_string(),
remaining: vec![],
..Default::default()
};
let reason = agent_blocked_reason(&resp);
assert_eq!(
reason,
"Could not access required data sources in sandboxed environment"
);
}
#[test]
fn agent_blocked_reason_generic_fallback() {
let resp = crate::parser::AgentResponse {
status: "blocked".to_string(),
..Default::default()
};
let reason = agent_blocked_reason(&resp);
assert_eq!(reason, "agent returned blocked status without a reason");
}
#[test]
fn classify_agent_blocked_passes_through() {
let status = classify_final_status(&DecisionInput {
agent_status: "blocked",
..Default::default()
});
assert_eq!(status, "blocked");
}
#[test]
fn classify_retryable_agent_blocked_reroutes() {
let status = classify_final_status(&DecisionInput {
agent_status: "blocked",
is_retryable_blocked: true,
..Default::default()
});
assert_eq!(status, "new");
}
#[test]
fn retryable_blocked_reason_detects_transient_patterns() {
assert!(is_retryable_blocked_reason(
"Twitter access unavailable: service timed out"
));
assert!(is_retryable_blocked_reason(
"final commit blocked by git worktree lock permission error"
));
assert!(is_retryable_blocked_reason(
"could not commit due a worktree git lock permission error"
));
assert!(is_retryable_blocked_reason(
"git lockfile permission restrictions in this environment"
));
assert!(is_retryable_blocked_reason(
"API rate limit hit, try again later"
));
assert!(is_retryable_blocked_reason(
"Credential bean/hyperliquid-address exists in the pass store but GPG decryption fails in this agent session — no passphrase available."
));
assert!(is_retryable_blocked_reason(
"credential resolution failed via PasswordStoreProvider; AgentVaultProvider unavailable"
));
}
#[test]
fn retryable_blocked_reason_rejects_permanent_patterns() {
assert!(!is_retryable_blocked_reason(
"invalid request payload: malformed JSON"
));
assert!(!is_retryable_blocked_reason(
"resource not found in repository"
));
}
}