artifactize_core/inspect/session/
send.rs

1//! Continue saved conversations without changing their recorded verdicts.
2use super::Located;
3use crate::{
4    agent::{
5        self,
6        session::{self, Recorder},
7    },
8    cache,
9    config::read_workspace_config,
10    store, workspace,
11};
12use std::path::Path;
13
14/// Metadata for a recorded follow-up.
15pub struct Sent {
16    pub number: usize,
17    pub model: String,
18    pub files_changed: Option<bool>,
19}
20
21/// Continue the conversation under its lock, so concurrent sends make one thread.
22pub async fn send(
23    state: &Path,
24    repo: Option<&Path>,
25    located: &Located,
26    message: &str,
27    cancellation: tokio_util::sync::CancellationToken,
28) -> Result<(Sent, agent::FollowUp), String> {
29    if matches!(
30        located.request.request.status(),
31        crate::types::RequestStatus::Queued | crate::types::RequestStatus::Running
32    ) {
33        return Err(format!(
34            "Request {} is still running; send a follow-up once its review has finished.",
35            located.request.request.id
36        ));
37    }
38    // A session that was never saved, or is gone, gets no lock file.
39    located.load()?;
40    let _lock = tokio::select! {
41        lock = session::lock(state, &located.id) => lock?,
42        () = cancellation.cancelled() => {
43            return Err("Cancelled while waiting for another send to this session.".into());
44        }
45    };
46    let conversation = located.load()?;
47    let request = &located.request.request;
48    let repo = match repo {
49        Some(repo) => repo.to_path_buf(),
50        None => store::read_run(state, &request.run_id).await?.run.repo_path,
51    };
52    let config = read_workspace_config(&repo).map_err(|error| {
53        format!(
54            "Cannot resolve the review's tools in {}: {error}",
55            crate::platform::path_text(&repo)
56        )
57    })?;
58    let eval = config
59        .evals
60        .iter()
61        .find(|eval| eval.id == request.eval_id)
62        .ok_or_else(|| {
63            format!(
64                "Eval {} is no longer declared in {}; its tools cannot be resolved.",
65                request.eval_id,
66                crate::platform::path_text(&repo)
67            )
68        })?;
69    let output = workspace::prepare_directory(
70        &state.join("runs").join(request.run_id.as_str()),
71        &config.root,
72    )
73    .map_err(|e| e.to_string())?;
74    let files_changed = files_changed(&config, request, &output, cancellation.clone()).await;
75    let header = conversation.header();
76    let sent = Sent {
77        number: conversation.sends() + 1,
78        model: format!(
79            "{} {}{}",
80            header
81                .backend
82                .as_ref()
83                .map_or("?", crate::config::Backend::as_str),
84            header.model.as_deref().unwrap_or("?"),
85            header.reasoning.map_or(String::new(), |reasoning| format!(
86                " (reasoning {reasoning})"
87            ))
88        ),
89        files_changed,
90    };
91    let mut recorder = Recorder::append(state, &conversation.path, &located.id, sent.number)?;
92    let answer = agent::follow_up(
93        &config,
94        eval,
95        agent::Continuation {
96            conversation: &conversation,
97            files_changed,
98            output: &output,
99            state,
100        },
101        message,
102        &mut recorder,
103        cancellation,
104    )
105    .await;
106    Ok((sent, answer))
107}
108
109/// Whether an Artifact the review's key covered has another fingerprint now; `None` when
110/// the review had no key or a fingerprint cannot be computed.
111async fn files_changed(
112    config: &crate::config::RepoConfig,
113    request: &store::Request,
114    output: &Path,
115    cancellation: tokio_util::sync::CancellationToken,
116) -> Option<bool> {
117    if request.fingerprints.is_empty() {
118        return None;
119    }
120    if request
121        .fingerprints
122        .keys()
123        .any(|id| !config.artifacts.contains_key(id))
124    {
125        return Some(true);
126    }
127    let current = cache::prepare(
128        config,
129        request.fingerprints.keys(),
130        output,
131        &cache::Parallelism::new(cache::Parallelism::available()),
132        cancellation,
133    )
134    .await
135    .ok()?;
136    Some(
137        request
138            .fingerprints
139            .iter()
140            .any(|(id, fingerprint)| current.get(id).is_none_or(|now| &now.value != fingerprint)),
141    )
142}
artifactize_core/inspect/session/send.rs

artifactize_core/inspect/session/
send.rs
