artifactize-core 0.13.0

//! Human waiting, claim locks, tools, and submission.

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

use serde_json::{Value, json};
use tokio_util::sync::CancellationToken;

use crate::{
    agent::verdict::validate_result,
    cache,
    config::{Eval, RepoConfig, read_workspace_config},
    remote::Session,
    scope,
    store::{self, HumanClaim, Receipts, Request},
    tools::human::{CommandLine, Registry, ToolResult},
};

/// Bound owner fields and serialized Human results consistently across CLI, editor
/// and submission so untrusted JSON cannot grow the review audit without limit.
pub const MAX_RESULT_BYTES: usize = 256_000;
/// Read one sentinel byte beyond the bound to detect oversized files without loading them.
pub const FIELDS_READ_BYTES: u64 = MAX_RESULT_BYTES as u64 + 1;

pub fn default_reviewer() -> Result<crate::types::ReviewerId, String> {
    crate::platform::USER_VARIABLES
        .iter()
        .find_map(|name| crate::platform::environment::var_text(name))
        .ok_or("Provide an explicit reviewer id; no default reviewer is available.")?
        .parse()
}

/// The saved request's Run repository and its receipts, where its Human actions are recorded.
pub async fn open(
    state: &Path,
    request: &crate::types::RequestId,
) -> Result<(Receipts, PathBuf), String> {
    let (_, repo) = store::read_request_repo(state, request).await?;
    Ok((Receipts::open(state, &repo).await?, repo))
}

pub async fn claim(
    receipts: &Receipts,
    request: &crate::types::RequestId,
    reviewer: &crate::types::ReviewerId,
) -> Result<HumanClaim, String> {
    receipts.claim_human(request, reviewer).await
}

/// Release the reviewer lock without a verdict, so another reviewer can claim the request.
pub async fn unclaim(
    receipts: &Receipts,
    request: &crate::types::RequestId,
    reviewer: &crate::types::ReviewerId,
) -> Result<HumanClaim, String> {
    receipts.release_human(request, reviewer).await
}

/// Run `tool` for the reviewer who holds the claim, its output reaching `run`'s Tool log. A
/// refusal before it starts ends the run with the refusal.
pub async fn run_human_tool(
    receipts: &Receipts,
    request: &crate::types::RequestId,
    reviewer: &crate::types::ReviewerId,
    tool: &crate::config::ToolName,
    cancellation: CancellationToken,
    run: crate::tools::human::log::Run,
) -> Result<ToolResult, String> {
    let ready = async {
        let (mut request, _) = receipts.human_request(request, reviewer).await?;
        let config = reconnect(&request)?;
        let registry = Registry::new(&config, &request.eval_id)?;
        if !registry.list().any(|entry| &entry.name == tool) {
            return Err::<_, String>("Unknown registered Human tool.".into());
        }
        recheck(
            receipts,
            &mut request,
            reviewer,
            &config,
            cancellation.clone(),
        )
        .await?;
        receipts.human_request(&request.id, reviewer).await?;
        Ok((config, request.eval_id))
    };
    let (config, eval) = match ready.await {
        Ok(ready) => ready,
        Err(error) => {
            run.end(crate::tools::human::log::End::Failed(error.clone()));
            return Err(error);
        }
    };
    let registry = Registry::new(&config, &eval).expect("the registry was just built");
    Ok(registry.call_logged(tool, cancellation, run).await)
}

/// What every registered Human tool of a waiting request would run, by tool name; needs no
/// claim and runs nothing. A tool that cannot be resolved carries its error.
pub async fn tool_commands(
    receipts: &Receipts,
    request: &crate::types::RequestId,
) -> Result<std::collections::BTreeMap<crate::config::ToolName, Result<CommandLine, String>>, String>
{
    let request = receipts.waiting_human(request).await?;
    let config = reconnect(&request)?;
    let registry = Registry::new(&config, &request.eval_id)?;
    Ok(registry
        .list()
        .filter(|tool| registry.is_command(&tool.name))
        .map(|tool| (tool.name.clone(), registry.command(&tool.name)))
        .collect())
}

/// Invalid results remain correctable; only a valid submission performs the final fingerprint
/// check.
pub async fn submit(
    receipts: &Receipts,
    request: &crate::types::RequestId,
    reviewer: &crate::types::ReviewerId,
    result: &Value,
    cancellation: CancellationToken,
) -> Result<Request, String> {
    let (mut request, _) = receipts.human_request(request, reviewer).await?;
    let config = reconnect(&request)?;
    let eval = config
        .evals
        .iter()
        .find(|eval| eval.id == request.eval_id)
        .expect("reconnected eval");
    if serde_json::to_vec(result).map_err(|e| e.to_string())?.len() > MAX_RESULT_BYTES {
        return Err(format!("Human result exceeds {MAX_RESULT_BYTES} bytes."));
    }
    let result = validate_result(&eval.declaration, result)?;
    recheck(receipts, &mut request, reviewer, &config, cancellation).await?;
    request.state = crate::store::RequestState::completed(result.into(), crate::broker::now());
    receipts.settle_human(&request, reviewer).await
}

/// `submit`, then publish the settled result to the configured remote review store.
pub async fn submit_and_publish(
    state: &Path,
    request: &crate::types::RequestId,
    reviewer: &crate::types::ReviewerId,
    result: &Value,
    cancellation: CancellationToken,
) -> Result<Request, String> {
    let (receipts, repo) = open(state, request).await?;
    let result = async {
        let remote = Session::open(Some(state), Some(&repo))?;
        let request = submit(&receipts, request, reviewer, result, cancellation).await?;
        if let Some(remote) = remote {
            remote
                .publish_request(&receipts, &request)
                .await
                .map_err(|error| {
                    format!("The Human result is saved locally, but publishing it failed: {error}")
                })?;
        }
        Ok(request)
    }
    .await;
    let closed = receipts.close().await;
    result.and_then(|request| closed.map(|()| request))
}

pub(crate) fn definition(
    config: &RepoConfig,
    eval: &Eval,
) -> Result<crate::store::HumanDefinition, String> {
    let scope = scope::eval_scope(config, eval).map_err(|e| e.to_string())?;
    serde_json::from_value(json!({"repo":crate::platform::path_text(&config.root),"eval":eval,"artifacts":scope.artifacts})).map_err(|e| e.to_string())
}

fn reconnect(request: &Request) -> Result<RepoConfig, String> {
    let repo = &request
        .provenance
        .as_ref()
        .ok_or("Human request has no provenance.")?
        .repo_path;
    let config = read_workspace_config(repo).map_err(|e| e.to_string())?;
    let eval = config
        .evals
        .iter()
        .find(|eval| eval.id == request.eval_id)
        .ok_or("Recorded Human eval no longer exists.")?;
    if Some(definition(&config, eval)?) != request.human_definition {
        return Err(
            "Human Artifact scope or eval declarations changed since the request was recorded."
                .into(),
        );
    }
    Ok(config)
}

async fn recheck(
    receipts: &Receipts,
    request: &mut Request,
    reviewer: &crate::types::ReviewerId,
    config: &RepoConfig,
    cancellation: CancellationToken,
) -> Result<(), String> {
    if cancellation.is_cancelled() {
        return Err("Human action was cancelled.".into());
    }
    let Some(expected) = &request.key else {
        return Ok(());
    };
    let output = request
        .run_dir
        .as_deref()
        .ok_or("Human request has no output directory.")?;
    let eval = config
        .evals
        .iter()
        .find(|eval| eval.id == request.eval_id)
        .ok_or("Recorded Human eval no longer exists.")?;
    let parallelism = cache::Parallelism::new(cache::Parallelism::available());
    let (code, error) =
        match cache::recheck(config, eval, output, &parallelism, cancellation.clone()).await {
            Ok(Some(actual)) if &actual == expected => return Ok(()),
            Ok(_) => (
                crate::types::FailureCode::InputChanged,
                "Fingerprint changed during review.".into(),
            ),
            Err(error) if cancellation.is_cancelled() => return Err(error),
            Err(error) => (crate::types::FailureCode::FingerprintRecheckFailed, error),
        };
    request.state =
        crate::store::RequestState::failed(error.clone(), Some(code), crate::broker::now());
    receipts.settle_human(request, reviewer).await?;
    Err(error)
}