magi/

conduct.rs

//! The conductor: a single agent seat that arranges the queue.
//!
//! [`crate::queue`] orders runnable tasks by `priority` alone; nothing in it
//! can express that one task should wait for another, or that a task whose
//! last run stopped short deserves a second look before the loop blindly
//! retries it. Once per polling cycle, [`Conductor::maybe_run`] shows one
//! agent seat three things - the runnable tasks, the tasks stuck `running`
//! with no live daemon behind them, and the `failed`/`held` tasks nobody has
//! decided about yet - and asks it to decide `blocked_by` for the first and a
//! recovery for the other two. Everything else about the loop - which
//! unblocked task runs next, one at a time, in `priority` order - is
//! unchanged; see `crate::daemon`.
//!
//! # What the conductor may not do
//!
//! [`Decision`] has no field for `priority`, for deleting a task, or for
//! touching git, a worktree, or a branch directly. [`Recovery::Review`] only
//! ever reopens a branch `crate::daemon` itself resolved from the task's own
//! run record ([`surviving_branch`]) - never a name the model wrote - through
//! `crate::graph::Runner::review`, which reviews and verifies but never
//! rewrites history.
//!
//! # Non-blocking by construction
//!
//! [`crate::ask::ask_and_wait`] is never called from here, and the prompt
//! tells the model the same: that CLI command blocks until a human answers,
//! and calling it from inside the conductor's own invocation would park the
//! whole polling loop behind one task's question. Instead a decision that
//! wants the operator's judgement carries a `question` field, and [`apply`]
//! files it with [`Question::new`] and [`Questions::put`] and moves on in the
//! same call.
//!
//! # Fails soft, always
//!
//! [`Conductor::maybe_run`] never returns an error: an unusable roster, a
//! timed-out invocation, or a reply [`verdict::extract_json`] cannot parse are
//! all logged and treated as "this cycle changes nothing." `crate::daemon`'s
//! loop always falls through to its own `Queue::next_runnable` regardless of
//! what happened here.

use std::collections::BTreeSet;
use std::path::{Path, PathBuf};
use std::time::Duration;

use anyhow::{Context as _, Result, bail};
use serde::Deserialize;

use crate::agent::{self, Invocation, SeatState};
use crate::ask::{Question, Questions};
use crate::config::Config;
use crate::prompt;
use crate::queue::{Queue, Task, TaskStatus};
use crate::run::RunState;
use crate::verdict;

/// Seat name for the conductor's own CLI-side conversation, scoped away from
/// every other seat magi ever opens - the same rule every other seat follows.
const SEAT: &str = "conduct";

/// Node name reported to the invoked agent (`MAGI_NODE`) and recorded on any
/// question it files, so an operator reading the questions list can tell a
/// conductor's question from one a run's own agent asked.
pub const NODE: &str = "conduct";

/// Wall-clock limit for one conductor turn. The conductor reads a queue
/// listing and replies with json; it does not implement anything or run a
/// build, so this is short - the same order of magnitude as
/// `crate::chat`'s own single-turn, no-write invocations.
const TURN_TIMEOUT: Duration = Duration::from_secs(300);

/// How many of a task's [`Task::answers`] may already come from
/// `crate::conduct` before a further `question` decision is refused in
/// favor of [`Task::hold_machine`].
///
/// [`Task::answers`] only grows through [`crate::daemon::resolve_blockers`]
/// recording an answer this module's own question produced (see
/// [`Task::record_answer`]'s call site), so this counts settled
/// conductor-and-operator exchanges specifically, not every question a task
/// has ever seen. A model that does not register its own question as
/// already answered - misreading [`prompt::ConductTask::answers`], or simply
/// asking the same thing worded differently - would otherwise keep filing a
/// fresh [`Question`] every cycle its revision changes, growing `magi answer
/// --list` without bound and never letting the task actually rest; see
/// `apply_one`'s use of this constant. Two lets one genuine follow-up
/// through - a task that has needed more than that many rounds of the
/// operator's own words is better served by a human looking at it directly
/// than by another automated question.
const MAX_SETTLED_CONDUCT_ANSWERS: usize = 2;

/// What the conductor may choose for a `running`-but-stalled or a
/// `failed`/`held` task.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum Recovery {
    /// Put it back in line, attempts reset - the same effect as
    /// `magi task release`.
    Requeue,
    /// Leave it for a human, unchanged otherwise - the same effect as
    /// `magi task hold`.
    Hold,
    /// Reopen the task's surviving branch as a review-only pass
    /// (`crate::graph::Runner::review`) instead of competing from scratch.
    /// Only takes effect when [`surviving_branch`] can actually name one;
    /// otherwise `crate::daemon` falls back to [`Recovery::Requeue`].
    Review,
    /// Close the task outright, as [`Task::succeed`] - the same effect as
    /// `magi task done`. For a `failed`/`held` task only, never `running`:
    /// this is for a task whose own goal is already known to be met outside
    /// the loop entirely (the branch was merged and the worktree cleaned up
    /// by hand, say) and competing it again would only spend attempts on
    /// work that has nothing left to do, not for a task that merely stopped
    /// mid-competition and might still need to run.
    Done,
}

/// The conductor's decision for one task. Deliberately has no `priority`
/// field: see this module's doc.
#[derive(Debug, Clone, Default, Deserialize)]
pub struct Decision {
    /// Task id, expected to be copied verbatim from what it was shown.
    pub id: String,
    /// Ids this task should be blocked on. Meaningful only for a runnable
    /// (`queued`) task; ignored otherwise.
    #[serde(default)]
    pub blocked_by: Vec<String>,
    /// One line explaining the block or the recovery.
    #[serde(default)]
    pub reason: Option<String>,
    /// Recovery for a stalled or finished task. For a runnable (`queued`)
    /// one, only [`Recovery::Hold`] has any effect - and only when
    /// `blocked_by` is empty, since a `queued` task with something to wait on
    /// is handled by that field instead - holding a task the operator has
    /// already said should not compete again without filing another
    /// `question` that only restates the same answer. `Requeue`, `Review`,
    /// and `Done` have no meaning for a task that is already in line.
    #[serde(default)]
    pub recovery: Option<Recovery>,
    /// A question for the operator. When present, [`apply`] files it (unless
    /// one is already open for this task) and blocks the task on its id
    /// instead of acting on `blocked_by` or `recovery`.
    #[serde(default)]
    pub question: Option<String>,
    /// Fixed answers for `question`, if it has any. Empty means free text.
    #[serde(default)]
    pub choices: Vec<String>,
}

/// The conductor's whole reply for one cycle.
///
/// `decisions` is deliberately **not** `#[serde(default)]`, unlike every
/// other field in this module. [`verdict::extract_json`] disambiguates
/// between several balanced `{...}` spans in one reply by trying the type the
/// caller wants against each of them, last first, and keeping the first that
/// fits - which only works when a span that is not really the answer can
/// fail to fit. A `Verdict` with no required field at all would make every
/// span fit, including a `{}` left by stray trailing prose, and the reply's
/// real `decisions` - earlier in the text - would never be reached. Requiring
/// the key costs nothing: the prompt already asks for it on every reply, `[]`
/// included.
#[derive(Debug, Clone, Default, Deserialize)]
pub struct Verdict {
    /// One entry per task the conductor chose to say something about. A task
    /// left out of this list is left exactly as it was.
    pub decisions: Vec<Decision>,
}

/// A view of a task built for [`prompt::conduct`], shared by the runnable and
/// stalled sections.
fn view(t: &Task, max_attempts: usize) -> prompt::ConductTask {
    prompt::ConductTask {
        id: t.id.clone(),
        title: t.title.clone(),
        instruction: t.instruction.clone(),
        repo: t.repo.display().to_string(),
        priority: t.priority,
        status: t.status.as_str().to_owned(),
        attempts: t.attempts,
        max_attempts,
        last_error: t.last_error.clone(),
        hold_reason: t.hold_reason.clone(),
        hold_source: t.hold_source.map(|source| source.label().to_owned()),
        blocked_by: t.blocked_by.clone(),
        answers: t
            .answers
            .iter()
            .map(|a| prompt::ConductAnswer {
                question: a.question.clone(),
                answer: a.answer.clone(),
            })
            .collect(),
        operator_resume: t.resume_override.as_ref().map(|o| {
            format!(
                "the operator explicitly answered \"resume\" at {}; do not hold this \
                 task again for the same reason unless there is new information",
                o.at
            )
        }),
    }
}

/// May the conductor's `Recovery::Hold` take effect on `task`, given an
/// operator's recorded "resume" answer? The conductor is allowed to override
/// that answer once - and the override is recorded so `crate::triage` can
/// show the operator the contradiction - but never after the operator
/// insisted (`forced`), and never a second time.
///
/// Returns `false` when the hold must not be applied (nothing to write).
/// When it returns `true` the override, if any, has been recorded on `task`.
fn may_hold(task: &mut Task, reason: &str) -> bool {
    let Some(o) = task.resume_override.as_mut() else {
        return true;
    };
    if o.forced {
        tracing::warn!(
            "conductor tried to hold task {} after the operator forced a resume: {reason}",
            task.id
        );
        return false;
    }
    if o.conductor_rehold.is_some() {
        return false;
    }
    o.conductor_rehold = Some(reason.to_owned());
    true
}

/// Severity as a lowercase word, matching how `crate::verdict::Severity` is
/// spelled everywhere else an operator or a model reads it.
fn severity_str(s: crate::verdict::Severity) -> &'static str {
    match s {
        crate::verdict::Severity::Nit => "nit",
        crate::verdict::Severity::Minor => "minor",
        crate::verdict::Severity::Major => "major",
        crate::verdict::Severity::Blocker => "blocker",
    }
}

/// The branch a task's last run left behind, if the tally ever ran on it -
/// what [`Recovery::Review`] reopens. Derived from `crate::run::RunState`
/// alone, never from anything the conductor wrote, so a hallucinated branch
/// name can never reach `crate::graph::Runner::review`.
fn surviving_branch(task: &Task) -> Option<String> {
    let last = task.runs.last()?;
    let state = RunState::load(last).ok()?;
    state.winner().map(|c| c.branch.clone())
}

/// The reason to record when the conductor holds a `failed`/`held` task via
/// [`Recovery::Hold`]. Always `Some`, never a bare `d.reason.clone()`.
///
/// [`Task::hold_machine`] only overwrites [`Task::hold_reason`] when given
/// one, precisely so a stalled task's existing diagnosis survives a
/// decision that has nothing new to add. That is the right default when the
/// task was not already `held` - but a task that *is* already `held`, and
/// is being held again here, is a different case: its current
/// `hold_reason` may still read as a cause `crate::triage` knows how to
/// re-check on its own (a disk-pressure message, say - see
/// `triage::machine_cause_resolved`), and if the model gives no reason,
/// leaving that text untouched would make this decision - the conductor
/// choosing, informed by the operator's own answer, to keep the task held
/// anyway - indistinguishable from the original, never-reconsidered hold.
/// `crate::triage` would then read the stale text the next time its one
/// recognised cause looks resolved and release the task straight through
/// the decision this call was recording. Prepending (not appending) the new
/// note keeps the old text as context without leaving the string starting
/// with whatever pattern `crate::triage` matched before.
fn reaffirmed_hold_reason(task: &Task, d: &Decision) -> String {
    let note = match &d.reason {
        Some(reason) => reason.clone(),
        None => match task.answers.last() {
            Some(a) => format!(
                "conduct held this again with no new reason given; last operator \
                 answer on record: {}",
                a.answer
            ),
            None => "conduct held this again with no reason given".to_owned(),
        },
    };
    match task.hold_reason.as_deref() {
        Some(prior) if !prior.is_empty() => format!("{note}\n\n(previously: {prior})"),
        _ => note,
    }
}

/// The conductor's stated reason for a hold, or a stand-in when it gave none.
fn hold_note(d: &Decision) -> String {
    d.reason
        .clone()
        .unwrap_or_else(|| "(no reason given)".to_owned())
}

/// Everything the conductor is shown about a `failed`/`held` task's last run.
async fn outcome_for(task: &Task, repo: &Path) -> prompt::ConductOutcome {
    let Some(run_id) = task.runs.last().cloned() else {
        return prompt::ConductOutcome {
            run_id: "(none)".to_owned(),
            unreadable: Some("this task has not produced a run yet".to_owned()),
            run_status: None,
            open_findings: Vec::new(),
            rounds_used: 0,
            rounds_max: 0,
            rounds: Vec::new(),
            branch: None,
            branch_head: None,
        };
    };
    let state = match RunState::load(&run_id) {
        Ok(s) => s,
        Err(e) => {
            // The exact failure this feature exists to stop hiding: a schema
            // mismatch (or any other unreadable state) must never be treated
            // as "nothing to recover" - it is surfaced here, verbatim, rather
            // than swallowed into a quiet re-competition.
            tracing::warn!(
                "conductor: could not read run {run_id} for task {}: {e:#}",
                task.short()
            );
            return prompt::ConductOutcome {
                run_id,
                unreadable: Some(format!("{e:#}")),
                run_status: None,
                open_findings: Vec::new(),
                rounds_used: 0,
                rounds_max: 0,
                rounds: Vec::new(),
                branch: None,
                branch_head: None,
            };
        }
    };

    let finding_view = |f: &crate::verdict::Finding| prompt::ConductFinding {
        id: f.id.clone(),
        title: f.title.clone(),
        severity: severity_str(f.severity).to_owned(),
    };
    let open_findings = state
        .open_findings()
        .into_iter()
        .map(finding_view)
        .collect();
    let rounds = state
        .reviews
        .iter()
        .map(|r| prompt::ConductRound {
            round: r.round,
            findings: r
                .reviews
                .iter()
                .flat_map(|rec| rec.findings.iter())
                .map(finding_view)
                .collect(),
            addressed: r
                .fix
                .as_ref()
                .map(|fx| fx.addressed.clone())
                .unwrap_or_default(),
            rejected: r
                .fix
                .as_ref()
                .map(|fx| {
                    fx.rejected
                        .iter()
                        .map(|rej| prompt::ConductRejection {
                            id: rej.id.clone(),
                            why: rej.why.clone(),
                        })
                        .collect()
                })
                .unwrap_or_default(),
        })
        .collect();
    let branch = state.winner().map(|c| c.branch.clone());
    let branch_head = match &branch {
        Some(b) => crate::git::rev_parse(repo, b)
            .await
            .ok()
            .map(|h| h.chars().take(8).collect()),
        None => None,
    };

    prompt::ConductOutcome {
        run_id,
        unreadable: None,
        run_status: Some(state.status.as_str().to_owned()),
        open_findings,
        rounds_used: state.reviews.len(),
        rounds_max: state.config.graph.review_rounds,
        rounds,
        branch,
        branch_head,
    }
}

/// A `failed`/`held` task together with how its last run ended.
async fn finished_view(t: &Task, repo: &Path, max_attempts: usize) -> prompt::ConductFinished {
    prompt::ConductFinished {
        task: view(t, max_attempts),
        outcome: outcome_for(t, &repo_for(t, repo)).await,
    }
}

/// The repository containing a task's branch. A task filed without a
/// repository uses the daemon's repository, exactly as its later attempt does.
fn repo_for(task: &Task, fallback: &Path) -> PathBuf {
    if task.repo.as_os_str().is_empty() || task.repo == Path::new(".") {
        fallback.to_path_buf()
    } else {
        task.repo.clone()
    }
}

/// Apply one decision to the queue and the question store.
///
/// Takes the task's own claim before touching it: a model call spans a whole
/// agent turn, and the queue can have moved on by the time its answer comes
/// back. A claim that cannot be taken means something else owns this task
/// right now - most often a live daemon mid-competition on it - so the
/// conductor's now-stale view of it is dropped rather than raced against; see
/// `crate::queue::Queue::claim`'s own doc on why a claim is proof, not a
/// guess.
///
/// A decision is matched against the task's *current* status, re-read under
/// the claim, not against whichever section of the prompt it came from: a
/// `blocked_by` only takes effect on a `queued` task, and `recovery` only on
/// one `running` (stalled) or `failed`/`held`, so a decision that no longer
/// matches what the task actually is - it moved on between the read that
/// built the prompt and this write - changes nothing.
fn apply_one(queue: &Queue, questions: &Questions, d: &Decision) -> Result<()> {
    let _claim = queue
        .claim(&d.id)
        .with_context(|| format!("task {} is claimed elsewhere right now", d.id))?;
    let mut task = queue.get(&d.id).context("no such task")?;

    // A conductor answer is never operator authorization.  In particular,
    // do this before questions and blocking too: either would reclassify a
    // manual hold and let a later deterministic resolver queue it.
    if task.operator_held() {
        return Ok(());
    }

    // `crate::triage` is already running its own question-and-answer cycle
    // on this hold - see that module's doc on why it never calls
    // `Task::block`, and `crate::triage::pending_for`'s own doc on why "open"
    // alone is not enough here: an answered-but-not-yet-applied triage
    // question is still triage's to finish, since `crate::triage::run_once`
    // only runs on a fully idle tick and only ever looks at tasks still
    // `held`. Blocking this task out from under it - even on an unrelated
    // question - would move it to `Blocked`, and the pending triage answer
    // would never be read back.
    if task.status == TaskStatus::Held && crate::triage::pending_for(questions, &task) {
        return Ok(());
    }

    if let Some(text) = &d.question {
        if task.status == TaskStatus::Done {
            return Ok(());
        }
        // `Question::run` is a task id for conductor questions, but ordinary
        // graph questions use it as a run id. A coincidental equality must
        // not block this task on an answer meant for another node.
        let question_id = match questions
            .list()
            .into_iter()
            .find(|q| q.status.open() && q.node == NODE && q.run == task.id)
        {
            Some(existing) => existing.id,
            // See `MAX_SETTLED_CONDUCT_ANSWERS`'s own doc: this many
            // conductor questions have already been answered about this
            // task with nothing left open, so a further one is refused in
            // favor of a hold rather than growing the question list forever.
            None if task.answers.len() >= MAX_SETTLED_CONDUCT_ANSWERS => {
                task.hold_machine(Some(format!(
                    "conduct tried to ask another question after {} were \
                     already answered about this task: {text}",
                    task.answers.len()
                )));
                return queue.put(&mut task);
            }
            None => {
                let mut q = Question::new(
                    task.id.clone(),
                    NODE.to_owned(),
                    SEAT.to_owned(),
                    text.clone(),
                    d.reason.clone().unwrap_or_default(),
                    d.choices.clone(),
                );
                questions.put(&mut q)?;
                q.id
            }
        };
        task.block(vec![question_id], d.reason.clone());
        return queue.put(&mut task);
    }

    match task.status {
        TaskStatus::Queued if !d.blocked_by.is_empty() => {
            task.block(d.blocked_by.clone(), d.reason.clone());
            queue.put(&mut task)?;
        }
        // See `Decision::recovery`'s doc: the only lever a runnable task has
        // besides `blocked_by` is holding it outright, for a task whose
        // answers already say it should not compete again.
        TaskStatus::Queued if d.recovery == Some(Recovery::Hold) => {
            if may_hold(&mut task, &hold_note(d)) {
                task.hold_machine(d.reason.clone());
                queue.put(&mut task)?;
            }
        }
        TaskStatus::Running => match d.recovery {
            Some(Recovery::Requeue) => {
                task.requeue();
                queue.put(&mut task)?;
            }
            Some(Recovery::Hold) if may_hold(&mut task, &hold_note(d)) => {
                task.hold_machine(d.reason.clone());
                queue.put(&mut task)?;
            }
            // `Review` reopens a branch, which only makes sense once a run
            // has actually stopped; a task still `running` has nothing to
            // reopen yet.
            _ => {}
        },
        TaskStatus::Failed | TaskStatus::Held => match d.recovery {
            Some(Recovery::Requeue) => {
                task.requeue();
                queue.put(&mut task)?;
            }
            Some(Recovery::Hold) => {
                if may_hold(&mut task, &hold_note(d)) {
                    task.hold_machine(Some(reaffirmed_hold_reason(&task, d)));
                    queue.put(&mut task)?;
                }
            }
            Some(Recovery::Review) => {
                if let Some(branch) = surviving_branch(&task) {
                    task.request_review(branch);
                    queue.put(&mut task)?;
                }
                // No survivable branch: a decision naming `review` here is
                // simply not actionable, and is dropped rather than guessed
                // at - `crate::daemon` applies the same "no branch, no
                // review" rule again, from its own read, right before it
                // would actually start the run.
            }
            Some(Recovery::Done) => {
                task.succeed();
                queue.put(&mut task)?;
            }
            None => {}
        },
        // `queued` with nothing to block on, `done`, or already `blocked`:
        // nothing for this decision to do.
        _ => {}
    }
    Ok(())
}

/// Apply every decision in `verdict`. A single bad decision - a task id that
/// no longer exists, one already claimed elsewhere - is logged and skipped
/// rather than losing every other decision in the same reply.
pub fn apply(queue: &Queue, questions: &Questions, verdict: &Verdict) -> Result<()> {
    for d in &verdict.decisions {
        if let Err(e) = apply_one(queue, questions, d) {
            tracing::warn!("conductor decision for task {}: {e:#}", d.id);
        }
    }
    Ok(())
}

/// The conductor's state across polling cycles: its own CLI-side conversation
/// and the last (revision, stalled ∪ finished ids) pair it actually acted on.
#[derive(Debug, Default)]
pub struct Conductor {
    seat: Option<SeatState>,
    last_seen: Option<(u64, BTreeSet<String>)>,
}

impl Conductor {
    /// A conductor that has never run.
    #[must_use]
    pub fn new() -> Self {
        Self::default()
    }

    fn snapshot(queue: &Queue, stalled: &[Task], finished: &[Task]) -> (u64, BTreeSet<String>) {
        let ids = stalled
            .iter()
            .chain(finished)
            .map(|t| t.id.clone())
            .collect();
        (queue.revision(), ids)
    }

    /// Whether calling the conductor could possibly do anything different
    /// from last time: [`Queue::revision`] moved, or the set of stalled and
    /// finished task ids changed.
    ///
    /// Deliberately **not** "stalled or finished is non-empty" - a task
    /// sitting stalled or finished with nobody changing anything about it
    /// must not be re-shown to the model every single poll forever; only a
    /// change in *which* tasks are stalled or finished, or a queue write
    /// changing something about a runnable one, is worth another look.
    ///
    /// Cheap and config-free on purpose, so `crate::daemon`'s poll loop can
    /// skip `Config::discover`'s synchronous I/O entirely on a cycle where
    /// this says no - which [`Conductor::maybe_run`] would otherwise only
    /// discover after paying for that load. Both ask the identical question,
    /// from the same [`Conductor::last_seen`], so they can never disagree
    /// about whether there is anything to look at.
    #[must_use]
    pub fn worth_a_look(&self, queue: &Queue, stalled: &[Task], finished: &[Task]) -> bool {
        self.last_seen.as_ref() != Some(&Self::snapshot(queue, stalled, finished))
    }

    /// Call the conductor once, unless nothing has changed since the last
    /// time it was worth calling - see [`Conductor::worth_a_look`], the exact
    /// same test. Never fatal - see this module's doc.
    #[allow(clippy::too_many_arguments)]
    pub async fn maybe_run(
        &mut self,
        cfg: &Config,
        repo: &Path,
        queue: &Queue,
        questions: &Questions,
        home: &Path,
        queued: &[Task],
        stalled: &[Task],
        finished: &[Task],
        max_attempts: usize,
    ) {
        let snapshot = Self::snapshot(queue, stalled, finished);
        if self.last_seen.as_ref() == Some(&snapshot) {
            return;
        }
        self.last_seen = Some(snapshot);
        if let Err(e) = self
            .run_once(
                cfg,
                repo,
                queue,
                questions,
                home,
                queued,
                stalled,
                finished,
                max_attempts,
            )
            .await
        {
            tracing::warn!("conductor: {e:#}");
        }
    }

    #[allow(clippy::too_many_arguments)]
    async fn run_once(
        &mut self,
        cfg: &Config,
        repo: &Path,
        queue: &Queue,
        questions: &Questions,
        home: &Path,
        queued: &[Task],
        stalled: &[Task],
        finished: &[Task],
        max_attempts: usize,
    ) -> Result<()> {
        if queued.is_empty() && stalled.is_empty() && finished.is_empty() {
            return Ok(());
        }

        let spec = cfg
            .resolve_roles()
            .context("resolving the conductor seat")?
            .conductor;
        let needs_new_seat = !matches!(&self.seat, Some(s) if s.agent == spec.id);
        if needs_new_seat {
            self.seat = Some(SeatState::new(SEAT, &spec.id, crate::rng::entropy()));
        }
        let seat = self.seat.as_mut().expect("just ensured a seat exists");

        let runnable_views: Vec<prompt::ConductTask> =
            queued.iter().map(|t| view(t, max_attempts)).collect();
        let stalled_views: Vec<prompt::ConductTask> =
            stalled.iter().map(|t| view(t, max_attempts)).collect();
        let mut finished_views = Vec::with_capacity(finished.len());
        for t in finished {
            finished_views.push(finished_view(t, repo, max_attempts).await);
        }

        let body = prompt::with_overlay(
            prompt::conduct(
                &runnable_views,
                &stalled_views,
                &finished_views,
                &cfg.graph.language,
            ),
            cfg.prompts.overlay(NODE),
        );

        let artifacts = home.join("conduct").join("artifacts");
        let stem = format!("turn-{}", seat.turns + 1);
        // Bound to a local: `Invocation` only borrows the cache path, and the
        // `Option<PathBuf>` `cache_dir()` returns has to outlive that borrow.
        let cache_dir = cfg.cache_dir();
        let inv = Invocation {
            cwd: repo,
            prompt: &body,
            timeout: TURN_TIMEOUT,
            // The conductor never edits anything - it only decides what
            // blocks a task and what to do about one stuck or finished.
            allow_write: false,
            sessions: cfg.graph.sessions,
            artifacts: &artifacts,
            stem: &stem,
            run: NODE,
            node: NODE,
            cache_dir: cache_dir.as_deref(),
            attachments: &[],
        };

        let out = agent::invoke(&spec, seat, &inv)
            .await
            .context("invoking the conductor")?;
        if !out.usable() {
            bail!(
                "no usable reply (exit {:?}, timed out {})",
                out.exit_code,
                out.timed_out
            );
        }
        let verdict: Verdict = verdict::extract_json(&out.text)
            .context("the conductor's reply could not be parsed")?;
        apply(queue, questions, &verdict)
    }
}

#[cfg(test)]
mod tests {
    use std::collections::BTreeMap;

    use tempfile::tempdir;

    use super::*;
    use crate::ask::{Answer, QuestionStatus};
    use crate::config::{AgentKind, AgentSpec, Graph};
    use crate::queue::Source;

    fn mock_agent(dir: &Path, script: &str, env: BTreeMap<String, String>) -> AgentSpec {
        let path = dir.join("mock-conduct-agent.sh");
        std::fs::write(&path, script).expect("write mock");
        AgentSpec {
            id: "mock".to_owned(),
            kind: AgentKind::Command,
            model: None,
            command: vec!["sh".to_owned(), path.to_string_lossy().into_owned()],
            extra_args: Vec::new(),
            env,
            prompt_delivery: None,
        }
    }

    fn config(spec: AgentSpec) -> Config {
        Config {
            agents: vec![spec],
            graph: Graph {
                language: "en".to_owned(),
                ..Graph::default()
            },
            ..Config::default()
        }
    }

    fn task(title: &str) -> Task {
        Task::new(
            title.to_owned(),
            format!("do {title}"),
            std::path::PathBuf::from("."),
            Source::Human,
        )
    }

    const BROKEN: &str = "#!/bin/sh\ncat >/dev/null\nexit 3\n";
    const GARBAGE: &str = "#!/bin/sh\ncat >/dev/null\nprintf 'not json at all\\n'\n";

    fn env(reply: &str) -> BTreeMap<String, String> {
        BTreeMap::from([("MOCK_REPLY".to_owned(), reply.to_owned())])
    }

    const REPLY: &str = "#!/bin/sh\ncat >/dev/null\nprintf '%s\\n' \"$MOCK_REPLY\"\n";

    /// A throwaway repo with one commit on `main` and a second branch ahead
    /// of it, so `outcome_for`'s own `git::rev_parse` call has a real head to
    /// resolve.
    fn init_repo_with_branch(dir: &Path, branch: &str) {
        use crate::proc::Quiet as _;
        let run = |args: &[&str]| {
            let out = std::process::Command::new("git")
                .args(args)
                .current_dir(dir)
                .quiet()
                .output()
                .expect("spawn git");
            assert!(
                out.status.success(),
                "git {args:?} failed: {}",
                String::from_utf8_lossy(&out.stderr)
            );
        };
        run(&["init", "-b", "main"]);
        run(&["config", "user.name", "magi test"]);
        run(&["config", "user.email", "magi@example.com"]);
        std::fs::write(dir.join("README.md"), "# fixture\n").unwrap();
        run(&["add", "-A"]);
        run(&["commit", "-m", "init"]);
        run(&["checkout", "-b", branch]);
        std::fs::write(dir.join("change.txt"), "x\n").unwrap();
        run(&["add", "-A"]);
        run(&["commit", "-m", "candidate work"]);
    }

    fn review_round_with_finding(
        round: usize,
        finding_id: &str,
        title: &str,
        addressed: &[&str],
        rejected: &[(&str, &str)],
    ) -> crate::run::ReviewRound {
        crate::run::ReviewRound {
            round,
            head: "deadbeef".to_owned(),
            verified_head: None,
            verified_at: None,
            reviews: vec![crate::run::ReviewRecord {
                attempts: 0,
                reviewer: 1,
                agent: "mock".to_owned(),
                summary: String::new(),
                findings: vec![crate::verdict::Finding {
                    id: finding_id.to_owned(),
                    severity: crate::verdict::Severity::Major,
                    file: None,
                    line: None,
                    title: title.to_owned(),
                    detail: String::new(),
                }],
                vote: None,
                failed: None,
                duration_ms: 0,
            }],
            e2e: Vec::new(),
            verify_retried: false,
            e2e_deferred: false,
            e2e_defer_reason: None,
            fix: Some(crate::run::FixRecord {
                agent: "mock".to_owned(),
                addressed: addressed.iter().map(|s| (*s).to_owned()).collect(),
                rejected: rejected
                    .iter()
                    .map(|(id, why)| crate::verdict::Rejection {
                        id: (*id).to_owned(),
                        why: (*why).to_owned(),
                    })
                    .collect(),
                notes: String::new(),
                committed: false,
                failed: None,
                duration_ms: 0,
                continuation: None,
            }),
            blocking: 1,
            answered: 1,
            expected: 1,
            clean: false,
            progressed: true,
            vote_split: false,
            reconsideration: Vec::new(),
            verdict: None,
        }
    }

    #[test]
    fn outcome_for_carries_every_rounds_findings_and_the_branch_head() {
        crate::run::set_home(std::env::temp_dir().join("magi-conduct-tests-home"));
        let dir = tempdir().unwrap();
        let default_repo = dir.path().join("default");
        let task_repo = dir.path().join("task");
        std::fs::create_dir_all(&default_repo).unwrap();
        std::fs::create_dir_all(&task_repo).unwrap();
        init_repo_with_branch(&default_repo, "other-branch");
        init_repo_with_branch(&task_repo, "magi/f00d/A");

        let mut config = Config::default();
        config.graph.review_rounds = 6;
        let mut state = crate::run::RunState::new(
            task_repo.clone(),
            "main".to_owned(),
            "deadbeef".to_owned(),
            "task".to_owned(),
            config,
        );
        state.status = crate::run::RunStatus::Blocked;
        state.candidates.push(crate::run::Candidate {
            index: 0,
            label: 'A',
            agent: "mock".to_owned(),
            branch: "magi/f00d/A".to_owned(),
            worktree: task_repo.clone(),
            summary: String::new(),
            stat: String::new(),
            files: 1,
            commits: 1,
            empty: false,
            failed: None,
            verified_noop: None,
            duration_ms: 0,
            folded: false,
        });
        state.tally = Some(crate::run::Tally {
            first_choice: std::collections::BTreeMap::new(),
            borda: std::collections::BTreeMap::new(),
            winner: 'A',
            rankings: 0,
            unanimous_initial: false,
            deliberated: false,
            changed_votes: 0,
            unanimous_final: false,
            tie_break: None,
            judges: 0,
            present: 0,
            quorum: 0,
            met_quorum: true,
            uncontested: Some("solo".to_owned()),
        });
        state.reviews = vec![
            review_round_with_finding(
                1,
                "R1-1-2",
                "answer content is dropped",
                &[],
                &[("R1-1-2", "the id leaving blocked_by is enough")],
            ),
            review_round_with_finding(2, "R2-1-3", "answer content is still dropped", &[], &[]),
        ];
        state.save().unwrap();

        let mut t = task("outcome test");
        t.repo = task_repo;
        t.runs.push(state.id.clone());

        let finished = tokio_test_block_on(finished_view(&t, &default_repo, 2));
        let outcome = finished.outcome;

        assert!(outcome.unreadable.is_none());
        assert_eq!(outcome.run_status.as_deref(), Some("blocked"));
        assert_eq!(outcome.rounds_used, 2);
        assert_eq!(outcome.rounds_max, 6);
        assert_eq!(outcome.rounds.len(), 2);
        assert_eq!(outcome.rounds[0].findings[0].id, "R1-1-2");
        assert_eq!(outcome.rounds[0].rejected[0].id, "R1-1-2");
        assert!(outcome.rounds[1].addressed.is_empty());
        assert!(outcome.rounds[1].rejected.is_empty());
        assert_eq!(outcome.branch.as_deref(), Some("magi/f00d/A"));
        assert!(
            outcome.branch_head.is_some(),
            "a real branch must resolve a head commit: {outcome:?}"
        );
    }

    /// A tiny single-threaded block-on, so an `async fn` can be exercised
    /// from a plain `#[test]` without pulling `tokio::test`'s multi-thread
    /// runtime into a test that does no other async work.
    fn tokio_test_block_on<F: std::future::Future>(f: F) -> F::Output {
        tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap()
            .block_on(f)
    }

    #[test]
    fn view_carries_a_tasks_recorded_answers_into_the_conductor_prompt_input() {
        let mut t = task("answered");
        t.record_answer("Which backend?".to_owned(), "SQLite".to_owned());
        let v = view(&t, 2);
        assert_eq!(v.answers.len(), 1);
        assert_eq!(v.answers[0].question, "Which backend?");
        assert_eq!(v.answers[0].answer, "SQLite");
    }

    #[test]
    fn a_dependency_decision_blocks_the_task_and_leaves_priority_alone() {
        let dir = tempdir().unwrap();
        let queue = Queue::at(dir.path().join("queue"));
        let questions = Questions::at(dir.path().join("questions"));
        let mut a = task("a");
        a.priority = 9;
        queue.put(&mut a).unwrap();

        let verdict = Verdict {
            decisions: vec![Decision {
                id: a.id.clone(),
                blocked_by: vec!["20260101-000000-dead".to_owned()],
                reason: Some("waits on the other task".to_owned()),
                recovery: None,
                question: None,
                choices: Vec::new(),
            }],
        };
        apply(&queue, &questions, &verdict).unwrap();

        let back = queue.get(&a.id).unwrap();
        assert_eq!(back.status, TaskStatus::Blocked);
        assert_eq!(back.blocked_by, ["20260101-000000-dead"]);
        assert_eq!(
            back.priority, 9,
            "the conductor's reply cannot carry priority"
        );
    }

    #[test]
    fn a_question_decision_files_one_and_blocks_on_its_id() {
        let dir = tempdir().unwrap();
        let queue = Queue::at(dir.path().join("queue"));
        let questions = Questions::at(dir.path().join("questions"));
        let mut t = task("ambiguous");
        queue.put(&mut t).unwrap();

        let verdict = Verdict {
            decisions: vec![Decision {
                id: t.id.clone(),
                blocked_by: Vec::new(),
                reason: Some("which backend?".to_owned()),
                recovery: None,
                question: Some("Which storage backend?".to_owned()),
                choices: vec!["SQLite".to_owned(), "Redis".to_owned()],
            }],
        };
        apply(&queue, &questions, &verdict).unwrap();

        let back = queue.get(&t.id).unwrap();
        assert_eq!(back.status, TaskStatus::Blocked);
        assert_eq!(back.blocked_by.len(), 1);
        let q = questions.get(&back.blocked_by[0]).unwrap();
        assert_eq!(q.summary, "Which storage backend?");
        assert_eq!(q.node, NODE);
        assert!(q.status.open());
    }

    #[test]
    fn a_task_with_an_open_question_already_reuses_it_rather_than_filing_a_second_one() {
        let dir = tempdir().unwrap();
        let queue = Queue::at(dir.path().join("queue"));
        let questions = Questions::at(dir.path().join("questions"));
        let mut t = task("asked once");
        queue.put(&mut t).unwrap();

        let decision = Decision {
            id: t.id.clone(),
            reason: Some("still deciding".to_owned()),
            question: Some("Which backend?".to_owned()),
            ..Decision::default()
        };
        apply(
            &queue,
            &questions,
            &Verdict {
                decisions: vec![decision.clone()],
            },
        )
        .unwrap();
        assert_eq!(questions.list().len(), 1);
        let first_question_id = queue.get(&t.id).unwrap().blocked_by[0].clone();

        // An operator releasing the blocked task by hand, without answering,
        // puts it back at `Queued` while the question stays open - exactly
        // the case the guard in `apply_one` exists for: a later cycle
        // proposing the very same question must reuse it, not file a second.
        let mut released = queue.get(&t.id).unwrap();
        released.release();
        queue.put(&mut released).unwrap();

        apply(
            &queue,
            &questions,
            &Verdict {
                decisions: vec![decision],
            },
        )
        .unwrap();
        assert_eq!(questions.list().len(), 1, "no duplicate question was filed");
        let after = queue.get(&t.id).unwrap();
        assert_eq!(
            after.blocked_by,
            [first_question_id],
            "the existing open question is reused, not replaced"
        );
    }

    #[test]
    fn a_same_id_question_from_another_node_is_not_reused() {
        let dir = tempdir().unwrap();
        let queue = Queue::at(dir.path().join("queue"));
        let questions = Questions::at(dir.path().join("questions"));
        let mut t = task("must ask the conductor");
        queue.put(&mut t).unwrap();

        let mut unrelated = Question::new(
            t.id.clone(),
            "review".to_owned(),
            "reviewer-1".to_owned(),
            "An unrelated review question".to_owned(),
            String::new(),
            Vec::new(),
        );
        questions.put(&mut unrelated).unwrap();

        apply(
            &queue,
            &questions,
            &Verdict {
                decisions: vec![Decision {
                    id: t.id.clone(),
                    question: Some("Which backend?".to_owned()),
                    ..Decision::default()
                }],
            },
        )
        .unwrap();

        let blocked_by = &queue.get(&t.id).unwrap().blocked_by;
        assert_eq!(blocked_by.len(), 1);
        assert_ne!(blocked_by[0], unrelated.id);
        assert!(questions.get(&unrelated.id).unwrap().status.open());
        assert_eq!(questions.get(&blocked_by[0]).unwrap().node, NODE);
    }

    #[test]
    fn answering_the_question_lets_the_resolver_clear_the_block_with_the_answer_kept() {
        let dir = tempdir().unwrap();
        let queue = Queue::at(dir.path().join("queue"));
        let questions = Questions::at(dir.path().join("questions"));
        let mut t = task("waits on an answer");
        queue.put(&mut t).unwrap();

        apply(
            &queue,
            &questions,
            &Verdict {
                decisions: vec![Decision {
                    id: t.id.clone(),
                    blocked_by: Vec::new(),
                    reason: None,
                    recovery: None,
                    question: Some("Which backend?".to_owned()),
                    choices: Vec::new(),
                }],
            },
        )
        .unwrap();
        let blocked = queue.get(&t.id).unwrap();
        let question_id = blocked.blocked_by[0].clone();

        let mut q = questions.get(&question_id).unwrap();
        q.answer(Answer::Text("SQLite".to_owned())).unwrap();
        questions.put(&mut q).unwrap();
        assert_eq!(q.status, QuestionStatus::Answered);

        // `crate::daemon::resolve_blockers` is the deterministic resolver
        // that actually does this on the real queue; here it is enough to
        // prove the pure steps it is built from behave together.
        let mut task_after = queue.get(&t.id).unwrap();
        task_after.record_answer(q.summary.clone(), "SQLite".to_owned());
        task_after.unblock(&question_id);
        assert_eq!(task_after.status, TaskStatus::Queued);
        assert_eq!(task_after.answers[0].answer, "SQLite");
    }

    #[test]
    fn a_stalled_task_can_be_requeued_or_held() {
        let dir = tempdir().unwrap();
        let queue = Queue::at(dir.path().join("queue"));
        let questions = Questions::at(dir.path().join("questions"));

        let mut requeue_me = task("stuck a");
        requeue_me.start("run-1".to_owned());
        queue.put(&mut requeue_me).unwrap();

        let mut hold_me = task("stuck b");
        hold_me.start("run-2".to_owned());
        queue.put(&mut hold_me).unwrap();

        apply(
            &queue,
            &questions,
            &Verdict {
                decisions: vec![
                    Decision {
                        id: requeue_me.id.clone(),
                        recovery: Some(Recovery::Requeue),
                        ..Decision::default()
                    },
                    Decision {
                        id: hold_me.id.clone(),
                        recovery: Some(Recovery::Hold),
                        reason: Some("looks broken".to_owned()),
                        ..Decision::default()
                    },
                ],
            },
        )
        .unwrap();

        let requeued = queue.get(&requeue_me.id).unwrap();
        assert_eq!(requeued.status, TaskStatus::Queued);
        assert_eq!(requeued.attempts, 0);

        let held = queue.get(&hold_me.id).unwrap();
        assert_eq!(held.status, TaskStatus::Held);
        assert_eq!(held.hold_reason.as_deref(), Some("looks broken"));
    }

    #[test]
    fn a_machine_held_task_asked_about_restores_to_held_once_answered() {
        // Reproduces the reported bug: a task held after burning its
        // attempts, once the conductor asks a follow-up question about it,
        // must come back `held` - not `queued` - once the question is
        // answered, whatever the answer said.
        let dir = tempdir().unwrap();
        let queue = Queue::at(dir.path().join("queue"));
        let questions = Questions::at(dir.path().join("questions"));
        let mut t = task("held out of attempts");
        t.hold_machine(Some("out of attempts".to_owned()));
        queue.put(&mut t).unwrap();

        apply(
            &queue,
            &questions,
            &Verdict {
                decisions: vec![Decision {
                    id: t.id.clone(),
                    reason: Some("what should happen to this one?".to_owned()),
                    question: Some("Hold it, or try again?".to_owned()),
                    ..Decision::default()
                }],
            },
        )
        .unwrap();
        let blocked = queue.get(&t.id).unwrap();
        assert_eq!(blocked.status, TaskStatus::Blocked);
        let question_id = blocked.blocked_by[0].clone();

        let mut q = questions.get(&question_id).unwrap();
        q.answer(Answer::Text("leave it held".to_owned())).unwrap();
        questions.put(&mut q).unwrap();

        // What `crate::daemon::resolve_blockers` does on the real queue.
        let mut after = queue.get(&t.id).unwrap();
        after.record_answer(q.summary.clone(), "leave it held".to_owned());
        after.unblock(&question_id);
        assert_eq!(
            after.status,
            TaskStatus::Held,
            "must not fall back to queued"
        );
        assert_eq!(after.hold_reason.as_deref(), Some("out of attempts"));
    }

    #[test]
    fn a_reaffirmed_hold_with_no_new_reason_is_not_silently_auto_released_by_triage() {
        // A different path to the same bug round 1 fixed: a task
        // machine-held for disk pressure, blocked on a conductor question,
        // answered "keep it held", and restored to `held`. If the
        // conductor's next decision reconfirms `recovery: hold` with no new
        // `reason` - allowed, `Decision::reason` is optional - `hold_machine`
        // must not silently leave the stale disk-pressure text in place, or
        // `crate::triage::run_once` reads it as an unexamined, auto-resolvable
        // hold and releases the task straight through the operator's answer
        // the instant disk space looks fine again.
        let dir = tempdir().unwrap();
        let queue = Queue::at(dir.path().join("queue"));
        let questions = Questions::at(dir.path().join("questions"));
        let mut t = task("disk pressure, then reconsidered");
        t.hold_machine(Some(
            "not enough free space to start a run: 10 bytes free, 100 required by \
             `[disk] min_free_bytes`"
                .to_owned(),
        ));
        t.record_answer(
            "How should this be handled?".to_owned(),
            "keep it held, a human will look at it later".to_owned(),
        );
        queue.put(&mut t).unwrap();

        apply(
            &queue,
            &questions,
            &Verdict {
                decisions: vec![Decision {
                    id: t.id.clone(),
                    recovery: Some(Recovery::Hold),
                    ..Decision::default()
                }],
            },
        )
        .unwrap();

        let after = queue.get(&t.id).unwrap();
        assert_eq!(after.status, TaskStatus::Held);
        assert!(
            !after
                .hold_reason
                .as_deref()
                .unwrap_or_default()
                .starts_with("not enough free space"),
            "the stale disk-pressure text must not survive a reconfirmed hold: {:?}",
            after.hold_reason
        );

        // The disk gate would report space is fine now - `crate::triage`
        // must not read the old text and release the task through it.
        let cfg_dir = tempdir().unwrap();
        let config = cfg_dir.path().join("magi.toml");
        std::fs::write(&config, "[disk]\nmin_free_bytes = 0\n").unwrap();
        let report =
            crate::triage::run_once(&queue, &questions, Some(&config), jiff::Timestamp::now());
        assert!(report.resumed.is_empty(), "must not be auto-released");
        assert_eq!(queue.get(&t.id).unwrap().status, TaskStatus::Held);
    }

    #[test]
    fn a_conductor_rehold_after_a_resume_answer_is_not_asked_again_identically() {
        let dir = tempdir().unwrap();
        let queue = Queue::at(dir.path().join("queue"));
        let questions = Questions::at(dir.path().join("questions"));
        let config = dir.path().join("magi.toml");
        std::fs::write(&config, "[disk]\nmin_free_bytes = 0\n").unwrap();
        let now = jiff::Timestamp::now();
        let triage = || crate::triage::run_once(&queue, &questions, Some(&config), now);
        let open = || {
            questions
                .list()
                .into_iter()
                .filter(|q| q.node == "triage" && q.status.open())
                .collect::<Vec<_>>()
        };
        let hold = Decision {
            recovery: Some(Recovery::Hold),
            reason: Some("waiting on manual worktree cleanup".to_owned()),
            ..Decision::default()
        };

        let mut t = task("looping hold");
        t.hold_machine(Some("waiting on manual worktree cleanup".to_owned()));
        queue.put(&mut t).unwrap();
        let hold = Decision {
            id: t.id.clone(),
            ..hold
        };

        // 1. triage asks the ordinary machine question; the operator resumes.
        assert_eq!(triage().asked.len(), 1);
        let first = open().remove(0);
        let mut q = questions.get(&first.id).unwrap();
        let resume = q.choices[0].clone();
        q.answer(Answer::Choice(resume)).unwrap();
        questions.put(&mut q).unwrap();
        assert_eq!(triage().answered.len(), 1);
        assert_eq!(queue.get(&t.id).unwrap().status, TaskStatus::Queued);

        // 2. the conductor holds it again - allowed once, and recorded.
        apply(
            &queue,
            &questions,
            &Verdict {
                decisions: vec![hold.clone()],
            },
        )
        .unwrap();
        assert_eq!(queue.get(&t.id).unwrap().status, TaskStatus::Held);

        // 3. triage must say something different, once.
        assert_eq!(triage().asked.len(), 1);
        let second = open().remove(0);
        assert_ne!(second.summary, first.summary);
        assert_ne!(second.choices, first.choices);
        assert!(second.detail.contains("waiting on manual worktree cleanup"));
        assert!(triage().asked.is_empty(), "no duplicate question");
        assert_eq!(open().len(), 1);

        // 4. forcing a requeue makes the conductor's hold a no-op.
        let mut q = questions.get(&second.id).unwrap();
        let force = q.choices[0].clone();
        q.answer(Answer::Choice(force)).unwrap();
        questions.put(&mut q).unwrap();
        assert_eq!(triage().answered.len(), 1);
        apply(
            &queue,
            &questions,
            &Verdict {
                decisions: vec![hold],
            },
        )
        .unwrap();
        assert_eq!(queue.get(&t.id).unwrap().status, TaskStatus::Queued);
    }

    #[test]
    fn a_runnable_task_can_be_held_directly_without_a_question() {
        let dir = tempdir().unwrap();
        let queue = Queue::at(dir.path().join("queue"));
        let questions = Questions::at(dir.path().join("questions"));
        let mut t = task("already answered, should stay put");
        queue.put(&mut t).unwrap();

        apply(
            &queue,
            &questions,
            &Verdict {
                decisions: vec![Decision {
                    id: t.id.clone(),
                    recovery: Some(Recovery::Hold),
                    reason: Some("operator already said keep this held".to_owned()),
                    ..Decision::default()
                }],
            },
        )
        .unwrap();

        let after = queue.get(&t.id).unwrap();
        assert_eq!(after.status, TaskStatus::Held);
        assert_eq!(after.hold_source, Some(crate::queue::HoldSource::Machine));
    }

    #[test]
    fn done_recovery_closes_a_held_task_whose_goal_is_already_met() {
        // Reproduces the other half of the reported bug (task 6081): once
        // the operator's answer says the work already happened outside the
        // loop - PR merged, worktree cleaned up - the conductor needs an
        // actual terminal state to put the task in, not just a hold it will
        // keep being re-asked about.
        let dir = tempdir().unwrap();
        let queue = Queue::at(dir.path().join("queue"));
        let questions = Questions::at(dir.path().join("questions"));
        let mut t = task("already merged by hand");
        t.hold_machine(Some("branch survived, awaiting a decision".to_owned()));
        t.record_answer(
            "Handle this one?".to_owned(),
            "already merged and cleaned up, close it".to_owned(),
        );
        queue.put(&mut t).unwrap();

        apply(
            &queue,
            &questions,
            &Verdict {
                decisions: vec![Decision {
                    id: t.id.clone(),
                    recovery: Some(Recovery::Done),
                    reason: Some("operator confirmed this already landed".to_owned()),
                    ..Decision::default()
                }],
            },
        )
        .unwrap();

        let after = queue.get(&t.id).unwrap();
        assert_eq!(after.status, TaskStatus::Done);
        assert!(after.hold_reason.is_none());
        assert_eq!(after.answers.len(), 1, "the record of why is kept");
    }

    #[test]
    fn done_recovery_is_ignored_for_a_runnable_or_running_task() {
        let dir = tempdir().unwrap();
        let queue = Queue::at(dir.path().join("queue"));
        let questions = Questions::at(dir.path().join("questions"));

        let mut queued = task("never ran yet");
        queue.put(&mut queued).unwrap();

        let mut running = task("mid-run");
        running.start("run-1".to_owned());
        queue.put(&mut running).unwrap();

        for id in [queued.id.clone(), running.id.clone()] {
            apply(
                &queue,
                &questions,
                &Verdict {
                    decisions: vec![Decision {
                        id,
                        recovery: Some(Recovery::Done),
                        ..Decision::default()
                    }],
                },
            )
            .unwrap();
        }

        assert_eq!(queue.get(&queued.id).unwrap().status, TaskStatus::Queued);
        assert_eq!(queue.get(&running.id).unwrap().status, TaskStatus::Running);
    }

    #[test]
    fn a_third_conductor_question_after_two_settled_answers_holds_instead_of_asking_again() {
        // Guards against the model not registering its own question as
        // already answered and re-asking a version of it forever: once this
        // many of the task's `answers` already came from `crate::conduct`,
        // a further `question` decision is refused in favor of a hold - see
        // `MAX_SETTLED_CONDUCT_ANSWERS`'s own doc.
        let dir = tempdir().unwrap();
        let queue = Queue::at(dir.path().join("queue"));
        let questions = Questions::at(dir.path().join("questions"));
        let mut t = task("asked about repeatedly");
        t.hold_machine(Some("out of attempts".to_owned()));
        t.record_answer("Handle this one? (1)".to_owned(), "not yet".to_owned());
        t.record_answer(
            "Handle this one? (2)".to_owned(),
            "still not yet".to_owned(),
        );
        queue.put(&mut t).unwrap();
        assert_eq!(questions.list().len(), 0);

        apply(
            &queue,
            &questions,
            &Verdict {
                decisions: vec![Decision {
                    id: t.id.clone(),
                    question: Some("Handle this one? (3)".to_owned()),
                    ..Decision::default()
                }],
            },
        )
        .unwrap();

        assert_eq!(questions.list().len(), 0, "no third question was filed");
        let after = queue.get(&t.id).unwrap();
        assert_eq!(after.status, TaskStatus::Held);
        assert!(after.blocked_by.is_empty());
        assert_eq!(after.answers.len(), 2, "the prior answers are untouched");
    }

    #[test]
    fn a_second_conductor_question_is_still_allowed_after_one_settled_answer() {
        let dir = tempdir().unwrap();
        let queue = Queue::at(dir.path().join("queue"));
        let questions = Questions::at(dir.path().join("questions"));
        let mut t = task("asked about once already");
        t.hold_machine(Some("out of attempts".to_owned()));
        t.record_answer("Handle this one?".to_owned(), "not yet".to_owned());
        queue.put(&mut t).unwrap();

        apply(
            &queue,
            &questions,
            &Verdict {
                decisions: vec![Decision {
                    id: t.id.clone(),
                    question: Some("Still not sure - now what?".to_owned()),
                    ..Decision::default()
                }],
            },
        )
        .unwrap();

        assert_eq!(questions.list().len(), 1, "the second question was filed");
        assert_eq!(queue.get(&t.id).unwrap().status, TaskStatus::Blocked);
    }

    #[test]
    fn a_held_task_with_an_open_triage_question_is_left_to_triage() {
        // `crate::triage` never calls `Task::block`, and only ever looks at
        // tasks still `held` - so a conductor question that moved this task
        // to `blocked` would orphan triage's own open question. The guard in
        // `apply_one` must leave the task alone until triage's question
        // settles.
        let dir = tempdir().unwrap();
        let queue = Queue::at(dir.path().join("queue"));
        let questions = Questions::at(dir.path().join("questions"));
        let mut t = task("held, triage already asking about it");
        t.hold_machine(Some("cause unclear".to_owned()));
        queue.put(&mut t).unwrap();

        let mut triage_q = Question::new(
            t.id.clone(),
            crate::triage::NODE.to_owned(),
            "triage".to_owned(),
            "Still needed?".to_owned(),
            String::new(),
            vec![
                "resume".to_owned(),
                "not yet".to_owned(),
                "discard".to_owned(),
            ],
        );
        questions.put(&mut triage_q).unwrap();

        for decision in [
            Decision {
                id: t.id.clone(),
                question: Some("what now?".to_owned()),
                ..Decision::default()
            },
            Decision {
                id: t.id.clone(),
                recovery: Some(Recovery::Requeue),
                ..Decision::default()
            },
        ] {
            apply(
                &queue,
                &questions,
                &Verdict {
                    decisions: vec![decision],
                },
            )
            .unwrap();
        }

        let after = queue.get(&t.id).unwrap();
        assert_eq!(
            after.status,
            TaskStatus::Held,
            "triage still owns this hold"
        );
        assert!(after.blocked_by.is_empty());
        assert_eq!(
            questions.list().len(),
            1,
            "no second, conductor-owned question was filed"
        );
    }

    #[test]
    fn a_held_task_with_an_answered_but_unapplied_triage_question_is_still_left_alone() {
        // The race `crate::triage::pending_for` exists to close: the
        // operator has already answered the triage question (it is no
        // longer `open`), but `crate::triage::run_once` - which only runs on
        // a fully idle daemon tick, far less often than the conductor polls
        // - has not had a turn to apply it yet. `apply_one` must not treat
        // "not open" as "settled" here, or it would block the task out from
        // under an answer triage has not read back yet.
        let dir = tempdir().unwrap();
        let queue = Queue::at(dir.path().join("queue"));
        let questions = Questions::at(dir.path().join("questions"));
        let mut t = task("held, triage question answered but not yet applied");
        t.hold_machine(Some("cause unclear".to_owned()));
        queue.put(&mut t).unwrap();

        let mut triage_q = Question::new(
            t.id.clone(),
            crate::triage::NODE.to_owned(),
            "triage".to_owned(),
            "Still needed?".to_owned(),
            String::new(),
            vec![
                "resume".to_owned(),
                "not yet".to_owned(),
                "discard".to_owned(),
            ],
        );
        questions.put(&mut triage_q).unwrap();
        triage_q
            .answer(Answer::Choice("not yet".to_owned()))
            .unwrap();
        questions.put(&mut triage_q).unwrap();
        assert!(!triage_q.status.open());

        apply(
            &queue,
            &questions,
            &Verdict {
                decisions: vec![Decision {
                    id: t.id.clone(),
                    question: Some("what now?".to_owned()),
                    ..Decision::default()
                }],
            },
        )
        .unwrap();

        let after = queue.get(&t.id).unwrap();
        assert_eq!(
            after.status,
            TaskStatus::Held,
            "triage's own answer is not yet applied - conduct must wait"
        );
        assert_eq!(
            questions.list().len(),
            1,
            "no conductor question was filed over the pending triage answer"
        );
    }

    #[test]
    fn manual_hold_rejects_hostile_or_stale_conductor_recovery() {
        let dir = tempdir().unwrap();
        let queue = Queue::at(dir.path().join("queue"));
        let questions = Questions::at(dir.path().join("questions"));
        let mut held = task("manual recovery");
        held.priority = 300;
        held.runs.push("run20260912-224242-daf5".to_owned());
        held.hold_manual(Some(
            "active manual recovery run20260912-224242-daf5".to_owned(),
        ));
        queue.put(&mut held).unwrap();

        // Every field a conductor may use to alter lifecycle state is ignored:
        // requeue/review would dispatch duplicate work, hold could overwrite
        // evidence, and a question would turn the hold into `blocked`.
        for decision in [
            Decision {
                id: held.id.clone(),
                recovery: Some(Recovery::Requeue),
                ..Decision::default()
            },
            Decision {
                id: held.id.clone(),
                recovery: Some(Recovery::Hold),
                reason: Some("stale replacement reason".to_owned()),
                ..Decision::default()
            },
            Decision {
                id: held.id.clone(),
                recovery: Some(Recovery::Review),
                ..Decision::default()
            },
            Decision {
                id: held.id.clone(),
                blocked_by: vec!["other-task".to_owned()],
                question: Some("retry now?".to_owned()),
                ..Decision::default()
            },
        ] {
            apply(
                &queue,
                &questions,
                &Verdict {
                    decisions: vec![decision],
                },
            )
            .unwrap();
        }

        let after = queue.get(&held.id).unwrap();
        assert_eq!(after.status, TaskStatus::Held);
        assert!(after.operator_held());
        assert_eq!(after.priority, 300);
        assert_eq!(after.runs, ["run20260912-224242-daf5"]);
        assert_eq!(
            after.hold_reason.as_deref(),
            Some("active manual recovery run20260912-224242-daf5")
        );
        assert!(after.blocked_by.is_empty());
        assert!(questions.list().is_empty());
        assert!(
            queue.next_runnable().is_none(),
            "must not dispatch a duplicate"
        );
    }

    #[test]
    fn machine_holds_remain_recoverable_and_manual_release_is_authorization() {
        let dir = tempdir().unwrap();
        let queue = Queue::at(dir.path().join("queue"));
        let questions = Questions::at(dir.path().join("questions"));

        let mut automatic = task("disk gate");
        automatic.hold_machine(Some("disk full".to_owned()));
        queue.put(&mut automatic).unwrap();
        let requeue = || Verdict {
            decisions: vec![Decision {
                id: automatic.id.clone(),
                recovery: Some(Recovery::Requeue),
                ..Decision::default()
            }],
        };
        apply(&queue, &questions, &requeue()).unwrap();
        assert_eq!(queue.get(&automatic.id).unwrap().status, TaskStatus::Queued);

        let mut manual = task("operator gate");
        manual.hold_manual(Some("wait for operator".to_owned()));
        queue.put(&mut manual).unwrap();
        apply(
            &queue,
            &questions,
            &Verdict {
                decisions: vec![Decision {
                    id: manual.id.clone(),
                    recovery: Some(Recovery::Requeue),
                    ..Decision::default()
                }],
            },
        )
        .unwrap();
        assert_eq!(queue.get(&manual.id).unwrap().status, TaskStatus::Held);

        // This mirrors the CLI and web release routes: only an explicit
        // operator action clears the manual boundary.
        let mut released = queue.get(&manual.id).unwrap();
        released.release();
        queue.put(&mut released).unwrap();
        assert_eq!(queue.get(&manual.id).unwrap().status, TaskStatus::Queued);
    }

    #[test]
    fn legacy_reasoned_hold_is_protected_without_losing_its_metadata() {
        let dir = tempdir().unwrap();
        let queue = Queue::at(dir.path().join("queue"));
        let questions = Questions::at(dir.path().join("questions"));
        let mut legacy = task("old explicit hold");
        legacy.status = TaskStatus::Held;
        legacy.hold_reason = Some("manual recovery already active".to_owned());
        legacy.hold_source = None;
        legacy.blocked_by = vec!["dependency".to_owned()];
        queue.put(&mut legacy).unwrap();

        apply(
            &queue,
            &questions,
            &Verdict {
                decisions: vec![Decision {
                    id: legacy.id.clone(),
                    recovery: Some(Recovery::Requeue),
                    ..Decision::default()
                }],
            },
        )
        .unwrap();

        let after = queue.get(&legacy.id).unwrap();
        assert_eq!(after.status, TaskStatus::Held);
        assert_eq!(after.hold_source, None);
        assert_eq!(after.hold_reason, legacy.hold_reason);
        assert_eq!(after.blocked_by, legacy.blocked_by);
    }

    #[test]
    fn review_recovery_is_a_no_op_without_a_survivable_branch() {
        // `surviving_branch` reaches `RunState::load`, which reaches the
        // process-global `run::home()` - a `OnceLock`, so this only wins the
        // race the first time it runs in the binary; every other test still
        // reaches the same directory whichever call won, and this test's own
        // run id never collides with another test's.
        crate::run::set_home(std::env::temp_dir().join("magi-conduct-tests-home"));
        let dir = tempdir().unwrap();
        let queue = Queue::at(dir.path().join("queue"));
        let questions = Questions::at(dir.path().join("questions"));
        let mut t = task("blocked with no readable run");
        t.start("20260101-000000-dead".to_owned()); // no such run on disk
        t.fail("blocked", 5);
        queue.put(&mut t).unwrap();

        apply(
            &queue,
            &questions,
            &Verdict {
                decisions: vec![Decision {
                    id: t.id.clone(),
                    recovery: Some(Recovery::Review),
                    ..Decision::default()
                }],
            },
        )
        .unwrap();

        let after = queue.get(&t.id).unwrap();
        assert_eq!(
            after.status,
            TaskStatus::Failed,
            "with nothing to reopen, the decision is dropped rather than guessed at"
        );
        assert!(after.review_branch.is_none());
    }

    #[test]
    fn requeue_and_review_recovery_are_ignored_for_a_runnable_task() {
        // `Hold` is the one exception - see `Decision::recovery`'s doc and
        // `a_runnable_task_can_be_held_directly_without_a_question` - but a
        // task already in line has nothing for `requeue` or `review` to do.
        let dir = tempdir().unwrap();
        let queue = Queue::at(dir.path().join("queue"));
        let questions = Questions::at(dir.path().join("questions"));

        for recovery in [Recovery::Requeue, Recovery::Review] {
            let mut t = task("ordinary");
            queue.put(&mut t).unwrap();

            apply(
                &queue,
                &questions,
                &Verdict {
                    decisions: vec![Decision {
                        id: t.id.clone(),
                        recovery: Some(recovery),
                        ..Decision::default()
                    }],
                },
            )
            .unwrap();

            assert_eq!(queue.get(&t.id).unwrap().status, TaskStatus::Queued);
        }
    }

    #[tokio::test]
    async fn a_broken_agent_leaves_the_queue_untouched_and_does_not_error() {
        let dir = tempdir().unwrap();
        let cfg = config(mock_agent(dir.path(), BROKEN, BTreeMap::new()));
        let queue = Queue::at(dir.path().join("queue"));
        let questions = Questions::at(dir.path().join("questions"));
        let mut t = task("normal");
        queue.put(&mut t).unwrap();

        let mut conductor = Conductor::new();
        conductor
            .maybe_run(
                &cfg,
                dir.path(),
                &queue,
                &questions,
                dir.path(),
                &[t.clone()],
                &[],
                &[],
                2,
            )
            .await;

        assert_eq!(
            queue.get(&t.id).unwrap().status,
            TaskStatus::Queued,
            "a failed invocation must change nothing"
        );
        assert!(
            queue.next_runnable().is_some(),
            "the loop must still be able to take the next task"
        );
    }

    #[tokio::test]
    async fn a_reply_with_no_json_leaves_the_queue_untouched() {
        let dir = tempdir().unwrap();
        let cfg = config(mock_agent(dir.path(), GARBAGE, BTreeMap::new()));
        let queue = Queue::at(dir.path().join("queue"));
        let questions = Questions::at(dir.path().join("questions"));
        let mut t = task("normal");
        queue.put(&mut t).unwrap();

        let mut conductor = Conductor::new();
        conductor
            .maybe_run(
                &cfg,
                dir.path(),
                &queue,
                &questions,
                dir.path(),
                &[t.clone()],
                &[],
                &[],
                2,
            )
            .await;

        assert_eq!(queue.get(&t.id).unwrap().status, TaskStatus::Queued);
    }

    #[tokio::test]
    async fn json_survives_code_fences_and_a_preamble() {
        let dir = tempdir().unwrap();
        let mut t = task("fenced");
        let reply = format!(
            "Sure, here is my decision.\n\n```json\n{{\"decisions\":[{{\"id\":\"{}\",\
             \"blocked_by\":[\"x\"],\"reason\":\"why\"}}]}}\n```\n",
            t.id
        );
        let cfg = config(mock_agent(dir.path(), REPLY, env(&reply)));
        let queue = Queue::at(dir.path().join("queue"));
        let questions = Questions::at(dir.path().join("questions"));
        queue.put(&mut t).unwrap();

        let mut conductor = Conductor::new();
        conductor
            .maybe_run(
                &cfg,
                dir.path(),
                &queue,
                &questions,
                dir.path(),
                &[t.clone()],
                &[],
                &[],
                2,
            )
            .await;

        let back = queue.get(&t.id).unwrap();
        assert_eq!(back.status, TaskStatus::Blocked);
        assert_eq!(back.blocked_by, ["x"]);
    }

    #[tokio::test]
    async fn the_conductor_is_not_called_again_when_nothing_worth_looking_at_has_changed() {
        // Each real invocation writes its own artifact stem, `turn-<n>`, so
        // whether a second one happened is read off the artifacts directory.
        let dir = tempdir().unwrap();
        let cfg = config(mock_agent(dir.path(), REPLY, env("{\"decisions\":[]}")));
        let queue = Queue::at(dir.path().join("queue"));
        let questions = Questions::at(dir.path().join("questions"));
        let mut t = task("stable");
        queue.put(&mut t).unwrap();
        let artifacts = dir.path().join("conduct").join("artifacts");
        let turn = |n: usize| artifacts.join(format!("turn-{n}.out"));

        let mut conductor = Conductor::new();
        conductor
            .maybe_run(
                &cfg,
                dir.path(),
                &queue,
                &questions,
                dir.path(),
                &[t.clone()],
                &[],
                &[],
                2,
            )
            .await;
        assert!(turn(1).is_file(), "the first cycle must call the conductor");

        conductor
            .maybe_run(
                &cfg,
                dir.path(),
                &queue,
                &questions,
                dir.path(),
                &[t.clone()],
                &[],
                &[],
                2,
            )
            .await;
        assert!(
            !turn(2).is_file(),
            "an unchanged revision and an unchanged stalled/finished set must not call the \
             conductor twice"
        );

        // Once the queue actually changes, the next `maybe_run` calls again.
        t.priority = 1;
        queue.put(&mut t).unwrap();
        conductor
            .maybe_run(
                &cfg,
                dir.path(),
                &queue,
                &questions,
                dir.path(),
                &[t.clone()],
                &[],
                &[],
                2,
            )
            .await;
        assert!(turn(2).is_file(), "a moved revision calls it again");
    }

    #[tokio::test]
    async fn a_task_turning_stalled_calls_the_conductor_again_despite_an_unchanged_revision() {
        // The queue's own revision has not moved - nothing wrote to it - but
        // a task now looks stalled, purely because time passed. Calling
        // again here, and never again once this exact set has been shown
        // once, is the whole point of keying `worth_a_look` on the id set
        // rather than on "is it non-empty".
        let dir = tempdir().unwrap();
        let cfg = config(mock_agent(dir.path(), REPLY, env("{\"decisions\":[]}")));
        let queue = Queue::at(dir.path().join("queue"));
        let questions = Questions::at(dir.path().join("questions"));
        let mut t = task("quiet");
        queue.put(&mut t).unwrap();
        let artifacts = dir.path().join("conduct").join("artifacts");
        let turn = |n: usize| artifacts.join(format!("turn-{n}.out"));

        let mut conductor = Conductor::new();
        conductor
            .maybe_run(
                &cfg,
                dir.path(),
                &queue,
                &questions,
                dir.path(),
                &[t.clone()],
                &[],
                &[],
                2,
            )
            .await;
        assert!(turn(1).is_file());

        conductor
            .maybe_run(
                &cfg,
                dir.path(),
                &queue,
                &questions,
                dir.path(),
                &[],
                &[t.clone()],
                &[],
                2,
            )
            .await;
        assert!(
            turn(2).is_file(),
            "a task turning stalled must call the conductor again"
        );

        // But once shown at this exact revision, showing the *same* stalled
        // set again must not call a third time.
        conductor
            .maybe_run(
                &cfg,
                dir.path(),
                &queue,
                &questions,
                dir.path(),
                &[],
                &[t.clone()],
                &[],
                2,
            )
            .await;
        assert!(
            !turn(3).is_file(),
            "the same stalled task lingering must not call the conductor every cycle"
        );
    }

    #[test]
    fn worth_a_look_is_config_free_and_matches_maybe_runs_own_gate() {
        let dir = tempdir().unwrap();
        let queue = Queue::at(dir.path().join("queue"));
        let mut t = task("t");
        queue.put(&mut t).unwrap();

        let mut conductor = Conductor::new();
        assert!(
            conductor.worth_a_look(&queue, &[], &[]),
            "a conductor that has never run has something to look at"
        );

        conductor.last_seen = Some(Conductor::snapshot(&queue, &[], &[]));
        assert!(
            !conductor.worth_a_look(&queue, &[], &[]),
            "nothing changed and nothing is stalled or finished"
        );
        assert!(
            conductor.worth_a_look(&queue, &[t.clone()], &[]),
            "a stalled task is worth a look even at the same revision"
        );
        assert!(
            conductor.worth_a_look(&queue, &[], &[t.clone()]),
            "a finished task is worth a look even at the same revision"
        );
    }

    #[tokio::test]
    async fn the_conduct_path_never_calls_ask_and_wait() {
        // Structural: grepping this module and `daemon.rs` for
        // `ask_and_wait` is the actual assertion this module's own doc
        // promises; this test exists so the promise has a name in the test
        // output too. `apply_one`'s question path uses `Questions::put`
        // exclusively.
        let dir = tempdir().unwrap();
        let queue = Queue::at(dir.path().join("queue"));
        let questions = Questions::at(dir.path().join("questions"));
        let mut t = task("asks without blocking");
        queue.put(&mut t).unwrap();

        apply(
            &queue,
            &questions,
            &Verdict {
                decisions: vec![Decision {
                    id: t.id.clone(),
                    question: Some("ok?".to_owned()),
                    ..Decision::default()
                }],
            },
        )
        .unwrap();
        // Reaching here at all (no hang) is the assertion.
        assert_eq!(queue.get(&t.id).unwrap().status, TaskStatus::Blocked);
    }
}