a3s_code_core/
external_observation.rs

1//! Typed external observations are run inputs. Clearing one requires a newer
2//! observation of the same subject or a LOOP-1 waiver, not a sentence.
3//!
4//! Concurrent writes to a path another session has dirtied fail closed.
5
6use crate::harness_loop::CompletionWaiverV1;
7use serde::{Deserialize, Serialize};
8use serde_json::Value;
9use std::collections::HashMap;
10use std::path::{Component, Path, PathBuf};
11use std::sync::{Mutex, OnceLock};
12
13pub const EXTERNAL_OBSERVATION_SCHEMA: &str = "a3s.code.external-observation.v1";
14
15#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
16#[serde(rename_all = "snake_case")]
17pub enum RequiredAction {
18    None,
19    WorkspaceChange,
20    Address,
21}
22
23#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
24pub struct ExternalObservationV1 {
25    pub schema: String,
26    pub kind: String,
27    pub subject: String,
28    pub digest: String,
29    pub payload: String,
30    pub source_revision: Option<String>,
31    pub required_action: RequiredAction,
32}
33
34impl ExternalObservationV1 {
35    pub fn new(
36        kind: impl Into<String>,
37        subject: impl Into<String>,
38        digest: impl Into<String>,
39        payload: impl Into<String>,
40        required_action: RequiredAction,
41    ) -> Option<Self> {
42        let digest = digest.into();
43        let payload = payload.into();
44        if digest.trim().is_empty() || payload.len() > 8_192 {
45            return None;
46        }
47        Some(Self {
48            schema: EXTERNAL_OBSERVATION_SCHEMA.to_string(),
49            kind: kind.into(),
50            subject: subject.into(),
51            digest,
52            payload,
53            source_revision: None,
54            required_action,
55        })
56    }
57
58    pub fn with_revision(mut self, revision: impl Into<String>) -> Self {
59        self.source_revision = Some(revision.into());
60        self
61    }
62
63    pub fn model_input_fragment(&self) -> String {
64        format!(
65            "[external observation schema={} kind={} subject={} digest={} action={:?}]",
66            self.schema, self.kind, self.subject, self.digest, self.required_action
67        )
68    }
69}
70
71pub fn still_open(
72    observations: &[ExternalObservationV1],
73    waivers: &[CompletionWaiverV1],
74    effect_digest: &str,
75) -> Vec<ExternalObservationV1> {
76    let mut newest: HashMap<String, &ExternalObservationV1> = HashMap::new();
77    for observation in observations {
78        newest.insert(observation.subject.clone(), observation);
79    }
80    newest
81        .into_values()
82        .filter(|observation| observation.required_action != RequiredAction::None)
83        .filter(|observation| {
84            !waivers.iter().any(|waiver| {
85                waiver.effect_digest == observation.digest
86                    || (!effect_digest.is_empty() && waiver.effect_digest == effect_digest)
87            })
88        })
89        .cloned()
90        .collect()
91}
92
93pub fn blocks_success(open: &[ExternalObservationV1]) -> bool {
94    open.iter()
95        .any(|observation| observation.required_action == RequiredAction::WorkspaceChange)
96}
97
98fn dirty() -> &'static Mutex<HashMap<PathBuf, String>> {
99    static DIRTY: OnceLock<Mutex<HashMap<PathBuf, String>>> = OnceLock::new();
100    DIRTY.get_or_init(|| Mutex::new(HashMap::new()))
101}
102
103/// Refuse a missing session instead of sharing an `"anonymous"` owner.
104pub fn claim_bound_write(
105    session_id: Option<&str>,
106    workspace: &Path,
107    relative: &str,
108) -> Result<(), String> {
109    let Some(session_id) = session_id.filter(|id| !id.trim().is_empty()) else {
110        return Err("write requires a session id".to_string());
111    };
112    claim_write(session_id, workspace, relative)
113}
114
115/// One dirty-path identity for `guest.txt`, `./guest.txt`, and
116/// `subdir/../guest.txt`. Joining the raw spelling lets a second session
117/// hide a claim by respelling the same file.
118fn claim_identity(workspace: &Path, relative: &str) -> Result<PathBuf, String> {
119    let root = canonical_root(workspace);
120    let raw = Path::new(relative);
121    let under_root = if raw.is_absolute() {
122        raw.strip_prefix(&root)
123            .or_else(|_| raw.strip_prefix(workspace))
124            .map(Path::to_path_buf)
125            .map_err(|_| "write path must stay inside the workspace".to_string())?
126    } else {
127        raw.to_path_buf()
128    };
129    let mut parts = Vec::new();
130    for component in under_root.components() {
131        match component {
132            Component::Normal(part) => parts.push(part.to_os_string()),
133            Component::CurDir => {}
134            Component::ParentDir => {
135                if parts.pop().is_none() {
136                    return Err("write path must stay inside the workspace".to_string());
137                }
138            }
139            Component::RootDir | Component::Prefix(_) => {
140                return Err("write path must stay inside the workspace".to_string());
141            }
142        }
143    }
144    if parts.is_empty() {
145        return Err("write path must name a file".to_string());
146    }
147    Ok(parts.into_iter().fold(root, |path, part| path.join(part)))
148}
149
150fn canonical_root(workspace: &Path) -> PathBuf {
151    workspace
152        .canonicalize()
153        .unwrap_or_else(|_| workspace.to_path_buf())
154}
155
156pub fn claim_write(session_id: &str, workspace: &Path, relative: &str) -> Result<(), String> {
157    let path = claim_identity(workspace, relative)?;
158    let mut guard = dirty().lock().expect("dirty paths");
159    if let Some(owner) = guard.get(&path) {
160        if owner != session_id {
161            return Err(format!(
162                "concurrent write to {} is already owned by session {owner}",
163                path.display()
164            ));
165        }
166        return Ok(());
167    }
168    guard.insert(path, session_id.to_string());
169    Ok(())
170}
171
172/// Refuse a workspace-wide overwrite when another session already owns a path
173/// here. Checkout and stash do not name every file they will touch.
174pub fn foreign_workspace_claim(session_id: Option<&str>, workspace: &Path) -> Result<(), String> {
175    let Some(session_id) = session_id.filter(|id| !id.trim().is_empty()) else {
176        return Err("write requires a session id".to_string());
177    };
178    refuse_foreign_workspace_owner(Some(session_id), workspace)
179}
180
181/// Refuse when a different session already owns a dirty path in this workspace.
182///
183/// A missing session is not an owner, so any existing claim is foreign. An
184/// empty claim map is not a refusal: callers that still allow an unbound
185/// command (bash) must not treat "no session" as "no owner anywhere".
186pub fn refuse_foreign_workspace_owner(
187    session_id: Option<&str>,
188    workspace: &Path,
189) -> Result<(), String> {
190    let session_id = session_id.unwrap_or("").trim();
191    let workspace = canonical_root(workspace);
192    let guard = dirty().lock().expect("dirty paths");
193    if let Some((path, owner)) = guard.iter().find(|(path, owner)| {
194        !owner.is_empty() && owner.as_str() != session_id && path.starts_with(&workspace)
195    }) {
196        return Err(format!(
197            "concurrent write to {} is already owned by session {owner}",
198            path.display()
199        ));
200    }
201    Ok(())
202}
203
204pub fn session_owns_write(session_id: &str, workspace: &Path, relative: &str) -> bool {
205    let Ok(path) = claim_identity(workspace, relative) else {
206        return false;
207    };
208    dirty()
209        .lock()
210        .expect("dirty paths")
211        .get(&path)
212        .is_some_and(|owner| owner == session_id)
213}
214
215/// Drop a claim this session holds, and no other session's claim.
216/// Move a finished child's claim to the parent, or claim an unowned dirty
217/// path for the parent. A path another live session already owns is left
218/// alone.
219pub fn adopt_write_claim(
220    from_session: &str,
221    to_session: &str,
222    workspace: &Path,
223    relative: &str,
224) -> Result<(), String> {
225    let to_session = to_session.trim();
226    if to_session.is_empty() {
227        return Err("write requires a session id".to_string());
228    }
229    let path = claim_identity(workspace, relative)?;
230    let mut guard = dirty().lock().expect("dirty paths");
231    match guard.get(&path).map(String::as_str) {
232        Some(owner) if owner == to_session || owner == from_session => {
233            guard.insert(path, to_session.to_string());
234            Ok(())
235        }
236        Some(owner) => Err(format!(
237            "concurrent write to {} is already owned by session {owner}",
238            path.display()
239        )),
240        None => {
241            guard.insert(path, to_session.to_string());
242            Ok(())
243        }
244    }
245}
246
247pub fn release_write_claim(session_id: &str, workspace: &Path, relative: &str) {
248    let Ok(path) = claim_identity(workspace, relative) else {
249        return;
250    };
251    let mut guard = dirty().lock().expect("dirty paths");
252    if guard.get(&path).is_some_and(|owner| owner == session_id) {
253        guard.remove(&path);
254    }
255}
256
257pub fn release_session(session_id: &str) {
258    dirty()
259        .lock()
260        .expect("dirty paths")
261        .retain(|_, owner| owner != session_id);
262}
263
264pub fn observation_from_value(value: &Value) -> Option<ExternalObservationV1> {
265    serde_json::from_value(value.clone()).ok()
266}
267
268#[cfg(test)]
269mod tests {
270    use super::*;
271
272    #[test]
273    fn digest_is_on_the_model_input_fragment() {
274        let observation = ExternalObservationV1::new(
275            "ci",
276            "build",
277            "obs-1",
278            "job failed",
279            RequiredAction::WorkspaceChange,
280        )
281        .unwrap();
282        let fragment = observation.model_input_fragment();
283        assert!(fragment.contains("obs-1"));
284        assert!(fragment.contains(EXTERNAL_OBSERVATION_SCHEMA));
285    }
286
287    #[test]
288    fn final_answer_does_not_clear_a_required_workspace_change() {
289        let observation = ExternalObservationV1::new(
290            "ci",
291            "build",
292            "obs-1",
293            "job failed",
294            RequiredAction::WorkspaceChange,
295        )
296        .unwrap();
297        let open = still_open(&[observation], &[], "");
298        assert!(blocks_success(&open));
299        let waiver = CompletionWaiverV1::new("obs-1", "host accepted").unwrap();
300        let cleared = still_open(&open, &[waiver], "");
301        assert!(!blocks_success(&cleared));
302    }
303
304    #[test]
305    fn second_session_cannot_silently_win_a_dirty_path() {
306        let root = tempfile::tempdir().unwrap();
307        claim_write("one", root.path(), "src/lib.rs").unwrap();
308        let error = claim_write("two", root.path(), "src/lib.rs").unwrap_err();
309        assert!(error.contains("already owned"));
310        let missing = claim_bound_write(None, root.path(), "src/other.rs").unwrap_err();
311        assert!(missing.contains("session id"));
312        let blocked = foreign_workspace_claim(Some("two"), root.path()).unwrap_err();
313        assert!(blocked.contains("already owned"));
314        claim_write("two", root.path(), "src/other.rs").unwrap();
315        claim_write("one", root.path(), "src/lib.rs").unwrap();
316        release_session("one");
317        release_session("two");
318    }
319
320    #[test]
321    fn adopt_moves_a_child_claim_to_the_parent_and_does_not_steal() {
322        let root = tempfile::tempdir().unwrap();
323        claim_write("task-run-child", root.path(), "guest.txt").unwrap();
324        claim_write("other-live", root.path(), "held.txt").unwrap();
325
326        adopt_write_claim("task-run-child", "task-parent", root.path(), "guest.txt").unwrap();
327        let stolen = adopt_write_claim("task-run-child", "task-parent", root.path(), "held.txt");
328
329        assert!(session_owns_write("task-parent", root.path(), "guest.txt"));
330        assert!(!session_owns_write(
331            "task-run-child",
332            root.path(),
333            "guest.txt"
334        ));
335        assert!(stolen.unwrap_err().contains("already owned"));
336        assert!(session_owns_write("other-live", root.path(), "held.txt"));
337        release_session("task-parent");
338        release_session("task-run-child");
339        release_session("other-live");
340    }
341
342    #[test]
343    fn newer_observation_of_the_same_subject_replaces_the_open_action() {
344        let first = ExternalObservationV1::new(
345            "ci",
346            "build",
347            "obs-1",
348            "failed",
349            RequiredAction::WorkspaceChange,
350        )
351        .unwrap();
352        let second =
353            ExternalObservationV1::new("ci", "build", "obs-2", "green", RequiredAction::None)
354                .unwrap();
355        let open = still_open(&[first, second], &[], "");
356        assert!(!blocks_success(&open));
357    }
358}
a3s_code_core/external_observation.rs

a3s_code_core/
external_observation.rs
