magi/

handover.rs

//! Taking a branch over from an earlier attempt at the same task.
//!
//! A retried task can reopen a branch (`magi/f82f/A`) that the *previous*
//! run still has checked out in its own worktree. Git holds a branch in one
//! worktree at a time, so the new run's review cannot start, and the only way
//! forward used to be an agent asking the operator whether the old worktree
//! may be deleted. Auto-fold cannot help: it only folds terminal runs, and an
//! unfinished run keeps its worktrees on purpose so it can be resumed.
//!
//! When the run holding the branch is an earlier attempt at the same task
//! ([`crate::queue::Task::earlier_attempts`]) - superseded or not: blocked,
//! failed, stale or parked alike - and nothing is driving it, its worktree
//! (and only its worktree) is released before the review starts. The
//! branch, every commit and any pull request stay exactly where they were.
//!
//! A branch held by a run of *another* task is released on the same terms when
//! magi can prove the worktree is its own: the path is a candidate worktree
//! one run's record names, laid under that record's own worktree root, and the
//! run is not being driven. Foreign worktrees are never released: [`preflight`]
//! accepts clean holders for isolated review and refuses dirty or locked ones.
//! Live magi holders retain the existing refusal policy.
//!
//! [`decide`] is pure; [`release`] performs owned-worktree handover. A hand-run
//! `magi review` has no task, so it carries a [`Takeover`] with no earlier
//! attempts and can only release a worktree of a run magi itself recorded.

use std::path::{Path, PathBuf};

use anyhow::Result;

use crate::git;
use crate::run::{Liveness, RunState, RunStatus};

/// The earlier attempts at a task a new run may take a branch over from.
#[derive(Debug, Clone)]
pub struct Takeover {
    /// Ids of runs the new attempt supersedes, from
    /// [`crate::queue::Task::earlier_attempts`].
    pub earlier: Vec<String>,
    /// The magi home their records live under.
    pub home: PathBuf,
    /// The owner's answer to an earlier divergence question about the branch,
    /// applied once the earlier worktree is released (git refuses to move a
    /// branch that is checked out).
    pub choice: Option<crate::reconcile::Choice>,
    /// The task's own statement. Used instead of the commit-log stand-in when
    /// the earlier attempt's evidence is inherited, so the completion
    /// criteria survive the retry.
    pub task_instruction: Option<String>,
}

/// The operator has to decide: the takeover was refused, and the text says why.
///
/// A distinct type so the queue loop can hold the task for a person instead
/// of spending an attempt on it (see `daemon::attempt`).
///
/// Structured so the operator-facing hold reason can be rendered in the run's
/// language (`daemon::refused_text`); `Display` is the English text, unchanged.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Refused {
    /// The branch is held by a path magi cannot prove it made.
    Foreign {
        /// The contested branch.
        branch: String,
        /// The worktree that holds it.
        path: String,
        /// Machine detail of why it was refused (English).
        why: String,
    },
    /// A magi worktree holds the branch, and releasing it is not safe.
    Unsafe {
        /// The contested branch.
        branch: String,
        /// The worktree that holds it.
        path: String,
        /// Machine detail of why it was refused (English).
        why: String,
    },
    /// Releasing the earlier run's worktree failed and was undone.
    ReleaseFailed {
        /// The contested branch.
        branch: String,
        /// The worktree that holds it.
        path: String,
        /// Short id of the run that owned the worktree.
        run: String,
    },
}

impl std::fmt::Display for Refused {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Refused::Foreign { branch, path, why } => write!(
                f,
                "branch `{branch}` is checked out in {path}, which magi will not remove by \
                 itself: {why}. Commit or stash changes first. If no longer needed, run: \
                 git worktree remove {} (or renri remove {} -y --non-interactive), then try again.",
                crate::foreign_checkout::quote(path),
                crate::foreign_checkout::quote(branch)
            ),
            Refused::Unsafe { branch, path, why } => write!(
                f,
                "branch `{branch}` is checked out in {path}: {why}. Commit or discard the work \
                 there and remove that worktree (`git worktree remove`), or say the run may be \
                 discarded, and try again."
            ),
            Refused::ReleaseFailed { branch, path, run } => write!(
                f,
                "branch `{branch}` is checked out in {path} by run {run}, and releasing that \
                 worktree failed or found it changed (git refuses to remove a worktree with \
                 uncommitted changes); it was left as it was"
            ),
        }
    }
}

impl std::error::Error for Refused {}

/// A worktree that was released, with what is needed to put it back if the
/// review that took the branch over never starts.
#[derive(Debug, Clone)]
pub struct Released {
    /// The run whose worktree was released.
    pub old_id: String,
    index: usize,
    path: PathBuf,
    home: PathBuf,
    successor: String,
    /// The branch tip when the worktree was released; the branch sync may
    /// have moved it since.
    tip: String,
    /// What was released and why it was safe, for the new run's events.
    pub audit: String,
}

impl Released {
    /// Undo the release after the new run failed to start: check the branch
    /// out again at the old path and unmark the old run. Best-effort - if the
    /// worktree cannot be re-added the old run stays marked, which is true.
    pub async fn restore(&self, repo: &Path, branch: &str) {
        let Ok(_owner) = crate::pr_head::owner_lock(&self.home, &self.old_id) else {
            tracing::warn!("could not lock run {} for handover rollback", self.old_id);
            return;
        };
        let Ok(mut state) = RunState::load_under(&self.old_id, &self.home) else {
            return;
        };
        let claimed =
            crate::daemon::is_working_on_other(&self.home, &self.old_id, jiff::Timestamp::now())
                .unwrap_or(true);
        if state.released_to.as_deref() != Some(self.successor.as_str())
            || state.status == RunStatus::Superseded
            || state.pr_taken_over_by.is_some()
            || state.liveness_under(&self.home, claimed) != Liveness::Dead
        {
            tracing::warn!(
                "run {} changed ownership; handover rollback refused",
                self.old_id
            );
            return;
        }
        let path = self.path.to_string_lossy().to_string();
        // Nothing holds the branch now, so it can be put back where the old
        // run left it before its worktree is recreated at that commit.
        let refname = format!("refs/heads/{branch}");
        if git::rev_parse(repo, &refname).await.ok().as_deref() != Some(self.tip.as_str())
            && let Err(e) = git::git(repo, &["branch", "-f", branch, &self.tip]).await
        {
            tracing::warn!("could not put `{branch}` back at {}: {e:#}", self.tip);
            return;
        }
        if let Err(e) = git::git(repo, &["worktree", "add", &path, branch]).await {
            tracing::warn!(
                "could not put run {}'s worktree back at {path}: {e:#}",
                self.old_id
            );
            return;
        }
        if let Some(c) = state.candidates.get_mut(self.index) {
            c.folded = false;
        }
        state.released_to = None;
        state.released_branches.retain(|b| b != branch);
        state.events.pop();
        let put_back = state.save_under(&self.home);
        if let Err(e) = put_back {
            tracing::warn!("could not unmark run {}: {e:#}", self.old_id);
        }
    }
}

/// What is known about the run and worktree holding the branch.
#[derive(Debug, Clone)]
pub struct Holder {
    /// The run's id.
    pub run: String,
    /// Where the run got to.
    pub status: RunStatus,
    /// Whether anything is driving it.
    pub liveness: Liveness,
    /// Driver ownership could not be shown dead, including legacy records
    /// without a pid: "unknown" must not read as "gone".
    pub driver_unproven: bool,
    /// Any uncommitted change, untracked files included.
    pub dirty: bool,
    /// The first few changed paths, for the refusal text.
    pub dirty_files: Vec<String>,
    /// The recorded driver pid, for the refusal text.
    pub driver_pid: Option<u32>,
    /// HEAD of the worktree.
    pub head: String,
    /// The branch tip in the repository.
    pub tip: String,
}

/// Whose worktree holds the branch, as far as magi can prove.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Owner {
    /// A candidate worktree of an earlier attempt at the same task.
    EarlierAttempt,
    /// A candidate worktree some run's record names, laid under that run's
    /// own worktree root: magi made it.
    MagiRun,
    /// Not provably magi's; the text says why.
    Foreign(String),
}

/// The verdict of [`decide`].
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Decision {
    /// Safe to release the worktree.
    Release,
    /// Not safe to release; the text says exactly why, for the operator.
    Refuse(String),
}

fn short(sha: &str) -> String {
    sha.chars().take(7).collect()
}

/// Whether `holder`'s worktree may be released to a new run.
///
/// [`Liveness::Unknown`] is refused like a live driver: an owner's answer
/// cannot substitute for execution-lease or process-identity evidence. A run
/// in `landing` (possibly waiting on the owner's approval) is refused as well.
pub fn decide(owner: &Owner, holder: &Holder) -> Decision {
    if let Owner::Foreign(why) = owner {
        return Decision::Refuse(why.clone());
    }
    let mut why = Vec::new();
    if holder.liveness == Liveness::Live {
        why.push(format!(
            "run {} is being worked on right now{}",
            crate::run::short_of(&holder.run),
            holder
                .driver_pid
                .map(|p| format!(" (driver pid {p})"))
                .unwrap_or_default()
        ));
    } else if holder.liveness == Liveness::Unknown || holder.driver_unproven {
        why.push("its driver process could not be shown to be gone".to_owned());
    }
    if holder.status == RunStatus::Landing {
        why.push("that run is in `landing`, possibly waiting on an approval".to_owned());
    }
    if holder.dirty {
        let files = holder.dirty_files.join(", ");
        why.push(format!("its worktree has uncommitted changes ({files})"));
    }
    if holder.head != holder.tip {
        why.push("its HEAD is not at the branch tip".to_owned());
    }
    if why.is_empty() {
        return Decision::Release;
    }
    let kind = match owner {
        Owner::EarlierAttempt => "an earlier attempt at this task",
        _ => "a magi run of another task",
    };
    Decision::Refuse(format!(
        "run {} (status `{}`, worktree {}, HEAD {}, branch tip {}) is {kind} and still has the \
         branch checked out, so it was not released automatically: {}",
        crate::run::short_of(&holder.run),
        holder.status.as_str(),
        if holder.dirty { "dirty" } else { "clean" },
        short(&holder.head),
        short(&holder.tip),
        why.join("; ")
    ))
}

/// Read what [`decide`] needs from disk and git.
async fn inspect(
    repo: &Path,
    branch: &str,
    path: &Path,
    state: &RunState,
    home: &Path,
) -> Result<Holder> {
    let claimed = crate::daemon::is_working_on_other(home, &state.id, jiff::Timestamp::now())?;
    let liveness = state.liveness_under(home, claimed);
    // Spelled out: `status.showUntrackedFiles=no` in a user's config would
    // otherwise hide new files from `git status` and let them be deleted.
    let porcelain = git::git(path, &["status", "--porcelain", "--untracked-files=normal"]).await?;
    Ok(Holder {
        run: state.id.clone(),
        status: state.status,
        liveness,
        driver_unproven: liveness == Liveness::Unknown,
        dirty: !porcelain.trim().is_empty(),
        dirty_files: porcelain
            .lines()
            .take(5)
            .map(|l| l.get(3..).unwrap_or(l).trim().to_owned())
            .collect(),
        driver_pid: state.driver_pid,
        head: git::rev_parse(path, "HEAD").await?,
        tip: git::rev_parse(repo, &format!("refs/heads/{branch}")).await?,
    })
}

fn same_path(a: &Path, b: &Path) -> bool {
    let canon = |p: &Path| std::fs::canonicalize(p).unwrap_or_else(|_| p.to_path_buf());
    canon(a) == canon(b)
}

fn foreign(branch: &str, path: &Path, why: &str) -> anyhow::Error {
    Refused::Foreign {
        branch: branch.to_owned(),
        path: path.display().to_string(),
        why: why.to_owned(),
    }
    .into()
}

/// Check foreign holders before filing a review and again when it starts.
/// Recorded magi worktrees retain the existing handover policy.
pub async fn preflight(
    repo: &Path,
    branch: &str,
    home: &Path,
) -> Result<Option<crate::foreign_checkout::ForeignCheckout>> {
    let Some(path) = git::worktree_holding(repo, branch).await? else {
        return Ok(None);
    };
    match find_magi_owner(&path, home) {
        Ok(Some(_)) => return Ok(None),
        Err(why) => return Err(foreign(branch, &path, &why)),
        Ok(None) => {}
    }
    crate::foreign_checkout::inspect(repo, branch, &path)
        .await
        .map(Some)
        .map_err(|error| foreign(branch, &path, &format!("{error:#}")))
}

/// Preflight for a review request, including foreign-branch remote divergence.
/// This runs before queue writes; execution repeats it to catch later changes.
pub async fn preflight_review(
    repo: &Path,
    branch: &str,
    home: &Path,
    remote: &str,
) -> Result<Option<crate::foreign_checkout::ForeignCheckout>> {
    let checkout = preflight(repo, branch, home).await?;
    if let Some(checkout) = &checkout {
        let fetched = git::fetch(repo, remote, branch).await;
        if matches!(fetched, Ok(ref out) if out.ok()) {
            let tracking = format!("{remote}/{branch}");
            let tip = git::rev_parse(repo, &tracking).await?;
            if !git::is_ancestor(repo, &checkout.synced_head, &tip).await
                && !git::is_ancestor(repo, &tip, &checkout.synced_head).await
            {
                return Err(foreign(
                    branch,
                    &checkout.path,
                    &format!("branch diverged from {tracking}; reconcile it before review"),
                ));
            }
        }
    }
    Ok(checkout)
}

/// The run and candidate whose recorded worktree is `path`, when magi
/// provably made it: the record names the path, the path sits directly in the
/// record's own worktree root (`<root>/<short>/<dir>`), and exactly one run
/// says so. `Err` is the reason a path under a root could not be proven.
fn find_magi_owner(
    path: &Path,
    home: &Path,
) -> std::result::Result<Option<(RunState, usize)>, String> {
    let Some(bay) = path.parent() else {
        return Ok(None);
    };
    let Some(bay_name) = bay.file_name().and_then(|n| n.to_str()) else {
        return Ok(None);
    };
    let mut found = Vec::new();
    for id in crate::run::list_ids_in(&home.join("runs")) {
        if crate::run::short_of(&id) != bay_name {
            continue;
        }
        let Ok(state) = RunState::load_under(&id, home) else {
            return Err(format!("run {bay_name}'s record could not be read"));
        };
        if !same_path(&state.worktree_root(), bay) {
            continue;
        }
        if let Some(i) = state
            .candidates
            .iter()
            .position(|c| !c.folded && same_path(&c.worktree, path))
        {
            found.push((state, i));
        }
    }
    match found.len() {
        0 => Ok(None),
        1 => Ok(found.pop()),
        _ => Err(format!("more than one run record claims it ({bay_name})")),
    }
}

/// A structured PR record survives both stop messages replacing `merge.detail`
/// and older resumes accidentally resetting the graph status to Gating.
pub(crate) fn landing_checkpoint(state: &RunState) -> bool {
    state.status.resumable()
        && state.config.graph.land
        && state.pr.as_ref().is_some_and(|pr| pr.state == "open")
        && state
            .merge
            .as_ref()
            .is_some_and(|m| m.mode == crate::config::MergeMode::Pr && !m.empty)
}

/// Why a stopped execution cannot be resumed, in admission-check order.
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) enum ResumeRefusal {
    Status(RunStatus),
    Released(String),
    Retired(String),
    OwnershipTransferred(String),
    ReviewBudget { used: usize, limit: usize },
    ReviewDisabled,
    CleanReviewGateRan,
    BaseSyncStop,
}

impl ResumeRefusal {
    /// The recorded new owner, when admission was refused by an ownership move.
    pub(crate) fn successor(&self) -> Option<&str> {
        match self {
            Self::Released(to) | Self::Retired(to) | Self::OwnershipTransferred(to) => Some(to),
            _ => None,
        }
    }

    pub(crate) fn diagnostic(&self, state: &RunState) -> String {
        let reason = match self {
            Self::BaseSyncStop => "base synchronization is stopped; inspect the retained work. For movement exhaustion only, use `magi run --resume <run> --recover-base` to authorize a new bounded synchronization interval and fresh verification".into(),
            Self::Status(status) => format!("status `{}` is final; there is nothing left to resume", status.as_str()),
            Self::Released(_) => "execution was released".into(),
            Self::Retired(_) => "execution was retired".into(),
            Self::OwnershipTransferred(_) => "PR ownership was transferred".into(),
            Self::ReviewBudget { used, limit } => format!("review budget exhausted ({used}/{limit} rounds used)"),
            Self::CleanReviewGateRan => "review already has a clean round and the gate already ran; no retryable checkpoint remains".into(),
            Self::ReviewDisabled => "review is disabled (review_rounds = 0) and no retryable checkpoint or pending gate remains".into(),
        };
        let mut detail = format!("run {} cannot resume: {reason}.", state.short());
        if let Some(target) = self.successor() {
            let target = if matches!(self, Self::Retired(_)) {
                target
            } else {
                crate::run::short_of(target)
            };
            detail.push_str(&format!(" Successor: run {target}; inspect `magi show {target}` before choosing any further action."));
        }
        for (label, target) in [
            ("released to", &state.released_to),
            ("retired to", &state.retired_to),
            ("PR ownership moved to", &state.pr_taken_over_by),
        ] {
            if let Some(target) = target {
                if self.successor() == Some(target.as_str()) {
                    continue;
                }
                let target = if label == "retired to" {
                    target.as_str()
                } else {
                    crate::run::short_of(target)
                };
                detail.push_str(&format!(" {label} run {target}; inspect `magi show {target}` before choosing any further action."));
            }
        }
        detail.push_str(&format!(
            " Inspect `magi show {}` for retained work and ownership.",
            state.id
        ));
        if matches!(self, Self::ReviewBudget { .. } | Self::ReviewDisabled) {
            detail.push_str(" To continue the retained implementation, first confirm its saved candidate branch and owning task with `magi task show <task-id>`. When the task is eligible and no driver is live, explicitly use `magi review <saved-branch> --task <task-id> --merge none --review-rounds <positive-rounds>`. This creates a review-only successor with a new review budget; it does not resume this run or start a new implementation competition. Existing ownership and checkpoint checks still apply.");
        }
        detail
    }
}

/// Shared admission policy for queue retries, typed actions and control resume.
/// Review exhaustion does not spend publication or landing checkpoints, or
/// prevent pending verification or gate work from retrying infrastructure waits.
pub(crate) fn can_resume(state: &RunState) -> bool {
    resume_refusal(state).is_none()
}

pub(crate) fn resume_refusal(state: &RunState) -> Option<ResumeRefusal> {
    resume_refusal_under(state, crate::run::try_home().as_deref())
}

/// Explicit stores keep validation and question checkpoints in the same home.
pub(crate) fn resume_refusal_under(state: &RunState, home: Option<&Path>) -> Option<ResumeRefusal> {
    if !state.status.resumable() {
        return Some(ResumeRefusal::Status(state.status));
    }
    if let Some(to) = &state.released_to {
        return Some(ResumeRefusal::Released(to.clone()));
    }
    if let Some(to) = &state.retired_to {
        return Some(ResumeRefusal::Retired(to.clone()));
    }
    if let Some(to) = &state.pr_taken_over_by {
        return Some(ResumeRefusal::OwnershipTransferred(to.clone()));
    }
    if state.base_sync_blocked() {
        return if state.base_recovery_pending() || state.base_sync_closure_pending() {
            None
        } else {
            Some(ResumeRefusal::BaseSyncStop)
        };
    }
    let review = crate::graph::review_conclusion(&state.reviews, state.config.graph.review_rounds);
    let allowed = resume_checkpoint(state)
        || state
            .repair_grants
            .last()
            .is_some_and(|g| g.phase != crate::repair_budget::Phase::Complete)
        || home.is_some_and(|home| {
            state
                .origin
                .as_ref()
                .and_then(|o| o.task.as_ref())
                .and_then(|id| crate::queue::Queue::at(home.join("queue")).get(id).ok())
                .is_some_and(|task| crate::repair_budget::pending(&task, state))
        })
        || home.is_some_and(|home| has_resume_checkpoint(state, home))
        || ((state.config.graph.review_rounds == 0
            || state.reviews.iter().any(|round| round.clean))
            && (!state.gate_ran || state.resource_block.is_some()))
        || (state.resource_block.is_some() && review == Some(RunStatus::Gating))
        || (review.is_none()
            && state
                .reviews
                .last()
                .is_some_and(|round| round.e2e_status() == crate::run::E2eStatus::ResourceBlocked))
        || !(state.status == RunStatus::Blocked
            && state.reviews.len() >= state.config.graph.review_rounds);
    if allowed {
        None
    } else if state.config.graph.review_rounds == 0 {
        Some(ResumeRefusal::ReviewDisabled)
    } else if state.reviews.iter().any(|round| round.clean) {
        Some(ResumeRefusal::CleanReviewGateRan)
    } else {
        Some(ResumeRefusal::ReviewBudget {
            used: state.reviews.len(),
            limit: state.config.graph.review_rounds,
        })
    }
}

fn resume_checkpoint(state: &RunState) -> bool {
    state.status == RunStatus::Landing
        || landing_checkpoint(state)
        || state.pr_publication_pending
        || state.land_approval.is_some()
        || state.merge_decision.is_some()
        || state.land_armed_head.is_some()
        || !state.merge_repairs.is_empty()
}

/// A resumable publication, landing or owner decision must keep its existing
/// graph and worktree. Call under the owner lock before releasing or retiring
/// a predecessor; completed runs may retain these fields as historical evidence.
pub(crate) fn has_resume_checkpoint(state: &RunState, home: &Path) -> bool {
    state.status.resumable()
        && (resume_checkpoint(state)
            || crate::ask::Questions::at(home.join("questions"))
                .list()
                .iter()
                .any(|q| q.run == state.id && q.status.open()))
}

/// The scan only finds a possible holder. Driver claims share this lock, so
/// reload and validate the candidate while holding it through removal/rollback.
fn locked_holder(
    home: &Path,
    id: &str,
    path: &Path,
) -> Result<(RunState, usize, crate::pr_head::OwnerLock)> {
    let owner = crate::pr_head::owner_lock(home, id)?;
    let state = RunState::load_under(id, home)?;
    let index = state
        .candidates
        .iter()
        .position(|c| !c.folded && same_path(&c.worktree, path))
        .ok_or_else(|| anyhow::anyhow!("worktree holder changed; retry takeover"))?;
    anyhow::ensure!(
        state.released_to.is_none(),
        "run already handed over its worktree; retry takeover"
    );
    anyhow::ensure!(
        path.parent()
            .is_some_and(|p| same_path(&state.worktree_root(), p)),
        "worktree holder root changed; retry takeover"
    );
    Ok((state, index, owner))
}

/// Release the worktree holding `branch` to `new_run`, when an earlier attempt
/// at the same task, or a run magi itself recorded, holds it and it is safe. Returns the run whose worktree
/// was released (see [`Released`]), or `None` when nothing needed (or was allowed) to happen.
///
/// An `Err` is a refusal or a failed removal; the message is what the
/// operator reads. Nothing is left half-done: the old run's record is written
/// before the removal and rolled back if git refuses it.
pub async fn release(
    repo: &Path,
    branch: &str,
    new_run: &str,
    takeover: &Takeover,
) -> Result<Option<Released>> {
    let Some(path) = git::worktree_holding(repo, branch).await? else {
        return Ok(None);
    };
    // The holder has to be an unfolded candidate worktree of a run on record.
    // One nobody recorded (made by hand, say) is not ours to touch.
    let mut owner = None;
    for id in &takeover.earlier {
        let Ok(state) = RunState::load_under(id, &takeover.home) else {
            continue;
        };
        if let Some(i) = state
            .candidates
            .iter()
            .position(|c| !c.folded && same_path(&c.worktree, &path))
        {
            // Naming the path is not enough: magi must have laid it too.
            if !path
                .parent()
                .is_some_and(|b| same_path(&state.worktree_root(), b))
            {
                return Err(foreign(
                    branch,
                    &path,
                    &format!(
                        "run {} records it, but it is outside that run's worktree root, so \
                         magi did not make it",
                        crate::run::short_of(id)
                    ),
                ));
            }
            owner = Some((state, i, Owner::EarlierAttempt));
            break;
        }
    }
    if owner.is_none() {
        owner = match find_magi_owner(&path, &takeover.home) {
            Ok(found) => found.map(|(s, i)| (s, i, Owner::MagiRun)),
            Err(why) => return Err(foreign(branch, &path, &why)),
        };
    }
    let Some((state, _, kind)) = owner else {
        preflight(repo, branch, &takeover.home).await?;
        return Ok(None);
    };

    let (mut state, index, _owner) = locked_holder(&takeover.home, &state.id, &path)?;
    if has_resume_checkpoint(&state, &takeover.home) {
        return Err(Refused::Unsafe {
            branch: branch.to_owned(),
            path: path.display().to_string(),
            why: format!("run {} has a resumable publication, landing or decision checkpoint; resume that run instead of releasing its worktree", state.short()),
        }.into());
    }
    let holder = inspect(repo, branch, &path, &state, &takeover.home).await?;
    if let Decision::Refuse(why) = decide(&kind, &holder) {
        return Err(Refused::Unsafe {
            branch: branch.to_owned(),
            path: path.display().to_string(),
            why,
        }
        .into());
    }
    let audit = format!(
        "run {} (status `{}`, no driver, clean, HEAD {} = branch tip, branch `{branch}` kept)",
        crate::run::short_of(&holder.run),
        holder.status.as_str(),
        short(&holder.head)
    );

    let old_id = state.id.clone();
    state.candidates[index].folded = true;
    state.released_to = Some(new_run.to_owned());
    if !state.released_branches.iter().any(|b| b == branch) {
        state.released_branches.push(branch.to_owned());
    }
    state.event(
        "release",
        format!(
            "worktree of `{branch}` released to run {}: {audit}; this run can no longer be \
             resumed from here",
            crate::run::short_of(new_run)
        ),
    );
    state.save_under(&takeover.home)?;

    // Re-read right before the removal for the driver and the tip, and let git
    // itself refuse a dirty worktree: the removal is *not* `--force`, so an
    // edit made after the first look is refused rather than thrown away.
    // Ignored files such as `target/` go with the worktree by design.
    // Against the record as it is on disk now, not the copy read earlier: a
    // driver cannot claim between this check and removal: _owner remains
    // held through the save, removal and any rollback.
    let again = match RunState::load_under(&old_id, &takeover.home) {
        Ok(fresh) => inspect(repo, branch, &path, &fresh, &takeover.home).await,
        Err(e) => Err(e),
    };
    let safe = matches!(&again, Ok(h) if decide(&kind, h) == Decision::Release);
    let removed = safe
        && git::worktree_remove_clean(repo, &path)
            .await
            .unwrap_or(false);
    if !removed || path.exists() {
        state = RunState::load_under(&old_id, &takeover.home)?;
        state.candidates[index].folded = false;
        state.released_to = None;
        state.released_branches.retain(|b| b != branch);
        state.events.pop();
        state.event(
            "release",
            format!(
                "release of `{branch}` to run {} was undone: {audit}",
                crate::run::short_of(new_run)
            ),
        );
        state.save_under(&takeover.home)?;
        return Err(Refused::ReleaseFailed {
            branch: branch.to_owned(),
            path: path.display().to_string(),
            run: crate::run::short_of(&old_id).to_owned(),
        }
        .into());
    }
    Ok(Some(Released {
        old_id,
        index,
        path,
        home: takeover.home.clone(),
        successor: new_run.to_owned(),
        tip: holder.tip,
        audit,
    }))
}

#[cfg(test)]
mod tests {
    #[test]
    fn refused_display_is_the_english_text() {
        let r = Refused::Foreign {
            branch: "b".into(),
            path: "/w".into(),
            why: "w".into(),
        };
        assert_eq!(
            r.to_string(),
            "branch `b` is checked out in /w, which magi will not remove by itself: w. Commit or stash changes first. If no longer needed, run: git worktree remove '/w' (or renri remove 'b' -y --non-interactive), then try again."
        );
    }

    use super::*;

    fn holder() -> Holder {
        Holder {
            run: "20260901-000000-f82f".to_owned(),
            status: RunStatus::Gating,
            liveness: Liveness::Dead,
            driver_unproven: false,
            dirty: false,
            dirty_files: Vec::new(),
            driver_pid: None,
            head: "a".repeat(40),
            tip: "a".repeat(40),
        }
    }

    fn recorded_holder(home: &Path) -> (RunState, PathBuf) {
        let mut state = RunState::new(
            "repo".into(),
            "main".into(),
            "base".into(),
            "task".into(),
            crate::config::Config::default(),
        );
        state.config.graph.worktree_root = Some(home.join("worktrees"));
        let path = state.worktree_root().join("cand-A");
        std::fs::create_dir_all(&path).unwrap();
        state.driver_exited = true;
        state.candidates.push(crate::run::Candidate {
            index: 0,
            label: 'A',
            agent: "existing".into(),
            branch: "branch".into(),
            worktree: path.clone(),
            summary: String::new(),
            stat: String::new(),
            files: 1,
            commits: 1,
            empty: false,
            failed: None,
            verified_noop: None,
            duration_ms: 0,
            folded: false,
        });
        state.save_under(home).unwrap();
        (state, path)
    }

    #[test]
    fn handover_reloads_a_driver_claim_after_the_scan_and_excludes_new_claims() {
        let home = tempfile::tempdir().unwrap();
        let (mut state, path) = recorded_holder(home.path());
        let scanned = state.clone();
        // A resume won the race after the holder scan, before the owner lock.
        state.driver_exited = false;
        state.driver_pid = Some(std::process::id());
        state.driver_started_at = crate::proc::process_started_at(std::process::id());
        state.begin_pass();
        state.save_under(home.path()).unwrap();
        let mut driver = crate::driver::DriverGuard::acquire(home.path(), &state).unwrap();
        driver.record(&state).unwrap();
        let (fresh, _, owner) = locked_holder(home.path(), &scanned.id, &path).unwrap();
        assert!(!fresh.driver_exited);
        assert_eq!(fresh.passes, state.passes);
        assert_ne!(fresh.liveness_under(home.path(), false), Liveness::Dead);
        assert!(crate::pr_head::owner_lock(home.path(), &state.id).is_err());
        assert!(path.exists());
        assert!(
            RunState::load_under(&state.id, home.path())
                .unwrap()
                .released_to
                .is_none()
        );
        drop(owner);
        assert!(crate::pr_head::owner_lock(home.path(), &state.id).is_ok());
    }

    #[test]
    fn handover_revalidates_candidate_ownership_under_the_claim_lock() {
        let home = tempfile::tempdir().unwrap();
        let (mut state, path) = recorded_holder(home.path());
        state.candidates[0].folded = true;
        state.released_to = Some("next".into());
        state.save_under(home.path()).unwrap();
        assert!(locked_holder(home.path(), &state.id, &path).is_err());
        assert!(path.exists());
    }

    #[tokio::test]
    async fn a_stale_restore_does_not_undo_a_later_handover() {
        let home = tempfile::tempdir().unwrap();
        let (mut state, path) = recorded_holder(home.path());
        state.candidates[0].folded = true;
        state.released_to = Some("later".into());
        state.save_under(home.path()).unwrap();
        let released = Released {
            old_id: state.id.clone(),
            index: 0,
            path,
            home: home.path().to_owned(),
            successor: "original".into(),
            tip: "tip".into(),
            audit: String::new(),
        };
        released.restore(Path::new("nonexistent"), "branch").await;
        let disk = RunState::load_under(&state.id, home.path()).unwrap();
        assert_eq!(disk.released_to.as_deref(), Some("later"));
        assert!(disk.candidates[0].folded);
    }

    #[test]
    fn a_clean_superseded_stale_run_is_released() {
        assert_eq!(decide(&Owner::EarlierAttempt, &holder()), Decision::Release);
        let dead = Holder {
            liveness: Liveness::Dead,
            ..holder()
        };
        assert_eq!(decide(&Owner::EarlierAttempt, &dead), Decision::Release);
    }

    #[test]
    fn a_foreign_worktree_is_refused_with_its_reason() {
        let owner = Owner::Foreign("it is a foreign path".to_owned());
        assert_eq!(
            decide(&owner, &holder()),
            Decision::Refuse("it is a foreign path".to_owned())
        );
    }

    #[test]
    fn landing_is_protected_even_for_an_earlier_attempt() {
        let landing = Holder {
            status: RunStatus::Landing,
            ..holder()
        };
        assert!(matches!(
            decide(&Owner::MagiRun, &landing),
            Decision::Refuse(_)
        ));
        assert!(matches!(
            decide(&Owner::EarlierAttempt, &landing),
            Decision::Refuse(_)
        ));
    }

    #[test]
    fn a_dirty_worktree_is_refused_and_says_why() {
        let dirty = Holder {
            dirty: true,
            dirty_files: vec!["scratch.txt".to_owned()],
            ..holder()
        };
        let Decision::Refuse(why) = decide(&Owner::EarlierAttempt, &dirty) else {
            panic!("dirty must be refused");
        };
        assert!(
            why.contains("uncommitted") && why.contains("scratch.txt"),
            "{why}"
        );
        assert!(why.contains("f82f") && why.contains("gating") && why.contains("dirty"));
    }

    #[test]
    fn a_live_run_is_refused() {
        let live = Holder {
            liveness: Liveness::Live,
            driver_pid: Some(4242),
            ..holder()
        };
        let Decision::Refuse(why) = decide(&Owner::EarlierAttempt, &live) else {
            panic!("live must be refused");
        };
        assert!(
            why.contains("right now") && why.contains("f82f") && why.contains("4242"),
            "{why}"
        );
    }

    #[test]
    fn unknown_without_a_recorded_pid_is_not_proof_of_a_stopped_driver() {
        let holder = Holder {
            liveness: Liveness::Unknown,
            ..holder()
        };
        assert!(matches!(
            decide(&Owner::EarlierAttempt, &holder),
            Decision::Refuse(_)
        ));
    }

    #[test]
    fn a_driver_that_could_not_be_shown_dead_is_refused() {
        let unproven = Holder {
            driver_unproven: true,
            ..holder()
        };
        let Decision::Refuse(why) = decide(&Owner::EarlierAttempt, &unproven) else {
            panic!("an unproven driver must be refused");
        };
        assert!(why.contains("driver"), "{why}");
    }

    #[test]
    fn a_head_off_the_tip_is_refused() {
        let off = Holder {
            head: "b".repeat(40),
            ..holder()
        };
        assert!(matches!(
            decide(&Owner::EarlierAttempt, &off),
            Decision::Refuse(_)
        ));
    }
}