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                .map(|backend| serde_json::to_value(backend)
83                    .expect("backend is JSON")
84                    .as_str()
85                    .expect("backend is text")
86                    .to_owned())
87                .unwrap_or_else(|| "?".into()),
88            header.model.as_deref().unwrap_or("?"),
89            header.reasoning.map_or(String::new(), |reasoning| format!(
90                " (reasoning {reasoning})"
91            ))
92        ),
93        files_changed,
94    };
95    let mut recorder = Recorder::append(state, &conversation.path, &located.id, sent.number)?;
96    let answer = agent::follow_up(
97        &config,
98        eval,
99        agent::Continuation {
100            conversation: &conversation,
101            files_changed,
102            output: &output,
103            state,
104        },
105        message,
106        &mut recorder,
107        cancellation,
108    )
109    .await;
110    Ok((sent, answer))
111}
112
113/// Whether an Artifact the review's key covered has another fingerprint now; `None` when
114/// the review had no key or a fingerprint cannot be computed.
115async fn files_changed(
116    config: &crate::config::RepoConfig,
117    request: &store::Request,
118    output: &Path,
119    cancellation: tokio_util::sync::CancellationToken,
120) -> Option<bool> {
121    if request.fingerprints.is_empty() {
122        return None;
123    }
124    if request
125        .fingerprints
126        .keys()
127        .any(|id| !config.artifacts.contains_key(id))
128    {
129        return Some(true);
130    }
131    let current = cache::prepare(
132        config,
133        request.fingerprints.keys(),
134        output,
135        &cache::Parallelism::new(cache::Parallelism::available()),
136        cancellation,
137    )
138    .await
139    .ok()?;
140    Some(
141        request
142            .fingerprints
143            .iter()
144            .any(|(id, fingerprint)| current.get(id).is_none_or(|now| &now.value != fingerprint)),
145    )
146}
artifactize_core/inspect/session/send.rs

artifactize_core/inspect/session/
send.rs
