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) => command(def, argv, envelope, root, state),
154    }
155}
156
157fn post(
158    def: &HookDef,
159    url: &str,
160    secret_env: Option<&str>,
161    envelope: &Value,
162) -> Result<String, String> {
163    use agentplane::push::{
164        Delivered, Destination, PushConfig, PushMessage, PushSender, PushTransport,
165    };
166    speccoder_core::hooks::url_allowed(url)?;
167    let event = agentplane::core::CloudEvent::from_json(envelope.to_string().as_bytes())
168        .map_err(|e| format!("the event is not a CloudEvent: {e}"))?;
169    let mut destination = Destination::new(&def.name, url);
170    if let Some(var) = secret_env {
171        let key = std::env::var(var).map_err(|_| format!("${var} is not set"))?;
172        destination = destination
173            .try_signed_with(&agentplane::core::Secret::new(key))
174            .map_err(|e| format!("${var}: {e}"))?;
175    }
176    let sender = PushSender::for_operator_destinations(std::slice::from_ref(&destination))
177        .timeout(Duration::from_secs(def.timeout));
178    let config = PushConfig {
179        id: destination.registration_id(),
180        task: agentplane::core::RunId::generate(),
181        url: url.to_owned(),
182        token: None,
183        authentication: None,
184    };
185    let at = std::time::SystemTime::now()
186        .duration_since(std::time::UNIX_EPOCH)
187        .map_or(0, |d| d.as_secs());
188    let rt = crate::pipeline::tokio_runtime()?;
189    match rt.block_on(sender.deliver(&config, &PushMessage::cloudevent(&event), at)) {
190        Ok(Delivered::Accepted) => Ok("accepted".to_owned()),
191        Ok(Delivered::Rejected { status, .. }) => Err(format!("answered {status}")),
192        Ok(Delivered::Unreachable(why)) => Err(why),
193        Err(e) => Err(e.to_string()),
194    }
195}
196
197fn command(
198    def: &HookDef,
199    argv: &[String],
200    envelope: &Value,
201    root: &Path,
202    state: &Path,
203) -> Result<String, String> {
204    let (runner, _) = crate::sandbox::runner_at(root)?;
205    if !runner.is_sandboxed() {
206        return Err("no sandbox is enforced here, and an observer runs only in one".to_owned());
207    }
208    // The checkout is the observer's to read, not to change.
209    let (runner, _) = runner.carve(
210        &[root.to_path_buf()],
211        &super::gate::hidden_state(state, root),
212    );
213    if !runner.hides() {
214        return Err("no sandbox is enforced here that hides speccoder's state, and an observer runs only in one".to_owned());
215    }
216    let scratch = runner
217        .scratch()
218        .ok_or_else(|| "the sandbox has no scratch directory".to_owned())?;
219    let file = scratch.join(format!("event-{}.json", std::process::id()));
220    std::fs::write(&file, format!("{envelope}\n")).map_err(|e| e.to_string())?;
221    let mut env = crate::verify::process::scrubbed_env(std::env::vars());
222    for name in &def.env {
223        if let Ok(v) = std::env::var(name) {
224            env.retain(|(k, _)| k != name);
225            env.push((name.clone(), v));
226        }
227    }
228    env.push((
229        "SPECCODER_EVENT".to_owned(),
230        envelope["type"]
231            .as_str()
232            .unwrap_or_default()
233            .trim_start_matches("dev.speccoder.")
234            .to_owned(),
235    ));
236    // The event on stdin: the shell's redirect, then the hook's own argv.
237    let mut full = vec![
238        "sh".to_owned(),
239        "-c".to_owned(),
240        "exec \"$@\" < \"$0\"".to_owned(),
241        file.to_string_lossy().into_owned(),
242    ];
243    full.extend(argv.iter().cloned());
244    let inv = Invocation {
245        argv: full,
246        cwd: root.to_path_buf(),
247        env,
248        timeout: Duration::from_secs(def.timeout),
249        output_cap: OUTPUT_CAP,
250    };
251    let ended = runner.run_with(&inv, true);
252    let _ = std::fs::remove_file(&file);
253    if let Some(e) = ended.spawn_error {
254        return Err(format!("could not start: {e}"));
255    }
256    if ended.timed_out {
257        return Err(format!("timed out after {} s", def.timeout));
258    }
259    match ended.exit_code {
260        Some(0) => Ok("exit 0".to_owned()),
261        code => {
262            let stderr = String::from_utf8_lossy(&ended.stderr.kept);
263            let line = stderr.lines().map(str::trim).rfind(|l| !l.is_empty());
264            Err(line.map_or_else(
265                || code.map_or_else(|| "was killed".to_owned(), |c| format!("exit {c}")),
266                speccoder_core::hooks::clean,
267            ))
268        }
269    }
270}
271
272/// What one delivery pass did.
273#[derive(Debug, Clone, Default, PartialEq, Eq)]
274pub struct Delivery {
275    /// Events delivered.
276    pub delivered: u32,
277    /// Events still waiting (an observer failed on them).
278    pub pending: u32,
279    /// Observers not trusted, skipped.
280    pub untrusted: Vec<String>,
281    /// `(observer, why)` for each that failed.
282    pub failed: Vec<(String, String)>,
283    /// Another process was delivering: nothing was done.
284    pub busy: bool,
285}
286
287/// Failures per `observer/event` in this process: after [`TRIES`] the
288/// event waits for the next process.
289fn tries() -> &'static Mutex<HashMap<String, u32>> {
290    static TRIED: OnceLock<Mutex<HashMap<String, u32>>> = OnceLock::new();
291    TRIED.get_or_init(Mutex::default)
292}
293
294/// Delivers what is pending for every trusted observer of the project at
295/// `root`, until `deadline` (if given).
296#[must_use]
297pub fn pass(root: &Path, state: &Path, deadline: Option<Instant>) -> Delivery {
298    let mut out = Delivery::default();
299    let Ok(hooks) = super::load(root) else {
300        return out;
301    };
302    let observers: Vec<&HookDef> = hooks.iter().filter(|h| h.at().is_none()).collect();
303    if observers.is_empty() {
304        return out;
305    }
306    let dir = super::dir(state, root);
307    let _ = std::fs::create_dir_all(&dir);
308    let Ok(lock) = std::fs::File::create(dir.join(LOCK)) else {
309        return out;
310    };
311    if lock.try_lock().is_err() {
312        out.busy = true;
313        return out;
314    }
315    let project = super::project_id(state, root);
316    let mut all = cursors(&dir);
317    for def in observers {
318        let digest = super::digest(def, root).digest;
319        if super::trust::state(&dir, &def.name, &digest) != TrustState::Trusted {
320            out.untrusted.push(def.name.clone());
321            continue;
322        }
323        let Some(cursor) = all.get_mut(&def.name) else {
324            // Trusted without a cursor (from CI's flag): from here on.
325            all.insert(def.name.clone(), here(root));
326            continue;
327        };
328        // Saved attempts whose refs are gone need no memory.
329        let present: std::collections::HashSet<String> = events::tips(root)
330            .into_iter()
331            .filter(|(r, _)| r.starts_with(crate::attempts::PREFIX))
332            .map(|(_, sha)| format!("attempt:{sha}"))
333            .collect();
334        cursor.attempts.retain(|id| present.contains(id));
335        for (reference, event) in pending(root, &project, cursor) {
336            if !def.hears(event.kind) {
337                advance(cursor, &reference, &event);
338                continue;
339            }
340            // In order: what follows waits behind an event not sent now.
341            if deadline.is_some_and(|d| Instant::now() >= d) {
342                out.pending += 1;
343                break;
344            }
345            let key = format!("{}/{}", def.name, event.id);
346            let tried = tries()
347                .lock()
348                .map_or(0, |t| t.get(&key).copied().unwrap_or(0));
349            if tried >= TRIES {
350                out.pending += 1;
351                break;
352            }
353            let began = Instant::now();
354            let sent = send(def, &event.envelope, root, state);
355            let duration_ms = u64::try_from(began.elapsed().as_millis()).unwrap_or(u64::MAX);
356            super::record_run(
357                &dir,
358                &def.name,
359                HookRun {
360                    at: crate::attest::now_rfc3339(),
361                    what: event.id.clone(),
362                    outcome: if sent.is_ok() { "delivered" } else { "failed" }.to_owned(),
363                    duration_ms,
364                    error: sent.as_ref().err().cloned(),
365                },
366            );
367            match sent {
368                Ok(_) => {
369                    advance(cursor, &reference, &event);
370                    out.delivered += 1;
371                }
372                Err(e) => {
373                    if let Ok(mut t) = tries().lock() {
374                        *t.entry(key).or_default() += 1;
375                    }
376                    out.failed.push((def.name.clone(), e));
377                    out.pending += 1;
378                    // In order: the rest wait behind the one that failed.
379                    break;
380                }
381            }
382        }
383    }
384    super::write_json(&dir, CURSORS, &all);
385    out
386}
387
388/// A run's deliverer: delivers beside the run; dropped, it gives what is
389/// pending at most [`GRACE`]. One per process — a feature run's tasks share
390/// their run's.
391#[derive(Debug)]
392pub struct Deliverer {
393    stop: mpsc::Sender<()>,
394    done: mpsc::Receiver<()>,
395}
396
397/// How long a run waits for its last deliveries.
398pub const GRACE: Duration = Duration::from_secs(5);
399
400static ACTIVE: std::sync::atomic::AtomicBool = std::sync::atomic::AtomicBool::new(false);
401
402impl Deliverer {
403    /// Starts delivering for the project at `root`, when it has observers
404    /// and no deliverer runs in this process yet.
405    #[must_use]
406    pub fn start(root: &Path, state: &Path) -> Option<Self> {
407        use std::sync::atomic::Ordering;
408        let has_observers = super::load(root)
409            .map(|h| h.iter().any(|d| d.at().is_none()))
410            .unwrap_or(false);
411        if !has_observers || ACTIVE.swap(true, Ordering::SeqCst) {
412            return None;
413        }
414        let (stop, stopped) = mpsc::channel::<()>();
415        let (finished, done) = mpsc::channel::<()>();
416        let (root, state): (PathBuf, PathBuf) = (root.to_path_buf(), state.to_path_buf());
417        std::thread::spawn(move || {
418            loop {
419                let _ = pass(&root, &state, None);
420                match stopped.recv_timeout(Duration::from_secs(1)) {
421                    Err(mpsc::RecvTimeoutError::Timeout) => {}
422                    Ok(()) | Err(mpsc::RecvTimeoutError::Disconnected) => break,
423                }
424            }
425            let _ = pass(&root, &state, Some(Instant::now() + GRACE));
426            let _ = finished.send(());
427        });
428        Some(Self { stop, done })
429    }
430}
431
432impl Drop for Deliverer {
433    fn drop(&mut self) {
434        let _ = self.stop.send(());
435        let _ = self.done.recv_timeout(GRACE + Duration::from_secs(1));
436        ACTIVE.store(false, std::sync::atomic::Ordering::SeqCst);
437    }
438}
439
440#[cfg(test)]
441mod tests {
442    use super::*;
443
444    #[test]
445    fn a_cursor_moves_per_event() {
446        let mut c = Cursor::default();
447        let tick = Event {
448            id: "tick:abc".into(),
449            kind: "task.ticked",
450            envelope: Value::Null,
451        };
452        advance(&mut c, "refs/heads/speccoder/x", &tick);
453        assert_eq!(c.tips["refs/heads/speccoder/x"], "abc");
454        let stop = Event {
455            id: "attempt:def".into(),
456            kind: "task.stopped",
457            envelope: Value::Null,
458        };
459        advance(&mut c, "refs/speccoder/attempts/x/T1/r", &stop);
460        assert_eq!(c.attempts, ["attempt:def"]);
461    }
462}
speccoder_runtime/hooks/deliver.rs

speccoder_runtime/hooks/
deliver.rs
