artifactize 0.10.0

//! Run lifecycle, scheduling, cancellation, and budget counters.

mod execution;

use std::{
    collections::{BTreeMap, BTreeSet},
    sync::Arc,
    time::Duration,
};

use tokio::task::JoinSet;
use tokio_util::sync::CancellationToken;

use crate::{
    agent::error::Code,
    cache,
    config::{ProfileKind, RepoConfig},
    graph::{EvalEvaluation, EvalStatus, Evidence, Graph},
    limits::Limits,
    process,
    remote::Session,
    runtime::Verdict,
    store::{
        Capacity, Claim, Execution, Producer, Provenance, Receipts, Request, Run, StoppedBackend,
    },
    tools,
};

/// OS owner death and released machine-wide slots have no reliable writer notification.
/// Bound their reconciliation independently of immediate commit hints and task completion.
const OWNER_RECONCILE: Duration = Duration::from_secs(1);

/// Once an absolute Human deadline passed, outstanding work still drains but cannot re-arm
/// an immediately-ready timer. Its next wake is task completion, cancellation or reconciliation.
fn future_deadline(
    deadline: Option<tokio::time::Instant>,
    human_wait: bool,
    now: tokio::time::Instant,
) -> Option<tokio::time::Instant> {
    deadline.filter(|deadline| human_wait && *deadline > now)
}

#[cfg(test)]
mod wake_tests {
    use super::*;
    /// A timestamp's saved text, or `None` when it is not a sortable time.
    fn sortable(value: &str) -> Option<String> {
        value
            .parse::<crate::types::Timestamp>()
            .ok()
            .map(|time| time.to_string())
    }
    #[test]
    fn sortable_rfc3339_year_bounds_keep_fixed_width_utc_text() {
        assert_eq!(
            sortable("0000-01-01T00:00:00Z").as_deref(),
            Some("0000-01-01T00:00:00.000000000Z")
        );
        assert_eq!(
            sortable("9999-12-31T23:59:59Z").as_deref(),
            Some("9999-12-31T23:59:59.000000000Z")
        );
        assert_eq!(sortable("0000-01-01T00:00:00+01:00"), None);
        assert_eq!(sortable("10000-01-01T00:00:00Z"), None);
    }
    #[test]
    fn sortable_rejects_utc_offset_overflow_without_panicking() {
        assert_eq!(sortable("9999-12-31T23:59:59-01:00"), None);
        assert_eq!(sortable("0000-01-01T00:00:00+01:00"), None);
        assert_eq!(sortable("9999-12-31T23:59:59+24:00"), None);
        assert_eq!(sortable("0000-01-01T00:00:00-24:00"), None);
        assert_eq!(
            sortable("9999-12-31T22:59:59-01:00").as_deref(),
            Some("9999-12-31T23:59:59.000000000Z")
        );
        assert_eq!(
            sortable("0000-01-01T01:00:00+01:00").as_deref(),
            Some("0000-01-01T00:00:00.000000000Z")
        );
    }
    #[test]
    fn future_deadline_uses_the_supplied_instant_and_only_waiting_humans() {
        let now = tokio::time::Instant::now();
        let future = now + Duration::from_secs(1);
        assert_eq!(future_deadline(Some(future), true, now), Some(future));
        assert_eq!(future_deadline(Some(now), true, now), None);
        assert_eq!(
            future_deadline(Some(now - Duration::from_secs(1)), true, now),
            None
        );
        assert_eq!(future_deadline(Some(future), false, now), None);
        assert_eq!(future_deadline(None, true, now), None);
    }

    #[tokio::test(start_paused = true)]
    async fn expired_human_deadline_does_not_spin_while_other_work_drains() {
        let deadline = tokio::time::Instant::now() + Duration::from_millis(10);
        assert_eq!(
            future_deadline(Some(deadline), true, tokio::time::Instant::now()),
            Some(deadline)
        );
        tokio::time::advance(Duration::from_millis(11)).await;
        let mut reconcile = tokio::time::interval(OWNER_RECONCILE);
        reconcile.tick().await;
        let finished = tokio::time::Instant::now() + Duration::from_secs(3);
        let mut iterations = 0;
        loop {
            iterations += 1;
            let wake = future_deadline(Some(deadline), true, tokio::time::Instant::now());
            tokio::select! {
                biased;
                _ = tokio::time::sleep_until(finished) => break,
                _ = tokio::time::sleep_until(wake.unwrap_or_else(tokio::time::Instant::now)),
                    if wake.is_some() => panic!("expired deadline was re-armed"),
                _ = reconcile.tick() => {},
            }
        }
        assert!(
            iterations <= 4,
            "deadline causes a hot scheduler loop: {iterations}"
        );
    }
}

pub(crate) fn now() -> crate::types::Timestamp {
    crate::types::Timestamp::now()
}

/// The backends a Run stopped admitting reviews on, shared with its review tasks. A review
/// records its stop before its execution completes and frees its capacity slot, and taking a
/// slot asks again under the same write lock, so no other review of the backend can take the
/// slot and start.
#[derive(Clone)]
struct Stops(Arc<std::sync::Mutex<Vec<StoppedBackend>>>);

impl Stops {
    fn new(stopped: Vec<StoppedBackend>) -> Self {
        Self(Arc::new(std::sync::Mutex::new(stopped)))
    }

    fn all(&self) -> Vec<StoppedBackend> {
        self.0.lock().unwrap().clone()
    }

    /// The stop of the backend the request's Agent review uses, if any.
    fn of(&self, request: &Request) -> Option<StoppedBackend> {
        let backend = request.options.backend?;
        let stopped = self.0.lock().unwrap();
        stopped.iter().find(|stop| stop.backend == backend).cloned()
    }

    /// Whether `backend` stopped, asked later.
    fn check(&self, backend: &crate::config::Backend) -> Box<dyn FnOnce() -> bool + Send> {
        let (stops, backend) = (self.clone(), *backend);
        Box::new(move || {
            let stopped = stops.0.lock().unwrap();
            stopped.iter().any(|stop| stop.backend == backend)
        })
    }

    /// Stop admitting a backend after a failure every later review on it would repeat.
    fn record(&self, request: &Request) {
        let (Some(backend), Some(code)) = (&request.options.backend, &request.error_code()) else {
            return;
        };
        let mut stopped = self.0.lock().unwrap();
        if Code::stops_backend(code) && !stopped.iter().any(|stop| &stop.backend == backend) {
            stopped.push(StoppedBackend {
                backend: *backend,
                error_code: (*code).try_into().expect("backend stopping failure"),
                eval_id: request.eval_id.clone(),
                request_id: request.id.clone(),
                error: request.error().map(str::to_owned).unwrap_or_default(),
            });
        }
    }
}

/// Why an eval of a `--reuse-only` kind was not executed.
fn not_reused_reason(kind: ProfileKind) -> String {
    format!(
        "Not reused (--reuse-only {}): no cached or stored result, so it was not executed.",
        kind.name()
    )
}

pub(crate) fn budget_reason(run: &Run) -> String {
    format!(
        "maxExecutions budget exhausted ({} of {} executions started).",
        run.executions_started,
        run.max_executions.expect("limited Run")
    )
}

#[expect(
    clippy::too_many_arguments,
    reason = "the Run's inputs, its local and remote stores, and cancellation"
)]
pub(crate) async fn schedule(
    config: Arc<RepoConfig>,
    graph: &Graph<'_>,
    fingerprints: &BTreeMap<&crate::types::ArtifactName, cache::PreparedFingerprint>,
    keys: &BTreeMap<&crate::types::EvalId, cache::Key>,
    parallelism: &cache::Parallelism,
    limits: &Limits,
    run: &mut Run,
    requests: &mut [Request],
    receipts: &Receipts,
    remote: Option<Arc<Session>>,
    cancellation: CancellationToken,
) -> Result<BTreeMap<crate::config::EvalId, Evidence>, String> {
    let cancellation = cancellation.child_token();
    let _cancel_on_drop = cancellation.clone().drop_guard();
    let mut tasks = JoinSet::new();
    let result = Scheduler {
        config,
        graph,
        fingerprints,
        keys,
        parallelism,
        limits,
        run,
        requests,
        receipts,
        remote,
        cancellation: cancellation.clone(),
        tasks: &mut tasks,
    }
    .run()
    .await;
    if result.is_err() {
        cancellation.cancel();
        while let Some(result) = tasks.join_next().await {
            let _ = crate::task::joined(result);
        }
    }
    result
}

struct Scheduler<'a, 'g> {
    config: Arc<RepoConfig>,
    graph: &'a Graph<'g>,
    fingerprints: &'a BTreeMap<&'g crate::types::ArtifactName, cache::PreparedFingerprint>,
    keys: &'a BTreeMap<&'g crate::types::EvalId, cache::Key>,
    parallelism: &'a cache::Parallelism,
    limits: &'a Limits,
    run: &'a mut Run,
    requests: &'a mut [Request],
    receipts: &'a Receipts,
    remote: Option<Arc<Session>>,
    cancellation: CancellationToken,
    tasks: &'a mut JoinSet<Result<(usize, Request), String>>,
}

/// A dependency request is audit-only: no execution, claim, capacity or cache evidence.
/// A GREEN or BLOCKED request completes at `at`.
pub(crate) fn derive(
    request: &mut Request,
    eval: &EvalEvaluation<'_>,
    at: crate::types::Timestamp,
) {
    request.blocked_by = eval.blocked_by.iter().copied().map(Into::into).collect();
    request.queue = None;
    request.blocked_reason = (!request.blocked_by.is_empty()).then(|| {
        format!(
            "Derived dependency verdict: waiting for current GREEN evidence from {}.",
            request
                .blocked_by
                .iter()
                .map(ToString::to_string)
                .collect::<Vec<_>>()
                .join(", ")
        )
    });
    request.state = if eval.status == EvalStatus::Green {
        crate::store::RequestState::completed(
            crate::agent::verdict::ValidatedResult {
                verdict: crate::runtime::Verdict::Green,
                fields: serde_json::Map::from_iter([(
                    "derived".into(),
                    serde_json::Value::Bool(true),
                )]),
            }
            .into(),
            at,
        )
    } else {
        crate::store::RequestState::pending(
            crate::project::verify::status(eval.status),
            (eval.status == EvalStatus::Blocked).then_some(at),
        )
        .expect("derived pending state")
    };
}

impl Scheduler<'_, '_> {
    async fn run(&mut self) -> Result<BTreeMap<crate::config::EvalId, Evidence>, String> {
        // Registration precedes any baseline/scheduling read; racing commits stay dirty.
        let mut changes = crate::changes::Subscription::new(&self.run.state_dir).await;
        let mut reconcile = tokio::time::interval(OWNER_RECONCILE);
        reconcile.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
        reconcile.tick().await;
        let mut remote_due = tokio::time::Instant::now();
        let owner = process::child_identity(std::process::id()).map_err(|e| e.to_string())?;
        let producer = Producer::current();
        // Agent conversations are saved under the state, named by its id and this producer.
        let saving = if self.limits.agent_sessions().enabled {
            Some(crate::agent::session::Saving {
                state: self.run.state_dir.clone(),
                state_id: self.receipts.state_id().await?,
                producer: producer.name.clone(),
            })
        } else {
            None
        };
        let run_dir = self.run.state_dir.join("runs").join(self.run.id.as_str());
        let mut evidence = BTreeMap::new();
        // Saved evals this Run has no request for: their evidence is recorded on the Run.
        let saved_evals: BTreeSet<crate::types::EvalId> = self
            .run
            .definitions
            .graph()
            .map(|graph| graph.evals().iter().map(|eval| eval.id.clone()).collect())
            .unwrap_or_default();
        let mut running = BTreeSet::new();
        let mut waiting = BTreeSet::new();
        // Requests waiting for a machine-wide backend slot: no job slot, no executor start.
        let mut capacity_waiting = BTreeSet::new();
        let stops = Stops::new(self.run.stopped_backends.clone());
        let deadline = self
            .run
            .wait_timeout_ms
            .map(|timeout| tokio::time::Instant::now() + timeout);
        loop {
            let completed = evidence.len();
            // Optional team-store lookups retain their separate due cadence; local session
            // events and cached redraws never schedule remote work. Forced Runs read none.
            if let Some(remote) = &self.remote
                && tokio::time::Instant::now() >= remote_due
                && !self.cancellation.is_cancelled()
                && !self.run.force
            {
                let keys = self
                    .requests
                    .iter()
                    .filter(|request| {
                        request.status() == crate::types::RequestStatus::WaitingHuman
                            && !request.force
                    })
                    .filter_map(|request| request.key.clone())
                    .collect();
                remote.refresh(self.receipts, keys).await?;
                remote_due = tokio::time::Instant::now() + crate::remote::REFRESH_INTERVAL;
            }
            let human_ids: Vec<_> = self
                .requests
                .iter()
                .filter(|request| request.status() == crate::types::RequestStatus::WaitingHuman)
                .map(|request| request.id.clone())
                .collect();
            if !human_ids.is_empty() {
                for request in self.receipts.settled_human_requests(human_ids).await? {
                    if let Some(saved) = self.requests.iter_mut().find(|saved| {
                        saved.id == request.id
                            && saved.status() == crate::types::RequestStatus::WaitingHuman
                    }) {
                        evidence.insert(
                            request.eval_id.clone(),
                            match request.status() {
                                crate::types::RequestStatus::Green => {
                                    Evidence::Current(Verdict::Green)
                                }
                                crate::types::RequestStatus::Red => Evidence::Current(Verdict::Red),
                                _ => Evidence::OperationalError,
                            },
                        );
                        *saved = request;
                    }
                }
            }
            if !self.cancellation.is_cancelled() {
                let recorded = self.run.evidence.len();
                for eval in &self.config.evals {
                    if evidence.contains_key(eval.id.as_str()) {
                        continue;
                    }
                    let index = self
                        .requests
                        .iter()
                        .position(|request| request.eval_id == eval.id);
                    if index.is_some_and(|index| {
                        running.contains(&index)
                            || waiting.contains(&index)
                            || self.requests[index].status()
                                == crate::types::RequestStatus::WaitingHuman
                            || self.requests[index].force
                    }) {
                        continue;
                    }
                    if let Err(error) = cache::validate_file_inputs(&self.config, eval) {
                        evidence.insert(eval.id.clone(), Evidence::OperationalError);
                        if index.is_none() && saved_evals.contains(&eval.id) {
                            self.run
                                .evidence
                                .insert(eval.id.clone(), crate::types::RequestStatus::Error);
                        }
                        if let Some(index) = index {
                            let request = &mut self.requests[index];
                            request.state = crate::store::RequestState::failed(
                                error,
                                Some(crate::types::FailureCode::PreparationFailed),
                                now(),
                            );
                            self.receipts.save_request(request).await?;
                        }
                        continue;
                    }
                    if let Some(key) = self.keys.get(&eval.id)
                        && let Some(execution) = self.receipts.cached_execution(&key.value).await?
                    {
                        let verdict = execution.verdict().expect("completed cache entry");
                        evidence.insert(eval.id.clone(), Evidence::Current(verdict));
                        if index.is_none() && saved_evals.contains(&eval.id) {
                            self.run.evidence.insert(
                                eval.id.clone(),
                                match verdict {
                                    Verdict::Green => crate::types::RequestStatus::Green,
                                    Verdict::Red => crate::types::RequestStatus::Red,
                                },
                            );
                        }
                        if let Some(index) = index {
                            waiting.remove(&index);
                            let request = &mut self.requests[index];
                            cache::reuse(request, &execution, now());
                            self.receipts.reuse_execution(request).await?;
                        }
                    }
                }
                if self.run.evidence.len() != recorded {
                    self.receipts.save_run(self.run).await?;
                }
                let mut evaluation = self
                    .graph
                    .evaluate_with_policy(&evidence, self.run.ignore_gates);
                // Backends found full in this pass are not asked again until the next one.
                let mut full = BTreeSet::new();
                // Request order preserves selection order, then recursive configuration order.
                for index in 0..self.requests.len() {
                    if self.cancellation.is_cancelled() {
                        break;
                    }
                    let request = &mut self.requests[index];
                    if running.contains(&index)
                        || request.status() == crate::types::RequestStatus::WaitingHuman
                        || evidence.contains_key(request.eval_id.as_str())
                    {
                        continue;
                    }
                    if request.profile.kind() == ProfileKind::Dependency {
                        let declared = self
                            .config
                            .evals
                            .iter()
                            .find(|eval| eval.id == request.eval_id)
                            .expect("included Eval");
                        if let Err(error) = cache::validate_file_inputs(&self.config, declared) {
                            request.state = crate::store::RequestState::failed(
                                error,
                                Some(crate::types::FailureCode::PreparationFailed),
                                now(),
                            );
                            self.receipts.save_request(request).await?;
                            evidence.insert(request.eval_id.clone(), Evidence::OperationalError);
                            continue;
                        }
                        let eval = &evaluation.evals[request.eval_id.as_str()];
                        let status = crate::project::verify::status(eval.status);
                        let blocked_by: Vec<_> =
                            eval.blocked_by.iter().copied().map(Into::into).collect();
                        if request.status() != status || request.blocked_by != blocked_by {
                            derive(request, eval, now());
                            self.receipts.save_request(request).await?;
                        }
                        continue;
                    }
                    if !evaluation.evals[request.eval_id.as_str()].can_execute() {
                        waiting.remove(&index);
                        continue;
                    }
                    if !waiting.contains(&index) && running.len() + waiting.len() >= self.run.jobs {
                        continue;
                    }
                    let declared = self
                        .config
                        .evals
                        .iter()
                        .find(|eval| eval.id == request.eval_id)
                        .expect("included Eval");
                    if let Err(error) = cache::validate_file_inputs(&self.config, declared) {
                        waiting.remove(&index);
                        request.state = crate::store::RequestState::failed(
                            error,
                            Some(crate::types::FailureCode::PreparationFailed),
                            now(),
                        );
                        self.receipts.save_request(request).await?;
                        evidence.insert(request.eval_id.clone(), Evidence::OperationalError);
                        evaluation = self
                            .graph
                            .evaluate_with_policy(&evidence, self.run.ignore_gates);
                        continue;
                    }
                    let mut execution = Execution {
                        state: crate::store::ExecutionState::Running,
                        id: format!("execution-{}", request.id).parse()?,
                        key: request.key.clone(),
                        fingerprint: request.fingerprint.clone(),
                        fingerprints: request.fingerprints.clone(),
                        artifact_kinds: self
                            .keys
                            .get(&request.eval_id)
                            .map(|key| key.artifact_kinds.clone())
                            .unwrap_or_default(),
                        eval_def_hash: request.eval_def_hash.clone(),
                        owner: Some(owner),
                        profile: request.profile.clone(),
                        options: request.options.clone(),
                        usage: None,
                        provenance: Provenance {
                            repo_path: self.config.root.clone(),
                            run_id: self.run.id.clone(),
                            request_id: request.id.clone(),
                            eval_id: request.eval_id.clone(),
                            eval_def_hash: request.eval_def_hash.clone(),
                            completed_at: None,
                            execution_paths: Default::default(),
                        },
                        started_at: now(),
                        producer: Some(producer.clone()),
                        reviewer: None,
                        origin: None,
                        manifest: None,
                    };
                    if execution.key.is_some() {
                        execution.manifest = self.fingerprints[&request.target].manifest.clone();
                    }
                    let kind = self
                        .config
                        .evals
                        .iter()
                        .find(|eval| eval.id == request.eval_id)
                        .expect("included eval")
                        .declaration
                        .profile()
                        .kind();
                    let human = kind == ProfileKind::Human;
                    // An eval of a --reuse-only kind starts nothing, joins nothing and requests
                    // no Human sign-off: it reuses a completed result or is not executed.
                    let reuse_only = self.run.reuse_only.contains(&kind);
                    // A backend this Run stopped admits no new reviews and takes no slot; reuse and
                    // joining a live execution of the key still work.
                    let mut backend_stopped = stops.of(request);
                    let allow_start = !reuse_only
                        && backend_stopped.is_none()
                        && (human
                            || self
                                .run
                                .max_executions
                                .is_none_or(|limit| self.run.executions_started < limit));
                    // An Agent review of a backend with a machine-wide limit (limits.json) also
                    // needs one of its slots. The claim is probed first, so a request that would
                    // reuse or join takes no slot; one that would start waits for a slot without
                    // holding a job slot or an executor start.
                    let limited = match &request.profile {
                        crate::config::StoredProfile::Agent { backend, .. } => request
                            .options
                            .backend
                            .as_ref()
                            .and_then(|name| Some((*name, self.limits.limit(*backend)?))),
                        _ => None,
                    };
                    let mut claim = claim(
                        self.receipts,
                        self.remote.as_deref(),
                        self.run.force,
                        &execution,
                        request,
                        allow_start && limited.is_none(),
                        None,
                    )
                    .await?;
                    if let (Claim::BudgetExhausted, Some((backend, limit)), true) =
                        (&claim, &limited, allow_start)
                    {
                        claim = if full.contains(backend) {
                            Claim::Full
                        } else {
                            let claim = self::claim(
                                self.receipts,
                                self.remote.as_deref(),
                                self.run.force,
                                &execution,
                                request,
                                true,
                                Some(Capacity {
                                    limit: *limit,
                                    stopped: stops.check(backend),
                                }),
                            )
                            .await?;
                            // The backend may have stopped while this review waited for a slot.
                            backend_stopped = stops.of(request);
                            claim
                        };
                    }
                    if !matches!(claim, Claim::Full) {
                        capacity_waiting.remove(&index);
                    }
                    match claim {
                        Claim::Full => {
                            let (backend, limit) = limited.as_ref().expect("a limited backend");
                            full.insert(*backend);
                            waiting.remove(&index);
                            if capacity_waiting.insert(index) {
                                request.queue =
                                    Some(crate::store::QueueReason::Slot { backend: *backend });
                                request.blocked_reason = Some(format!(
                                    "Waiting for a free {backend} slot: all {limit} are in use on this machine (limits.json)."
                                ));
                                self.receipts.save_request(request).await?;
                            }
                            continue;
                        }
                        Claim::Reuse(execution) => {
                            waiting.remove(&index);
                            evidence.insert(
                                request.eval_id.clone(),
                                Evidence::Current(
                                    execution.verdict().expect("completed cache entry"),
                                ),
                            );
                            cache::reuse(request, &execution, now());
                            self.receipts.reuse_execution(request).await?;
                            evaluation = self
                                .graph
                                .evaluate_with_policy(&evidence, self.run.ignore_gates);
                            continue;
                        }
                        // Nothing to reuse: the evidence stays stale, so dependents wait as
                        // behind any missing evidence, and the Run ends INCOMPLETE.
                        _ if reuse_only => {
                            waiting.remove(&index);
                            request.state = crate::store::RequestState::pending(
                                crate::types::RequestStatus::Stale,
                                request.completed_at(),
                            )
                            .expect("pending request state");
                            request.queue = None;
                            request.blocked_reason = Some(not_reused_reason(kind));
                            self.receipts.save_request(request).await?;
                            evidence.insert(request.eval_id.clone(), Evidence::Stale);
                            evaluation = self
                                .graph
                                .evaluate_with_policy(&evidence, self.run.ignore_gates);
                            continue;
                        }
                        Claim::WaitHuman(id) => {
                            waiting.remove(&index);
                            request.execution_id = Some(id);
                            request.state = crate::store::RequestState::WaitingHuman;
                            request.queue = None;
                            request.blocked_reason = Some(
                                "Waiting for the active Human execution of this reuse key.".into(),
                            );
                            *request = self.receipts.follow_human(request).await?;
                            if request.status() != crate::types::RequestStatus::WaitingHuman {
                                evidence.insert(
                                    request.eval_id.clone(),
                                    match request.status() {
                                        crate::types::RequestStatus::Green => {
                                            Evidence::Current(Verdict::Green)
                                        }
                                        crate::types::RequestStatus::Red => {
                                            Evidence::Current(Verdict::Red)
                                        }
                                        _ => Evidence::OperationalError,
                                    },
                                );
                                evaluation = self
                                    .graph
                                    .evaluate_with_policy(&evidence, self.run.ignore_gates);
                            }
                            continue;
                        }
                        Claim::Wait(id) => {
                            waiting.insert(index);
                            if request.execution_id.as_ref() != Some(&id) {
                                request.execution_id = Some(id);
                                request.state = crate::store::RequestState::pending(
                                    crate::types::RequestStatus::Queued,
                                    request.completed_at(),
                                )
                                .expect("pending request state");
                                request.queue = None;
                                request.blocked_reason = Some(
                                    "Waiting for the active execution of this reuse key.".into(),
                                );
                                self.receipts.save_request(request).await?;
                            }
                            continue;
                        }
                        Claim::BudgetExhausted if let Some(cause) = backend_stopped => {
                            waiting.remove(&index);
                            // A request waiting for a slot stops waiting.
                            request.queue = None;
                            request.blocked_reason = None;
                            request.state = crate::store::RequestState::failed(
                                format!(
                                    "Not started: this Run stopped admitting {} reviews after {} in {}: {}",
                                    cause.backend, cause.error_code, cause.eval_id, cause.error
                                ),
                                Some(crate::types::FailureCode::BackendStopped),
                                now(),
                            );
                            self.receipts.save_request(request).await?;
                            evidence.insert(request.eval_id.clone(), Evidence::OperationalError);
                            evaluation = self
                                .graph
                                .evaluate_with_policy(&evidence, self.run.ignore_gates);
                            continue;
                        }
                        Claim::BudgetExhausted => {
                            waiting.remove(&index);
                            let reason = budget_reason(self.run);
                            // Keep looking for reuse, but an unchanged budget decision cannot
                            // create a self-notification → write → notification hot loop.
                            if request.status() != crate::types::RequestStatus::BudgetExhausted
                                || request.blocked_reason.as_deref() != Some(reason.as_str())
                            {
                                request.state = crate::store::RequestState::pending(
                                    crate::types::RequestStatus::BudgetExhausted,
                                    request.completed_at(),
                                )
                                .expect("pending request state");
                                request.queue = None;
                                request.blocked_reason = Some(reason);
                                self.receipts.save_request(request).await?;
                            }
                            continue;
                        }
                        Claim::Owned => {}
                    }
                    waiting.remove(&index);
                    // Followers join a claimed execution by its id.
                    request.execution_id = execution
                        .key
                        .as_ref()
                        .filter(|_| !request.force)
                        .map(|_| execution.id.clone());
                    request.queue = None;
                    request.blocked_reason = None;
                    let eval = self
                        .config
                        .evals
                        .iter()
                        .find(|eval| eval.id == request.eval_id)
                        .expect("included eval");
                    let prepared = execution::prepare(
                        &self.config,
                        eval,
                        &run_dir,
                        &self.run.state_dir,
                        saving.as_ref(),
                        request,
                    );
                    if prepared.is_ok() && !human && !self.cancellation.is_cancelled() {
                        self.run.executions_started += 1;
                        execution.started_at = now();
                        request.started_at = Some(execution.started_at);
                    }
                    request.state = crate::store::RequestState::Running;
                    self.receipts.save_run(self.run).await?;
                    self.receipts.save_request(request).await?;
                    let config = self.config.clone();
                    let receipts = self.receipts.clone();
                    let request = request.clone();
                    let remote = self.remote.clone();
                    let cancellation = self.cancellation.clone();
                    let run_dir = run_dir.clone();
                    let parallelism = self.parallelism.clone();
                    let agent = request.profile.kind() == crate::config::ProfileKind::Agent;
                    let stops = stops.clone();
                    running.insert(index);
                    self.tasks.spawn(async move {
                        let mut execution = execution;
                        // An Agent result pins the files its tools execute, hashed before the
                        // review starts; a path that cannot be hashed fails the preparation.
                        let prepared = match prepared {
                            Ok(prepared) if agent && !cancellation.is_cancelled() => {
                                match tools::pins::execution_paths(&config, &request.eval_id).await
                                {
                                    Ok(pins) => {
                                        execution.provenance.execution_paths = pins;
                                        Ok(prepared)
                                    }
                                    Err(error) => Err(error),
                                }
                            }
                            prepared => prepared,
                        };
                        let request = execution::execute(
                            config,
                            receipts.clone(),
                            request,
                            execution,
                            prepared,
                            run_dir,
                            &parallelism,
                            &stops,
                            cancellation,
                        )
                        .await?;
                        // Publication runs after the local commit, outside any transaction.
                        if let Some(remote) = remote {
                            remote.publish_request(&receipts, &request).await?;
                        }
                        Ok((index, request))
                    });
                }
            }
            let cancelled = self.cancellation.is_cancelled();
            if !cancelled && evidence.len() != completed {
                continue;
            }
            let human_wait = deadline.is_some()
                && self
                    .requests
                    .iter()
                    .any(|request| request.status() == crate::types::RequestStatus::WaitingHuman);
            let current = tokio::time::Instant::now();
            if self.tasks.is_empty()
                && ((waiting.is_empty() && capacity_waiting.is_empty()) || cancelled)
            {
                if cancelled || !human_wait {
                    break;
                }
                if deadline.is_some_and(|deadline| current >= deadline) {
                    self.run.wait_timed_out = true;
                    break;
                }
            }
            let waiting_external =
                !waiting.is_empty() || !capacity_waiting.is_empty() || human_wait;
            let wake_deadline = future_deadline(deadline, human_wait, current);
            let deadline_at = wake_deadline.unwrap_or(current);
            tokio::select! {
                biased;
                result = self.tasks.join_next(), if !self.tasks.is_empty() => {
                    let (index, request) = crate::task::joined(result.expect("active tasks"))
                        .map_err(|e| e.to_string())??;
                    running.remove(&index);
                    let stopped = stops.all();
                    if stopped.len() != self.run.stopped_backends.len() {
                        self.run.stopped_backends = stopped;
                        self.receipts.save_run(self.run).await?;
                    }
                    if request.status() != crate::types::RequestStatus::WaitingHuman {
                        evidence.insert(request.eval_id.clone(), match request.status() {
                            crate::types::RequestStatus::Green => Evidence::Current(Verdict::Green),
                            crate::types::RequestStatus::Red => Evidence::Current(Verdict::Red),
                            _ => Evidence::OperationalError,
                        });
                    }
                    self.requests[index] = request;
                }
                _ = self.cancellation.cancelled(), if !cancelled => {},
                _ = changes.next_state(), if !cancelled => {},
                _ = reconcile.tick(), if waiting_external && !cancelled => {},
                _ = tokio::time::sleep_until(deadline_at),
                    if wake_deadline.is_some() && !cancelled => {},
                _ = tokio::time::sleep_until(remote_due),
                    if human_wait && self.remote.is_some() && !self.run.force && !cancelled => {},
            }
        }
        Ok(evidence)
    }
}

/// Claim an execution for a request: a forced review takes no claim of its key and joins
/// nothing, while any other first looks up the remote store again (a hit becomes a local
/// record). With `capacity`, starting also takes a slot of the backend.
async fn claim(
    receipts: &Receipts,
    remote: Option<&Session>,
    run_force: bool,
    execution: &Execution,
    request: &Request,
    allow_start: bool,
    capacity: Option<Capacity>,
) -> Result<Claim, String> {
    let keyed = execution.key.is_some() && !request.force;
    if let (Some(remote), Some(key), true, false) = (remote, &execution.key, keyed, run_force) {
        remote.refresh(receipts, vec![key.clone()]).await?;
    }
    receipts
        .claim_execution(
            execution,
            keyed,
            request.execution_id.as_ref(),
            allow_start,
            capacity,
        )
        .await
}