artifactize_core/project/

verify.rs

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

use serde_json::json;
use tokio_util::sync::CancellationToken;

use crate::{
    broker::{self, now},
    cache,
    config::{DependencyGates, ProfileKind, read_workspace_config},
    graph::{EvalStatus, Evidence, Graph},
    project::selection::{ProfileSelection, Selection, select_profiles},
    remote::Session,
    store::{self, ExecutionOptions, Receipts, Request, Run, RunView},
    workspace,
};

/// Give Human sign-off ten minutes by default, then return without cancelling
/// pending requests; later verify/run queries can still observe their settlement.
pub const DEFAULT_HUMAN_WAIT: Duration = Duration::from_secs(600);
/// Four concurrent evals keep ordinary machines responsive while overlapping review I/O;
/// shared by API/CLI defaults and legacy Run decoding, and overridable with --jobs.
pub const DEFAULT_JOBS: usize = 4;

#[derive(Debug, Clone)]
pub struct VerifyOptions {
    pub jobs: usize,
    /// Fingerprints computed at once, in preparation and end-of-review rechecks; `None` uses
    /// the available CPUs.
    pub fingerprint_jobs: Option<usize>,
    /// Print `Run: RUN_ID` to stderr as soon as the Run is saved, before any eval runs.
    pub announce_run: bool,
    pub max_executions: Option<u64>,
    /// How long this Run waits for Human submissions; the timeout never cancels a review.
    pub wait_timeout: Duration,
    /// Kinds whose evals only reuse a result, from the cache or the review store; one with
    /// nothing to reuse is not executed, and the Run ends INCOMPLETE.
    pub reuse_only: BTreeSet<ProfileKind>,
    pub profile: Option<ProfileSelection>,
    pub recursive: bool,
    /// Force only explicitly selected Evals, never recursive dependencies.
    pub force: bool,
    /// None uses root reviewPolicy; Some(false) explicitly enforces GREEN gates.
    pub ignore_gates: Option<bool>,
}

impl Default for VerifyOptions {
    fn default() -> Self {
        Self {
            jobs: DEFAULT_JOBS,
            fingerprint_jobs: None,
            announce_run: false,
            max_executions: None,
            wait_timeout: DEFAULT_HUMAN_WAIT,
            reuse_only: BTreeSet::new(),
            profile: None,
            recursive: false,
            force: false,
            ignore_gates: None,
        }
    }
}

/// Executes READY evals in the foreground, reusing completed results by fingerprint.
pub async fn verify(
    repo: &Path,
    state_dir: Option<&Path>,
    selection: &Selection,
    options: &VerifyOptions,
    cancellation: CancellationToken,
) -> Result<RunView, String> {
    if options.jobs == 0 {
        return Err("jobs must be at least 1.".into());
    }
    let parallelism = super::fingerprint_parallelism(options)?;
    let wait_timeout_ms = u32::try_from(options.wait_timeout.as_millis())
        .ok()
        .filter(|ms| (1..=crate::config::validation::MAX_TIMEOUT_MS).contains(&u64::from(*ms)))
        .ok_or("wait timeout must be between 1 and 2147483647 ms.")?;
    let config = read_workspace_config(repo).map_err(|e| e.to_string())?;
    let config = Arc::new(select_profiles(
        config,
        selection,
        options.profile.as_ref(),
        options.recursive,
    )?);
    let ignore_gates = options.ignore_gates.unwrap_or_else(|| {
        config
            .artifacts
            .values()
            .find(|artifact| artifact.path.as_os_str().is_empty())
            .and_then(|artifact| artifact.review_policy.as_ref())
            .is_some_and(|policy| matches!(policy.dependency_gates, Some(DependencyGates::Ignore)))
    });
    let graph = Graph::new(&config).map_err(|e| e.to_string())?;
    let selected = selection.resolve(&config)?;
    let selected_ids: BTreeSet<_> = selected.evals.iter().map(|eval| &eval.id).collect();
    let required: BTreeSet<_> = graph
        .dependency_closure(&selected.roots)
        .map_err(|e| e.to_string())?
        .into_iter()
        .collect();
    let evals = selection.included_evals(&config, options.recursive)?;
    let definitions =
        store::definitions::Definitions::from_view(&crate::query::graph(&config, selection)?)
            .map_err(|error| error.to_string())?;
    let state = store::state_dir(state_dir)?;
    let limits = crate::limits::Limits::read(&state)?;
    let receipts = Receipts::open(&state, &config.root).await?;
    let runs = workspace::prepare_directory(&state.join("runs"), &config.root)
        .map_err(|e| e.to_string())?;
    let fingerprints = cache::prepare(
        &config,
        cache::fingerprint_targets(&config, &required),
        &runs,
        &parallelism,
        cancellation.clone(),
    )
    .await?;
    if cancellation.is_cancelled() {
        return Err("Project preparation was cancelled.".into());
    }
    let keys = cache::eval_keys(&config, &fingerprints);
    // The store is asked once for every key, and a record that completed after the local
    // latest is mirrored into the local history before the Run claims anything. A forced Run
    // reads nothing from the store, but publishes its results like any other.
    let remote = Session::open(Some(&state), Some(&config.root))?;
    // Results from a fake provider must never reach the shared review store.
    if remote.is_some()
        && let Some(variable) = crate::llm::active_test_endpoint()
    {
        return Err(format!(
            "{variable} points Agent reviews at a local test endpoint; set ARTIFACTIZE_REMOTE=off so their results stay out of the review store."
        ));
    }
    if let Some(remote) = &remote
        && !options.force
    {
        remote
            .refresh(
                &receipts,
                keys.values().map(|key| key.value.clone()).collect(),
            )
            .await?;
    }
    let directory = tempfile::Builder::new()
        .prefix("run-")
        .tempdir_in(runs)
        .map_err(|e| e.to_string())?;
    let id = directory
        .path()
        .file_name()
        .unwrap()
        .to_str()
        .expect("generated run directory names are ASCII")
        .to_owned();
    let _ = directory.keep();
    let mut run = Run {
        id: id.parse()?,
        repo_path: config.root.clone(),
        repository: crate::repository::identify(&config.root),
        state_dir: state,
        state: crate::store::RunState::Running,
        created_at: now(),
        selection: selection.clone(),
        profile: options.profile.clone(),
        definitions,
        jobs: options.jobs,
        fingerprint_jobs: Some(parallelism.limit()),
        max_executions: options.max_executions,
        executions_started: 0,
        wait_timeout_ms: Some(Duration::from_millis(u64::from(wait_timeout_ms))),
        wait_timed_out: false,
        reuse_only: options.reuse_only.clone(),
        recursive: options.recursive,
        force: options.force,
        ignore_gates,
        validation: Default::default(),
        stopped_backends: Vec::new(),
        evidence: BTreeMap::new(),
    };
    let mut requests: Vec<_> = evals
        .iter()
        .enumerate()
        .map(|(ordinal, eval)| Request {
            state: crate::store::RequestState::pending(crate::types::RequestStatus::Queued, None)
                .expect("queued state"),
            id: format!("{}-{}", run.id, ordinal + 1)
                .parse()
                .expect("generated request id is a safe segment"),
            run_id: run.id.clone(),
            eval_id: eval.id.clone(),
            target: eval.target.clone(),
            title: eval.declaration.title.clone(),
            profile: (eval.declaration.profile()).into(),
            requested_profile: (eval.declaration.profile()).into(),
            eval_def_hash: cache::eval_definition_hash(&eval.declaration),
            options: ExecutionOptions::new(eval.declaration.profile(), eval.variant.as_ref()),
            execution_id: None,
            provenance: None,
            usage: None,
            reused_usage: None,
            joined: false,
            producer: None,
            reviewer: None,
            origin: None,
            session_id: None,
            session: None,
            human_definition: None,
            payload: eval.declaration.payload().into(),
            references: eval.references.clone(),
            deps: eval.deps.clone(),
            force: options.force
                && selected_ids.contains(&eval.id)
                && !matches!(
                    eval.declaration.profile(),
                    crate::config::Profile::Dependency { .. }
                ),
            fingerprint: fingerprints
                .get(&eval.target)
                .map(|fingerprint| fingerprint.value.clone()),
            key: keys.get(&eval.id).map(|key| key.value.clone()),
            fingerprints: keys
                .get(&eval.id)
                .map(|key| key.fingerprints.clone())
                .unwrap_or_default(),
            created_at: run.created_at,
            started_at: None,
            cwd: config.root.join(config.artifacts[&eval.target].folder()),
            run_dir: None,
            argv: None,
            child: None,
            blocked_reason: None,
            queue: None,
            blocked_by: Vec::new(),
        })
        .collect();
    receipts.create_run(&run, &requests).await?;
    if options.announce_run {
        use std::io::Write;
        let _ = writeln!(
            std::io::stderr().lock(),
            "Started {} (follow: artifactize monitor)",
            run.id
        );
    }
    let mut evidence = match broker::schedule(
        config.clone(),
        &graph,
        &fingerprints,
        &keys,
        &parallelism,
        &limits,
        &mut run,
        &mut requests,
        &receipts,
        remote,
        cancellation.clone(),
    )
    .await
    {
        Ok(evidence) => evidence,
        Err(error) => {
            // A failing Run, for example on a rejected remote token, does not stay RUNNING.
            run.state = crate::store::RunState::Error {
                at: now(),
                error: Some(error.clone()),
            };
            let _ = receipts.save_run(&run).await;
            return Err(error);
        }
    };
    let evaluation = graph.evaluate_with_policy(&evidence, ignore_gates);
    for request in &mut requests {
        if request.profile.kind() == ProfileKind::Dependency {
            if request.status() == crate::types::RequestStatus::Error {
                continue;
            }
            broker::derive(request, &evaluation.evals[request.eval_id.as_str()], now());
            continue;
        }
        if evidence.contains_key(request.eval_id.as_str())
            || request.status() == crate::types::RequestStatus::WaitingHuman
        {
            continue;
        }
        let eval = &evaluation.evals[request.eval_id.as_str()];
        if cancellation.is_cancelled() {
            request.state = crate::store::RequestState::failed(
                "Run was cancelled.".into(),
                Some(crate::types::FailureCode::Cancelled),
                now(),
            );
            evidence.insert(request.eval_id.clone(), Evidence::OperationalError);
        } else if request.status() == crate::types::RequestStatus::BudgetExhausted
            && eval.can_execute()
        {
            continue;
        } else {
            request.state =
                crate::store::RequestState::pending(status(eval.status), request.completed_at())
                    .expect("non-executed request state");
            request.queue = None;
            request.blocked_reason = Some(format!(
                "{}: {}",
                if eval.status == EvalStatus::Blocked {
                    "Dependency verdict RED"
                } else {
                    "Waiting for current GREEN dependency evidence"
                },
                eval.unmet_gates
                    .iter()
                    .map(ToString::to_string)
                    .collect::<Vec<_>>()
                    .join(", ")
            ));
        }
    }
    let evaluation = graph.evaluate_with_policy(&evidence, ignore_gates);
    let required_evals: Vec<_> = evaluation
        .evals
        .iter()
        .filter(|(id, _)| required.contains(&graph.eval_target(id).unwrap()))
        .collect();
    let satisfied = required
        .iter()
        .all(|id| evaluation.artifacts[id.as_str()].satisfied);
    let budget_exhausted = requests
        .iter()
        .any(|request| request.status() == crate::types::RequestStatus::BudgetExhausted);
    // Only a --reuse-only eval that had nothing to reuse leaves stale evidence in a Run.
    let not_reused = evidence
        .values()
        .filter(|evidence| matches!(evidence, Evidence::Stale))
        .count();
    let at = now();
    run.state = if cancellation.is_cancelled() {
        crate::store::RunState::Error {
            at,
            error: Some("Run was cancelled.".into()),
        }
    } else if run.wait_timed_out {
        crate::store::RunState::Incomplete {
            at,
            error: Some(
                "Human wait timed out; pending requests remain available for submission.".into(),
            ),
        }
    } else if budget_exhausted {
        crate::store::RunState::Incomplete {
            at,
            error: Some(broker::budget_reason(&run)),
        }
    } else if not_reused > 0 {
        crate::store::RunState::Incomplete {
            at,
            error: Some(format!(
                "{not_reused} eval{} had no result to reuse and {} not executed (--reuse-only).",
                if not_reused == 1 { "" } else { "s" },
                if not_reused == 1 { "was" } else { "were" }
            )),
        }
    } else if required_evals
        .iter()
        .any(|(_, c)| c.status == EvalStatus::Error)
    {
        crate::store::RunState::Error { at, error: None }
    } else if required_evals
        .iter()
        .any(|(_, c)| c.status == EvalStatus::Red)
    {
        crate::store::RunState::Red { at }
    } else if satisfied {
        crate::store::RunState::Green { at }
    } else {
        crate::store::RunState::Incomplete { at, error: None }
    };
    run.validation = serde_json::from_value(json!({
        "selection":run.selection,
        "recursive":run.recursive,
        "force":run.force,
        "ignoreGates":run.ignore_gates,
        "selectedEvalIds":selected.evals.iter().map(|eval| &eval.id).collect::<Vec<_>>(),
        "includedEvalIds":evals.iter().map(|eval| &eval.id).collect::<Vec<_>>(),
        "satisfied":satisfied && !cancellation.is_cancelled(),
        "obligations":evaluation
            .obligations
            .iter()
            .filter(|id| required.contains(*id))
            .collect::<Vec<_>>(),
        "artifacts":required
            .iter()
            .map(|id| {
                let a = &evaluation.artifacts[id.as_str()];
                let mut artifact = json!({
                    "id":id,
                    "status":format!("{:?}", a.status).to_uppercase(),
                    "passed":a.passed,
                    "total":a.total,
                    "satisfied":a.satisfied,
                });
                if let Some(fingerprint) = fingerprints.get(id) {
                    artifact["fingerprintKind"] = json!(if fingerprint.manifest.is_some() {
                        "artifactsum"
                    } else {
                        "script"
                    });
                    artifact["fingerprint"] = json!(fingerprint.value);
                }
                artifact
            })
            .collect::<Vec<_>>(),
        "evals":required_evals
            .iter()
            .map(|(id, eval)| {
                json!({
                    "id":id,
                    "status":status(eval.status),
                    "blockedBy":eval.blocked_by,
                })
            })
            .collect::<Vec<_>>(),
    }))
    .expect("generated validation snapshot");
    receipts.finish(&run, &requests).await?;
    // Saved Agent conversations past their size bound are collected; a failure only warns.
    let state = run.state_dir.clone();
    let bounds = limits.agent_sessions();
    let collected = crate::task::joined(
        tokio::task::spawn_blocking(move || crate::agent::session::collect(&state, bounds, false))
            .await,
    )
    .map_err(|e| e.to_string())
    .and_then(|collected| collected);
    if let Err(error) = collected {
        use std::io::Write;
        let _ = writeln!(
            std::io::stderr().lock(),
            "Agent session collection failed: {error}"
        );
    }
    store::read_run(&run.state_dir, &run.id).await
}

pub(crate) fn status(status: EvalStatus) -> crate::types::RequestStatus {
    match status {
        EvalStatus::Green => crate::types::RequestStatus::Green,
        EvalStatus::Red => crate::types::RequestStatus::Red,
        EvalStatus::Error => crate::types::RequestStatus::Error,
        EvalStatus::Stale => crate::types::RequestStatus::Stale,
        EvalStatus::Unreviewed => crate::types::RequestStatus::Unreviewed,
        EvalStatus::Wait => crate::types::RequestStatus::WaitDependency,
        EvalStatus::Blocked => crate::types::RequestStatus::Blocked,
    }
}