speccoder_runtime/hooks/
deliver.rs

1//! Telling observers what speccoder decided, at least once. Events are read
2//! from git (see [`super::events`]) past each observer's cursor, oldest
3//! first; the cursor moves only after the observer took one (2xx, or exit
4//! 0). Whichever speccoder process runs delivers; a run does it beside its
5//! own work and never waits on an observer beyond a short grace at its end.
6
7use std::collections::{BTreeMap, HashMap};
8use std::path::{Path, PathBuf};
9use std::sync::{Mutex, OnceLock, mpsc};
10use std::time::{Duration, Instant};
11
12use serde::{Deserialize, Serialize};
13use serde_json::Value;
14use speccoder_core::hooks::{HookDef, Runner, TrustState};
15
16use super::HookRun;
17use super::events::{self, Event};
18use crate::git::Git;
19use crate::verify::process::{Invocation, OUTPUT_CAP};
20
21const CURSORS: &str = "cursors.json";
22const LOCK: &str = "deliver.lock";
23
24/// Attempts a process makes at one event before leaving it to the next.
25const TRIES: u32 = 3;
26
27/// How far an observer got: the tips it has heard up to, and the saved
28/// attempts it heard.
29#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
30pub struct Cursor {
31    /// Branch → the last commit delivered on it (or its tip when the
32    /// observer started).
33    pub tips: BTreeMap<String, String>,
34    /// `attempt:<commit>` ids delivered (the last 2,000).
35    pub attempts: Vec<String>,
36}
37
38fn cursors(dir: &Path) -> BTreeMap<String, Cursor> {
39    std::fs::read_to_string(dir.join(CURSORS))
40        .ok()
41        .and_then(|t| serde_json::from_str(&t).ok())
42        .unwrap_or_default()
43}
44
45/// A cursor at what exists now: an observer hears what happens from here.
46#[must_use]
47pub fn here(root: &Path) -> Cursor {
48    let mut c = Cursor::default();
49    for (r, sha) in events::tips(root) {
50        if r.starts_with(crate::attempts::PREFIX) {
51            c.attempts.push(format!("attempt:{sha}"));
52        } else {
53            c.tips.insert(r, sha);
54        }
55    }
56    c
57}
58
59/// Whether `name` has a cursor yet.
60#[must_use]
61pub fn has_cursor(dir: &Path, name: &str) -> bool {
62    cursors(dir).contains_key(name)
63}
64
65/// Starts `name`'s cursor at what exists now (a newly trusted observer).
66pub fn start_here(dir: &Path, root: &Path, name: &str) {
67    // Under the delivery lock: a pass running now would otherwise write
68    // back the cursors it read before this one existed.
69    let _ = std::fs::create_dir_all(dir);
70    let lock = std::fs::File::create(dir.join(LOCK));
71    let _held = lock.as_ref().map(std::fs::File::lock);
72    let mut all = cursors(dir);
73    all.insert(name.to_owned(), here(root));
74    super::write_json(dir, CURSORS, &all);
75}
76
77/// The events past `cursor`, oldest first per branch, then the stops.
78#[must_use]
79pub fn pending(root: &Path, project: &str, cursor: &Cursor) -> Vec<(String, Event)> {
80    let git = Git::at(root);
81    // A tip that no longer exists (a branch reset, then collected) is
82    // dropped: delivery goes on from what is there, inventing nothing.
83    let seen: Vec<&str> = cursor
84        .tips
85        .values()
86        .map(String::as_str)
87        .filter(|sha| {
88            git.succeeds(&["cat-file", "-e", &format!("{sha}^{{commit}}")])
89                .unwrap_or(false)
90        })
91        .collect();
92    let mut out = Vec::new();
93    let tips = events::tips(root);
94    for (r, sha) in tips
95        .iter()
96        .filter(|(r, _)| !r.starts_with(crate::attempts::PREFIX))
97    {
98        if cursor.tips.get(r) == Some(sha) {
99            continue;
100        }
101        let mut args = vec![
102            "log".to_owned(),
103            "--reverse".to_owned(),
104            "--format=%H".to_owned(),
105            "--grep=^Verdict: ".to_owned(),
106            sha.clone(),
107        ];
108        if !seen.is_empty() {
109            args.push("--not".to_owned());
110            args.extend(seen.iter().map(|s| (*s).to_owned()));
111        }
112        let args: Vec<&str> = args.iter().map(String::as_str).collect();
113        let commits = git
114            .run(&args)
115            .map(|b| String::from_utf8_lossy(&b).into_owned())
116            .unwrap_or_default();
117        for c in commits.lines().filter(|c| !c.is_empty()) {
118            if let Some(e) = events::of(&git, project, r, c) {
119                out.push((r.clone(), e));
120            }
121        }
122    }
123    for (r, sha) in tips
124        .iter()
125        .filter(|(r, _)| r.starts_with(crate::attempts::PREFIX))
126    {
127        let id = format!("attempt:{sha}");
128        if !cursor.attempts.contains(&id)
129            && let Some(e) = events::of(&git, project, r, sha)
130        {
131            out.push((r.clone(), e));
132        }
133    }
134    out
135}
136
137fn advance(cursor: &mut Cursor, reference: &str, event: &Event) {
138    if let Some(commit) = event.id.strip_prefix("tick:") {
139        cursor.tips.insert(reference.to_owned(), commit.to_owned());
140    } else {
141        cursor.attempts.push(event.id.clone());
142    }
143}
144
145/// Sends `envelope` to one observer: `Ok` with what it answered, or why not.
146///
147/// # Errors
148///
149/// The observer refused it, failed, timed out or could not be reached.
150pub fn send(def: &HookDef, envelope: &Value, root: &Path, state: &Path) -> Result<String, String> {
151    match &def.runner {
152        Runner::Url { url, secret_env } => post(def, url, secret_env.as_deref(), envelope),
153        Runner::Argv(argv) => {
154            // An extension's `./` paths are inside its folder.
155            let folder = def.extension.as_ref().map(|(_, f)| Path::new(f));
156            let argv = crate::extensions::resolve(argv, folder, &[]);
157            command(def, &argv, envelope, root, state)
158        }
159    }
160}
161
162fn post(
163    def: &HookDef,
164    url: &str,
165    secret_env: Option<&str>,
166    envelope: &Value,
167) -> Result<String, String> {
168    use agentplane::push::{
169        Delivered, Destination, PushConfig, PushMessage, PushSender, PushTransport,
170    };
171    speccoder_core::hooks::url_allowed(url)?;
172    let event = agentplane::core::CloudEvent::from_json(envelope.to_string().as_bytes())
173        .map_err(|e| format!("the event is not a CloudEvent: {e}"))?;
174    let mut destination = Destination::new(&def.name, url);
175    if let Some(var) = secret_env {
176        let key = std::env::var(var).map_err(|_| format!("${var} is not set"))?;
177        destination = destination
178            .try_signed_with(&agentplane::core::Secret::new(key))
179            .map_err(|e| format!("${var}: {e}"))?;
180    }
181    let sender = PushSender::for_operator_destinations(std::slice::from_ref(&destination))
182        .timeout(Duration::from_secs(def.timeout));
183    let config = PushConfig {
184        id: destination.registration_id(),
185        task: agentplane::core::RunId::generate(),
186        url: url.to_owned(),
187        token: None,
188        authentication: None,
189    };
190    let at = std::time::SystemTime::now()
191        .duration_since(std::time::UNIX_EPOCH)
192        .map_or(0, |d| d.as_secs());
193    let rt = crate::pipeline::tokio_runtime()?;
194    match rt.block_on(sender.deliver(&config, &PushMessage::cloudevent(&event), at)) {
195        Ok(Delivered::Accepted) => Ok("accepted".to_owned()),
196        Ok(Delivered::Rejected { status, .. }) => Err(format!("answered {status}")),
197        Ok(Delivered::Unreachable(why)) => Err(why),
198        Err(e) => Err(e.to_string()),
199    }
200}
201
202fn command(
203    def: &HookDef,
204    argv: &[String],
205    envelope: &Value,
206    root: &Path,
207    state: &Path,
208) -> Result<String, String> {
209    let (runner, _) = crate::sandbox::runner_at(root)?;
210    if !runner.is_sandboxed() {
211        return Err("no sandbox is enforced here, and an observer runs only in one".to_owned());
212    }
213    // The checkout is the observer's to read, not to change.
214    let (runner, _) = runner.carve(
215        &[root.to_path_buf()],
216        &super::gate::hidden_state(state, root),
217    );
218    if !runner.hides() {
219        return Err("no sandbox is enforced here that hides speccoder's state, and an observer runs only in one".to_owned());
220    }
221    let scratch = runner
222        .scratch()
223        .ok_or_else(|| "the sandbox has no scratch directory".to_owned())?;
224    let file = scratch.join(format!("event-{}.json", std::process::id()));
225    std::fs::write(&file, format!("{envelope}\n")).map_err(|e| e.to_string())?;
226    let mut env = crate::verify::process::scrubbed_env(std::env::vars());
227    for name in &def.env {
228        if let Ok(v) = std::env::var(name) {
229            env.retain(|(k, _)| k != name);
230            env.push((name.clone(), v));
231        }
232    }
233    env.push((
234        "SPECCODER_EVENT".to_owned(),
235        envelope["type"]
236            .as_str()
237            .unwrap_or_default()
238            .trim_start_matches("dev.speccoder.")
239            .to_owned(),
240    ));
241    // The event on stdin: the shell's redirect, then the hook's own argv.
242    let mut full = vec![
243        "sh".to_owned(),
244        "-c".to_owned(),
245        "exec \"$@\" < \"$0\"".to_owned(),
246        file.to_string_lossy().into_owned(),
247    ];
248    full.extend(argv.iter().cloned());
249    let inv = Invocation {
250        argv: full,
251        cwd: root.to_path_buf(),
252        env,
253        timeout: Duration::from_secs(def.timeout),
254        output_cap: OUTPUT_CAP,
255    };
256    let ended = runner.run_with(&inv, true);
257    let _ = std::fs::remove_file(&file);
258    if let Some(e) = ended.spawn_error {
259        return Err(format!("could not start: {e}"));
260    }
261    if ended.timed_out {
262        return Err(format!("timed out after {} s", def.timeout));
263    }
264    match ended.exit_code {
265        Some(0) => Ok("exit 0".to_owned()),
266        code => {
267            let stderr = String::from_utf8_lossy(&ended.stderr.kept);
268            let line = stderr.lines().map(str::trim).rfind(|l| !l.is_empty());
269            Err(line.map_or_else(
270                || code.map_or_else(|| "was killed".to_owned(), |c| format!("exit {c}")),
271                speccoder_core::hooks::clean,
272            ))
273        }
274    }
275}
276
277/// What one delivery pass did.
278#[derive(Debug, Clone, Default, PartialEq, Eq)]
279pub struct Delivery {
280    /// Events delivered.
281    pub delivered: u32,
282    /// Events still waiting (an observer failed on them).
283    pub pending: u32,
284    /// Observers not trusted, skipped.
285    pub untrusted: Vec<String>,
286    /// `(observer, why)` for each that failed.
287    pub failed: Vec<(String, String)>,
288    /// Another process was delivering: nothing was done.
289    pub busy: bool,
290}
291
292/// Failures per `observer/event` in this process: after [`TRIES`] the
293/// event waits for the next process.
294fn tries() -> &'static Mutex<HashMap<String, u32>> {
295    static TRIED: OnceLock<Mutex<HashMap<String, u32>>> = OnceLock::new();
296    TRIED.get_or_init(Mutex::default)
297}
298
299/// Delivers what is pending for every trusted observer of the project at
300/// `root`, until `deadline` (if given).
301#[must_use]
302pub fn pass(root: &Path, state: &Path, deadline: Option<Instant>) -> Delivery {
303    let mut out = Delivery::default();
304    let Ok(hooks) = super::load(root) else {
305        return out;
306    };
307    let observers: Vec<&HookDef> = hooks.iter().filter(|h| h.at().is_none()).collect();
308    if observers.is_empty() {
309        return out;
310    }
311    let dir = super::dir(state, root);
312    let _ = std::fs::create_dir_all(&dir);
313    let Ok(lock) = std::fs::File::create(dir.join(LOCK)) else {
314        return out;
315    };
316    if lock.try_lock().is_err() {
317        out.busy = true;
318        return out;
319    }
320    let project = super::project_id(state, root);
321    let mut all = cursors(&dir);
322    for def in observers {
323        if super::digest(def, root).trust(&dir) != TrustState::Trusted {
324            out.untrusted.push(def.name.clone());
325            continue;
326        }
327        let Some(cursor) = all.get_mut(&def.name) else {
328            // Trusted without a cursor (from CI's flag): from here on.
329            all.insert(def.name.clone(), here(root));
330            continue;
331        };
332        // Saved attempts whose refs are gone need no memory.
333        let present: std::collections::HashSet<String> = events::tips(root)
334            .into_iter()
335            .filter(|(r, _)| r.starts_with(crate::attempts::PREFIX))
336            .map(|(_, sha)| format!("attempt:{sha}"))
337            .collect();
338        cursor.attempts.retain(|id| present.contains(id));
339        for (reference, event) in pending(root, &project, cursor) {
340            if !def.hears(event.kind) {
341                advance(cursor, &reference, &event);
342                continue;
343            }
344            // In order: what follows waits behind an event not sent now.
345            if deadline.is_some_and(|d| Instant::now() >= d) {
346                out.pending += 1;
347                break;
348            }
349            let key = format!("{}/{}", def.name, event.id);
350            let tried = tries()
351                .lock()
352                .map_or(0, |t| t.get(&key).copied().unwrap_or(0));
353            if tried >= TRIES {
354                out.pending += 1;
355                break;
356            }
357            let began = Instant::now();
358            let sent = send(def, &event.envelope, root, state);
359            let duration_ms = u64::try_from(began.elapsed().as_millis()).unwrap_or(u64::MAX);
360            super::record_run(
361                &dir,
362                &def.name,
363                HookRun {
364                    at: crate::attest::now_rfc3339(),
365                    what: event.id.clone(),
366                    outcome: if sent.is_ok() { "delivered" } else { "failed" }.to_owned(),
367                    duration_ms,
368                    error: sent.as_ref().err().cloned(),
369                    findings: Vec::new(),
370                    findings_error: None,
371                },
372            );
373            match sent {
374                Ok(_) => {
375                    advance(cursor, &reference, &event);
376                    out.delivered += 1;
377                }
378                Err(e) => {
379                    if let Ok(mut t) = tries().lock() {
380                        *t.entry(key).or_default() += 1;
381                    }
382                    out.failed.push((def.name.clone(), e));
383                    out.pending += 1;
384                    // In order: the rest wait behind the one that failed.
385                    break;
386                }
387            }
388        }
389    }
390    super::write_json(&dir, CURSORS, &all);
391    out
392}
393
394/// A run's deliverer: delivers beside the run; dropped, it gives what is
395/// pending at most [`GRACE`]. One per process — a feature run's tasks share
396/// their run's.
397#[derive(Debug)]
398pub struct Deliverer {
399    stop: mpsc::Sender<()>,
400    done: mpsc::Receiver<()>,
401}
402
403/// How long a run waits for its last deliveries.
404pub const GRACE: Duration = Duration::from_secs(5);
405
406static ACTIVE: std::sync::atomic::AtomicBool = std::sync::atomic::AtomicBool::new(false);
407
408impl Deliverer {
409    /// Starts delivering for the project at `root`, when it has observers
410    /// and no deliverer runs in this process yet.
411    #[must_use]
412    pub fn start(root: &Path, state: &Path) -> Option<Self> {
413        use std::sync::atomic::Ordering;
414        let has_observers = super::load(root)
415            .map(|h| h.iter().any(|d| d.at().is_none()))
416            .unwrap_or(false);
417        if !has_observers || ACTIVE.swap(true, Ordering::SeqCst) {
418            return None;
419        }
420        let (stop, stopped) = mpsc::channel::<()>();
421        let (finished, done) = mpsc::channel::<()>();
422        let (root, state): (PathBuf, PathBuf) = (root.to_path_buf(), state.to_path_buf());
423        std::thread::spawn(move || {
424            loop {
425                let _ = pass(&root, &state, None);
426                match stopped.recv_timeout(Duration::from_secs(1)) {
427                    Err(mpsc::RecvTimeoutError::Timeout) => {}
428                    Ok(()) | Err(mpsc::RecvTimeoutError::Disconnected) => break,
429                }
430            }
431            let _ = pass(&root, &state, Some(Instant::now() + GRACE));
432            let _ = finished.send(());
433        });
434        Some(Self { stop, done })
435    }
436}
437
438impl Drop for Deliverer {
439    fn drop(&mut self) {
440        let _ = self.stop.send(());
441        let _ = self.done.recv_timeout(GRACE + Duration::from_secs(1));
442        ACTIVE.store(false, std::sync::atomic::Ordering::SeqCst);
443    }
444}
445
446#[cfg(test)]
447mod tests {
448    use super::*;
449
450    #[test]
451    fn a_cursor_moves_per_event() {
452        let mut c = Cursor::default();
453        let tick = Event {
454            id: "tick:abc".into(),
455            kind: "task.ticked",
456            envelope: Value::Null,
457        };
458        advance(&mut c, "refs/heads/speccoder/x", &tick);
459        assert_eq!(c.tips["refs/heads/speccoder/x"], "abc");
460        let stop = Event {
461            id: "attempt:def".into(),
462            kind: "task.stopped",
463            envelope: Value::Null,
464        };
465        advance(&mut c, "refs/speccoder/attempts/x/T1/r", &stop);
466        assert_eq!(c.attempts, ["attempt:def"]);
467    }
468}
speccoder_runtime/hooks/deliver.rs

speccoder_runtime/hooks/
deliver.rs
