feat(surface,sdk,kernel): event triggers via webhook + inbox watcher (#301)

kernel/relayflowd-journal/src/lib.rs

@@ -79,28 +79,33 @@ impl SqliteJournal {
             std::fs::create_dir_all(parent)?;
         }
         let run_id = run_id.into();
-        let connection = Connection::open_with_flags(
+        let mut connection = Connection::open_with_flags(
             &path,
             OpenFlags::SQLITE_OPEN_READ_WRITE | OpenFlags::SQLITE_OPEN_CREATE,
         )?;
         configure(&connection)?;
-        connection.execute_batch(SCHEMA)?;
-        let mut journal = Self {
-            connection,
-            run_id,
-            path,
-        };
-        let transaction = journal.connection.transaction()?;
+        // Schema DDL + meta/segment rows land in one transaction so SIGKILL
+        // between create() and the first append() never leaves the file with
+        // schema but no meta row. Without this, open()'s
+        // "SELECT value FROM meta WHERE key = 'run_id'" surfaces as
+        // "Query returned no rows" on resume (kernel/journal SIGKILL race:
+        // reproduced by webhook-live's SIGKILL-after-spawn-before-ack test).
+        let transaction = connection.transaction()?;
+        transaction.execute_batch(SCHEMA)?;
         transaction.execute(
             "INSERT INTO meta(key, value) VALUES ('run_id', ?1), ('created_at_ms', ?2), ('journal_version', ?3)",
-            params![journal.run_id, created_at_ms.to_string(), relayflowd_core::JOURNAL_VERSION.to_string()],
+            params![run_id, created_at_ms.to_string(), relayflowd_core::JOURNAL_VERSION.to_string()],
         )?;
         transaction.execute(
             "INSERT INTO segments(segment_id, journal_version, opened_seq) VALUES (1, ?1, 1)",
             [i64::from(relayflowd_core::JOURNAL_VERSION)],
         )?;
         transaction.commit()?;
-        Ok(journal)
+        Ok(Self {
+            connection,
+            run_id,
+            path,
+        })
     }
 
     pub fn open(path: impl AsRef<Path>) -> Result<Self, JournalStoreError> {

kernel/relayflowd/src/engine/wake.rs

@@ -11,6 +11,14 @@ use ulid::Ulid;
 
 use super::{DriveOptions, Engine, RunOutcome, canonical_hash};
 
+/// Additive event-run payload: normal run.spawned readers retain compatibility.
+#[derive(Serialize)]
+struct EventRunSpawnedPayload {
+    #[serde(flatten)]
+    run: RunSpawnedPayload,
+    event: serde_json::Value,
+}
+
 /// Default silence budget when a trigger does not declare its own
 /// `stale_after_ms`. 5 minutes is long enough not to trip a sluggish
 /// external stream during a normal quiet stretch, short enough that a
@@ -32,8 +40,31 @@ impl<C: Clock> Engine<C> {
         spec: RunSpec,
         event: Event,
         created_by: &str,
+    ) -> Result<EventSubmitOutcome> {
+        self.submit_event_inner(spec, event, created_by, None)
+    }
+
+    /// Inbox retries resume the claimed run before acknowledging its file.
+    pub fn submit_webhook_event(
+        &self,
+        spec: RunSpec,
+        event: Event,
+        resume: &dyn Fn(&str) -> Result<RunOutcome>,
+    ) -> Result<EventSubmitOutcome> {
+        self.submit_event_inner(spec, event, "webhook", Some(resume))
+    }
+
+    fn submit_event_inner(
+        &self,
+        spec: RunSpec,
+        event: Event,
+        created_by: &str,
+        inbox_resume: Option<&dyn Fn(&str) -> Result<RunOutcome>>,
     ) -> Result<EventSubmitOutcome> {
         spec.validate().context("invalid run spec")?;
+        if inbox_resume.is_some() {
+            self.preflight_placement(&spec)?;
+        }
         let Some(trigger) = spec
             .triggers
             .iter()
@@ -107,16 +138,20 @@ impl<C: Clock> Engine<C> {
             stale_after_ms,
             self.clock.now_ms(),
         )?;
-        if self
-            .registry()?
-            .claim_event(&flow_key, &trigger.id, &event_key, &run_id, self.boot_id())?
-            .is_some()
-        {
+        if let Some(existing_run) = self.registry()?.claim_event(
+            &flow_key,
+            &trigger.id,
+            &event_key,
+            &run_id,
+            self.boot_id(),
+        )? {
             return Ok(EventSubmitOutcome {
                 matched: true,
                 deduped: true,
                 subscription_id: Some(trigger.id),
-                run: None,
+                run: inbox_resume
+                    .map(|resume| resume(&existing_run))
+                    .transpose()?,
             });
         }
         // The claim is now held by this boot, and `claim_event` will tell any
@@ -216,12 +251,15 @@ impl<C: Clock> Engine<C> {
                 None,
                 None,
                 now_ms,
-                RunSpawnedPayload {
-                    spec: spec_value.clone(),
-                    spec_hash: canonical_hash(&spec_value),
-                    parent_run_id: None,
-                    journal_version: relayflowd_core::JOURNAL_VERSION,
-                    created_by: created_by.to_owned(),
+                EventRunSpawnedPayload {
+                    run: RunSpawnedPayload {
+                        spec: spec_value.clone(),
+                        spec_hash: canonical_hash(&spec_value),
+                        parent_run_id: None,
+                        journal_version: relayflowd_core::JOURNAL_VERSION,
+                        created_by: created_by.to_owned(),
+                    },
+                    event: event.payload.clone(),
                 },
             ),
         )?;
@@ -292,16 +330,15 @@ impl Drop for ClaimGuard {
         // during an unwind, when holding a borrow of the engine would constrain
         // the guard's lifetime to it for no benefit. `Registry::open` is what
         // every other caller here does per operation anyway.
-        let released = Registry::open(self.data_dir.join("relayflowd.sqlite3")).and_then(
-            |registry| {
+        let released =
+            Registry::open(self.data_dir.join("relayflowd.sqlite3")).and_then(|registry| {
                 registry.release_claim(
                     &self.flow_key,
                     &self.subscription_id,
                     &self.event_key,
                     &self.run_id,
                 )
-            },
-        );
+            });
         if let Err(error) = released {
             // Report, do not panic. Panicking in `Drop` during an unwind aborts
             // the process, which would turn a stranded event into a dead daemon.
@@ -344,7 +381,9 @@ mod claim_guard_tests {
     /// Take a claim the way `submit_event` does, without registering a run.
     fn claim(data_dir: &std::path::Path, run_id: &str) {
         assert_eq!(
-            registry(data_dir).claim_event(FLOW, SUB, KEY, run_id, BOOT).unwrap(),
+            registry(data_dir)
+                .claim_event(FLOW, SUB, KEY, run_id, BOOT)
+                .unwrap(),
             None,
             "the first claim must be granted"
         );

kernel/relayflowd/src/lib.rs

@@ -4,6 +4,7 @@ pub mod exec_det;
 pub mod memory;
 pub mod server;
 pub mod socket_path;
+pub mod trigger_watcher;
 pub mod worker;
 
 pub use engine::{

kernel/relayflowd/src/server.rs

@@ -74,6 +74,16 @@ pub fn serve(data_dir: &Path) -> Result<()> {
     // because the failure mode it catches is "minutes without an event",
     // not "seconds without a heartbeat".
     liveness::spawn_liveness_sweep(data_dir.to_path_buf());
+    let trigger_hub = hub.clone();
+    let trigger_dir = data_dir.to_path_buf();
+    crate::trigger_watcher::spawn_watcher(trigger_dir.clone(), move |spec, event| {
+        let engine = Engine::with_runtime(&trigger_dir, trigger_hub.clone(), trigger_hub.clone());
+        engine.submit_webhook_event(spec, event, &|run_id| {
+            let lock = trigger_hub.run_lock(run_id);
+            let _guard = lock.lock().expect("run lock");
+            engine.resume_live(run_id, trigger_hub.as_ref())
+        })
+    });
     let next_connection = Arc::new(AtomicU64::new(1));
     for connection in listener.incoming() {
         let connection = connection?;

kernel/relayflowd/src/trigger_watcher.rs

@@ -0,0 +1,137 @@
+//! Local inbox ingress. Each `triggers/<name>.json` is a compiled RunSpec whose
+//! event_type and executor equal <name>. Provision these files before ingress;
+//! do not change a binding while its inbox is pending (dedupe is spec-scoped).
+//! TODO https://github.com/AgentWorkforce/flows/issues/301: bind sealed bundles
+//! through flows deploy; TS handler deployment and Cloud mounts are separate.
+//! No author process stays alive between events. The daemon polls at 1 Hz.
+
+use crate::engine::{EventSubmitOutcome, read_spec};
+use anyhow::{Context, Result, bail};
+use relayflowd_core::{Event, RunSpec};
+use std::{
+    fs,
+    path::{Path, PathBuf},
+    thread,
+    time::Duration,
+};
+
+const MAX_EVENT_BYTES: u64 = 1024 * 1024;
+
+pub fn valid_name(name: &str) -> bool {
+    !name.is_empty()
+        && name.len() <= 128
+        && name.as_bytes()[0].is_ascii_alphanumeric()
+        && name
+            .bytes()
+            .all(|c| c.is_ascii_alphanumeric() || c == b'_' || c == b'-')
+}
+
+/// The callback is the engine boundary; filesystem traversal never executes code.
+/// Errors retain the offending file and do not starve other inboxes.
+pub fn poll_once(
+    data_dir: &Path,
+    submit: &mut impl FnMut(RunSpec, Event) -> Result<EventSubmitOutcome>,
+) -> Result<Vec<String>> {
+    let inbox = data_dir.join("inbox");
+    fs::create_dir_all(&inbox)?;
+    let mut errors = Vec::new();
+    for directory in fs::read_dir(&inbox)? {
+        let directory = directory?;
+        if !directory.file_type()?.is_dir() {
+            continue;
+        }
+        let Some(name) = directory.file_name().to_str().map(str::to_owned) else {
+            continue;
+        };
+        if !valid_name(&name) {
+            continue;
+        }
+        for file in fs::read_dir(directory.path())? {
+            let file = file?;
+            if !file.file_type()?.is_file() {
+                continue;
+            }
+            let filename = file.file_name();
+            let Some(filename) = filename.to_str() else {
+                continue;
+            };
+            let Some(id) = filename.strip_suffix(".json") else {
+                continue;
+            };
+            if !valid_name(id) {
+                continue;
+            }
+            if let Err(error) = process_file(data_dir, &name, filename, &file.path(), submit) {
+                errors.push(format!("{}: {error:#}", file.path().display()));
+            }
+        }
+    }
+    Ok(errors)
+}
+
+fn process_file(
+    data_dir: &Path,
+    name: &str,
+    filename: &str,
+    path: &Path,
+    submit: &mut impl FnMut(RunSpec, Event) -> Result<EventSubmitOutcome>,
+) -> Result<()> {
+    if fs::metadata(path)?.len() > MAX_EVENT_BYTES {
+        bail!("event exceeds 1 MiB");
+    }
+    let event = Event {
+        event_type: name.to_owned(),
+        payload: serde_json::from_slice(&fs::read(path)?).context("parse inbox event")?,
+        key: Some(filename.to_owned()),
+    };
+    let binding = data_dir.join("triggers").join(format!("{name}.json"));
+    if !fs::symlink_metadata(&binding)?.is_file() {
+        bail!("trigger binding must be a regular file");
+    }
+    let spec = read_spec(&binding)?;
+    spec.validate().context("invalid trigger spec")?;
+    if spec.triggers.is_empty()
+        || spec
+            .triggers
+            .iter()
+            .any(|trigger| trigger.executor != name || trigger.event_type.as_deref() != Some(name))
+    {
+        bail!("trigger binding must declare executor and event_type {name:?}");
+    }
+    let outcome = submit(spec, event)?;
+    // A nonmatch is a consumed filter rejection. A matched submission must be
+    // durable and driven (or resumed) before moving: crash-before-move retries.
+    if outcome.matched && outcome.run.is_none() {
+        bail!("matched event has no durable run receipt");
+    }
+    let processed = data_dir.join("inbox-processed").join(name);
+    fs::create_dir_all(&processed)?;
+    if !fs::symlink_metadata(&processed)?.is_dir() {
+        bail!("processed inbox must be a directory");
+    }
+    fs::rename(path, processed.join(filename)).context("archive inbox event")?;
+    fs::File::open(&processed)?.sync_all()?;
+    fs::File::open(path.parent().context("inbox parent")?)?.sync_all()?;
+    Ok(())
+}
+
+pub(crate) fn spawn_watcher(
+    data_dir: PathBuf,
+    mut submit: impl FnMut(RunSpec, Event) -> Result<EventSubmitOutcome> + Send + 'static,
+) {
+    thread::spawn(move || {
+        loop {
+            match poll_once(&data_dir, &mut submit) {
+                Ok(errors) => {
+                    for error in errors {
+                        eprintln!("relayflowd: inbox retained: {error}");
+                    }
+                }
+                Err(error) => eprintln!("relayflowd: inbox poll failed: {error:#}"),
+            }
+            // TODO https://github.com/AgentWorkforce/flows/issues/301: native watch
+            // notifications are a follow-up; this slice deliberately polls at 1 Hz.
+            thread::sleep(Duration::from_secs(1));
+        }
+    });
+}