artifactize_core/
human.rs

1//! Human waiting, claim locks, tools, and submission.
2
3use std::path::{Path, PathBuf};
4
5use serde_json::{Value, json};
6use tokio_util::sync::CancellationToken;
7
8use crate::{
9    agent::verdict::validate_result,
10    cache,
11    config::{Eval, RepoConfig, read_workspace_config},
12    remote::Session,
13    scope,
14    store::{self, HumanClaim, Receipts, Request},
15    tools::human::{CommandLine, Registry, ToolResult},
16};
17
18/// Bound owner fields and serialized Human results consistently across CLI, editor
19/// and submission so untrusted JSON cannot grow the review audit without limit.
20pub const MAX_RESULT_BYTES: usize = 256_000;
21/// Read one sentinel byte beyond the bound to detect oversized files without loading them.
22pub const FIELDS_READ_BYTES: u64 = MAX_RESULT_BYTES as u64 + 1;
23
24pub fn default_reviewer() -> Result<crate::types::ReviewerId, String> {
25    crate::platform::USER_VARIABLES
26        .iter()
27        .find_map(|name| crate::platform::environment::var_text(name))
28        .ok_or("Provide an explicit reviewer id; no default reviewer is available.")?
29        .parse()
30}
31
32/// The saved request's Run repository and its receipts, where its Human actions are recorded.
33pub async fn open(
34    state: &Path,
35    request: &crate::types::RequestId,
36) -> Result<(Receipts, PathBuf), String> {
37    let (_, repo) = store::read_request_repo(state, request).await?;
38    Ok((Receipts::open(state, &repo).await?, repo))
39}
40
41pub async fn claim(
42    receipts: &Receipts,
43    request: &crate::types::RequestId,
44    reviewer: &crate::types::ReviewerId,
45) -> Result<HumanClaim, String> {
46    receipts.claim_human(request, reviewer).await
47}
48
49/// Release the reviewer lock without a verdict, so another reviewer can claim the request.
50pub async fn unclaim(
51    receipts: &Receipts,
52    request: &crate::types::RequestId,
53    reviewer: &crate::types::ReviewerId,
54) -> Result<HumanClaim, String> {
55    receipts.release_human(request, reviewer).await
56}
57
58/// Run `tool` for the reviewer who holds the claim, its output reaching `run`'s Tool log. A
59/// refusal before it starts ends the run with the refusal.
60pub async fn run_human_tool(
61    receipts: &Receipts,
62    request: &crate::types::RequestId,
63    reviewer: &crate::types::ReviewerId,
64    tool: &crate::config::ToolName,
65    cancellation: CancellationToken,
66    run: crate::tools::human::log::Run,
67) -> Result<ToolResult, String> {
68    let ready = async {
69        let (mut request, _) = receipts.human_request(request, reviewer).await?;
70        let config = reconnect(&request)?;
71        let registry = Registry::new(&config, &request.eval_id)?;
72        if !registry.list().any(|entry| &entry.name == tool) {
73            return Err::<_, String>("Unknown registered Human tool.".into());
74        }
75        recheck(
76            receipts,
77            &mut request,
78            reviewer,
79            &config,
80            cancellation.clone(),
81        )
82        .await?;
83        receipts.human_request(&request.id, reviewer).await?;
84        Ok((config, request.eval_id))
85    };
86    let (config, eval) = match ready.await {
87        Ok(ready) => ready,
88        Err(error) => {
89            run.end(crate::tools::human::log::End::Failed(error.clone()));
90            return Err(error);
91        }
92    };
93    let registry = Registry::new(&config, &eval).expect("the registry was just built");
94    Ok(registry.call_logged(tool, cancellation, run).await)
95}
96
97/// What every registered Human tool of a waiting request would run, by tool name; needs no
98/// claim and runs nothing. A tool that cannot be resolved carries its error.
99pub async fn tool_commands(
100    receipts: &Receipts,
101    request: &crate::types::RequestId,
102) -> Result<std::collections::BTreeMap<crate::config::ToolName, Result<CommandLine, String>>, String>
103{
104    let request = receipts.waiting_human(request).await?;
105    let config = reconnect(&request)?;
106    let registry = Registry::new(&config, &request.eval_id)?;
107    Ok(registry
108        .list()
109        .filter(|tool| registry.is_command(&tool.name))
110        .map(|tool| (tool.name.clone(), registry.command(&tool.name)))
111        .collect())
112}
113
114/// Invalid results remain correctable; only a valid submission performs the final fingerprint
115/// check.
116pub async fn submit(
117    receipts: &Receipts,
118    request: &crate::types::RequestId,
119    reviewer: &crate::types::ReviewerId,
120    result: &Value,
121    cancellation: CancellationToken,
122) -> Result<Request, String> {
123    let (mut request, _) = receipts.human_request(request, reviewer).await?;
124    let config = reconnect(&request)?;
125    let eval = config
126        .evals
127        .iter()
128        .find(|eval| eval.id == request.eval_id)
129        .expect("reconnected eval");
130    if serde_json::to_vec(result).map_err(|e| e.to_string())?.len() > MAX_RESULT_BYTES {
131        return Err(format!("Human result exceeds {MAX_RESULT_BYTES} bytes."));
132    }
133    let result = validate_result(&eval.declaration, result)?;
134    recheck(receipts, &mut request, reviewer, &config, cancellation).await?;
135    request.state = crate::store::RequestState::completed(result.into(), crate::broker::now());
136    receipts.settle_human(&request, reviewer).await
137}
138
139/// `submit`, then publish the settled result to the configured remote review store.
140pub async fn submit_and_publish(
141    state: &Path,
142    request: &crate::types::RequestId,
143    reviewer: &crate::types::ReviewerId,
144    result: &Value,
145    cancellation: CancellationToken,
146) -> Result<Request, String> {
147    let (receipts, repo) = open(state, request).await?;
148    let result = async {
149        let remote = Session::open(Some(state), Some(&repo))?;
150        let request = submit(&receipts, request, reviewer, result, cancellation).await?;
151        if let Some(remote) = remote {
152            remote
153                .publish_request(&receipts, &request)
154                .await
155                .map_err(|error| {
156                    format!("The Human result is saved locally, but publishing it failed: {error}")
157                })?;
158        }
159        Ok(request)
160    }
161    .await;
162    let closed = receipts.close().await;
163    result.and_then(|request| closed.map(|()| request))
164}
165
166pub(crate) fn definition(
167    config: &RepoConfig,
168    eval: &Eval,
169) -> Result<crate::store::HumanDefinition, String> {
170    let scope = scope::eval_scope(config, eval).map_err(|e| e.to_string())?;
171    serde_json::from_value(json!({"repo":crate::platform::path_text(&config.root),"eval":eval,"artifacts":scope.artifacts})).map_err(|e| e.to_string())
172}
173
174fn reconnect(request: &Request) -> Result<RepoConfig, String> {
175    let repo = &request
176        .provenance
177        .as_ref()
178        .ok_or("Human request has no provenance.")?
179        .repo_path;
180    let config = read_workspace_config(repo).map_err(|e| e.to_string())?;
181    let eval = config
182        .evals
183        .iter()
184        .find(|eval| eval.id == request.eval_id)
185        .ok_or("Recorded Human eval no longer exists.")?;
186    if Some(definition(&config, eval)?) != request.human_definition {
187        return Err(
188            "Human Artifact scope or eval declarations changed since the request was recorded."
189                .into(),
190        );
191    }
192    Ok(config)
193}
194
195async fn recheck(
196    receipts: &Receipts,
197    request: &mut Request,
198    reviewer: &crate::types::ReviewerId,
199    config: &RepoConfig,
200    cancellation: CancellationToken,
201) -> Result<(), String> {
202    if cancellation.is_cancelled() {
203        return Err("Human action was cancelled.".into());
204    }
205    let Some(expected) = &request.key else {
206        return Ok(());
207    };
208    let output = request
209        .run_dir
210        .as_deref()
211        .ok_or("Human request has no output directory.")?;
212    let eval = config
213        .evals
214        .iter()
215        .find(|eval| eval.id == request.eval_id)
216        .ok_or("Recorded Human eval no longer exists.")?;
217    let parallelism = cache::Parallelism::new(cache::Parallelism::available());
218    let (code, error) =
219        match cache::recheck(config, eval, output, &parallelism, cancellation.clone()).await {
220            Ok(Some(actual)) if &actual == expected => return Ok(()),
221            Ok(_) => (
222                crate::types::FailureCode::InputChanged,
223                "Fingerprint changed during review.".into(),
224            ),
225            Err(error) if cancellation.is_cancelled() => return Err(error),
226            Err(error) => (crate::types::FailureCode::FingerprintRecheckFailed, error),
227        };
228    request.state =
229        crate::store::RequestState::failed(error.clone(), Some(code), crate::broker::now());
230    receipts.settle_human(request, reviewer).await?;
231    Err(error)
232}
artifactize_core/human.rs

artifactize_core/
human.rs
