feat(daemon): connection-file handshake and CLI attach-or-spawn

docs/SURFACE.md

@@ -298,17 +298,43 @@ Gate 1 ships three CLI verbs over the journal protocol:
 
 ```text
 flows check [--json] <flow.yaml|spec.json>
-flows run [--json] [--data-dir <dir>] <flow.yaml|spec.json>
-flows run [--json] [--data-dir <dir>] <flow.ts> --input <inline-json-or-file>
-flows resume [--json] [--data-dir <dir>] <run-id>
+flows run [--json] [--no-spawn] [--data-dir <dir>] <flow.yaml|spec.json>
+flows run [--json] [--no-spawn] [--data-dir <dir>] <flow.ts> --input <inline-json-or-file>
+flows resume [--json] [--no-spawn] [--data-dir <dir>] <run-id>
 ```
 
 `check` compiles and preflights without starting a run. `run` performs that
 same preflight before contacting `relayflowd`, then submits the compiled spec
 to `<data-dir>/relayflowd.sock`; `resume` asks that daemon to continue an
 existing run from its journal. The data directory defaults to `.relayflowd`.
-Neither verb starts the daemon implicitly. `--json` writes one report-shaped
-object to stdout while diagnostics remain on stderr.
+`--json` writes one report-shaped object to stdout while diagnostics remain on
+stderr.
+
+`run` and `resume` attach to the daemon serving `<data-dir>` or start one
+(kernel/DAEMON-LIFECYCLE.md). The attach is decided by the socket, not by a
+file: `<data-dir>/connection.json` is read and validated, but nothing is
+attached to until a `hello` is answered on the socket path recomputed from
+`--data-dir`. When nothing answers, the CLI spawns `relayflowd serve
+--data-dir <dir>` detached — in its own session, with stdio never inheriting
+the CLI's and its stderr appended to `<data-dir>/relayflowd.log` — and polls
+for it, bounded. Concurrent invocations are safe: the daemon holds an
+exclusive lock on the data dir, so a redundant one exits without touching
+anything and its CLI attaches to the winner.
+
+`relayflowd serve` remains fully supported and unchanged for an operator who
+starts it by hand; `run` and `resume` attach to it and never signal, restart,
+or terminate a daemon. `--no-spawn` (or `FLOWS_NO_SPAWN=1`) refuses instead of
+starting one — the lever for CI that means to assert a daemon is already
+present. `check` never opens a socket and needs no daemon, no data directory,
+and no `relayflowd` binary at all.
+
+A `run` or `resume` that cannot get a daemon is refused before any journal
+write (exit 2) and names which step failed: `daemon_unreachable` under
+`--no-spawn`, `relayflowd_not_found` when no binary could be located,
+`daemon_start_failed` when the spawned daemon exited during startup,
+`daemon_start_timeout` when it never began serving, and
+`daemon_protocol_mismatch` against a live daemon speaking another protocol
+version — which refuses rather than starting a second daemon over it.
 
 A direct `.flow.ts` run requires `--input`. When its argument names an existing
 regular file, the CLI parses that file as JSON; otherwise it parses the argument

kernel/relayflowd/src/server.rs

@@ -11,6 +11,8 @@ mod cancel;
 #[cfg(unix)]
 mod channels;
 #[cfg(unix)]
+mod lifecycle;
+#[cfg(unix)]
 pub mod liveness;
 #[cfg(unix)]
 mod reconcile;
@@ -46,13 +48,22 @@ pub fn serve(data_dir: &Path) -> Result<()> {
     };
 
     std::fs::create_dir_all(data_dir)?;
+    let _lock = match lifecycle::acquire(data_dir) {
+        Ok(lock) => lock,
+        Err(lifecycle::AcquireError::AlreadyServing) => {
+            eprintln!(
+                "relayflowd: another relayflowd is already serving {}",
+                data_dir.display()
+            );
+            std::process::exit(3);
+        }
+        Err(lifecycle::AcquireError::Io(error)) => return Err(error.into()),
+    };
     let socket_path = data_dir.join("relayflowd.sock");
-    if socket_path.exists() {
-        std::fs::remove_file(&socket_path)
-            .with_context(|| format!("remove stale socket {}", socket_path.display()))?;
-    }
+    lifecycle::remove_residue(data_dir)?;
     let listener = UnixListener::bind(&socket_path)
         .with_context(|| format!("bind socket {}", socket_path.display()))?;
+    lifecycle::publish(&socket_path)?;
     let hub = Arc::new(ProtocolHub::default());
     reconcile::spawn_reconciler(data_dir.to_path_buf(), hub.clone());
     // Trigger-plane liveness sweep (RFC-0001 gate 2, Native silent-death

kernel/relayflowd/src/server/lifecycle.rs

@@ -0,0 +1,156 @@
+//! Filesystem ownership and advertisement for a serving daemon.
+
+use std::{
+    ffi::CString,
+    fs::{File, OpenOptions},
+    io::{self, Write},
+    os::{
+        raw::{c_char, c_int},
+        unix::{fs::OpenOptionsExt, io::AsRawFd},
+    },
+    path::Path,
+    ptr,
+    sync::atomic::{AtomicPtr, Ordering},
+    time::{SystemTime, UNIX_EPOCH},
+};
+
+use anyhow::{Context, Result};
+use relayflowd_core::PROTOCOL_VERSION;
+use serde::Serialize;
+
+static CONNECTION_PATH: AtomicPtr<c_char> = AtomicPtr::new(ptr::null_mut());
+static SOCKET_PATH: AtomicPtr<c_char> = AtomicPtr::new(ptr::null_mut());
+
+/// Keep this value alive for the entire `serve` call.
+pub(crate) struct DaemonLock {
+    _file: File,
+}
+
+#[derive(Debug)]
+pub(crate) enum AcquireError {
+    AlreadyServing,
+    Io(io::Error),
+}
+
+impl std::fmt::Display for AcquireError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            Self::AlreadyServing => f.write_str("another relayflowd is already serving"),
+            Self::Io(error) => error.fmt(f),
+        }
+    }
+}
+impl std::error::Error for AcquireError {}
+
+pub(crate) fn acquire(data_dir: &Path) -> std::result::Result<DaemonLock, AcquireError> {
+    let lock = OpenOptions::new()
+        .create(true)
+        .read(true)
+        .write(true)
+        .mode(0o600)
+        .open(data_dir.join("relayflowd.lock"))
+        .map_err(AcquireError::Io)?;
+    if unsafe { libc::flock(lock.as_raw_fd(), libc::LOCK_EX | libc::LOCK_NB) } == 0 {
+        return Ok(DaemonLock { _file: lock });
+    }
+    let error = io::Error::last_os_error();
+    if error.kind() == io::ErrorKind::WouldBlock {
+        Err(AcquireError::AlreadyServing)
+    } else {
+        Err(AcquireError::Io(error))
+    }
+}
+
+/// Safe only after `acquire`: no other daemon can own these paths.
+pub(crate) fn remove_residue(data_dir: &Path) -> Result<()> {
+    for name in ["relayflowd.sock", "connection.json"] {
+        let path = data_dir.join(name);
+        match std::fs::remove_file(&path) {
+            Ok(()) => {}
+            Err(error) if error.kind() == io::ErrorKind::NotFound => {}
+            Err(error) => {
+                return Err(error).with_context(|| format!("remove stale {}", path.display()));
+            }
+        }
+    }
+    Ok(())
+}
+
+#[derive(Serialize)]
+struct ConnectionFile<'a> {
+    socket_path: &'a str,
+    pid: u32,
+    version: &'a str,
+    protocol: u32,
+    started_at_ms: u128,
+}
+
+/// Publish only after `UnixListener::bind`, which has already called listen(2).
+pub(crate) fn publish(socket_path: &Path) -> Result<()> {
+    // `absolute` preserves the caller's data-dir spelling (unlike
+    // `canonicalize`, which would turn `/var` into macOS's `/private/var` and
+    // fail the client's lexical `resolve(dataDir)` comparison).
+    let socket_path = std::path::absolute(socket_path)
+        .with_context(|| format!("make socket path absolute {}", socket_path.display()))?;
+    let socket_text = socket_path
+        .to_str()
+        .context("socket path is not valid UTF-8")?;
+    let data_dir = socket_path.parent().context("socket path has no parent")?;
+    let connection_path = data_dir.join("connection.json");
+    let temporary_path = data_dir.join(format!("connection.json.tmp.{}", std::process::id()));
+    let record = ConnectionFile {
+        socket_path: socket_text,
+        pid: std::process::id(),
+        version: env!("CARGO_PKG_VERSION"),
+        protocol: PROTOCOL_VERSION,
+        started_at_ms: SystemTime::now().duration_since(UNIX_EPOCH)?.as_millis(),
+    };
+    let bytes = serde_json::to_vec(&record)?;
+    let result = (|| -> Result<()> {
+        let mut file = OpenOptions::new()
+            .write(true)
+            .create_new(true)
+            .mode(0o600)
+            .open(&temporary_path)
+            .with_context(|| format!("create {}", temporary_path.display()))?;
+        file.write_all(&bytes)?;
+        file.sync_all()?;
+        // Register before making the advertisement visible. A caller may send
+        // SIGTERM the instant rename returns, so visibility must imply cleanup
+        // is already installed.
+        install_cleanup_handlers(&connection_path, &socket_path)?;
+        std::fs::rename(&temporary_path, &connection_path)
+            .with_context(|| format!("publish {}", connection_path.display()))?;
+        Ok(())
+    })();
+    if result.is_err() {
+        let _ = std::fs::remove_file(&temporary_path);
+    }
+    result
+}
+
+fn install_cleanup_handlers(connection_path: &Path, socket_path: &Path) -> Result<()> {
+    // These strings remain valid until the signal handler calls `_exit`.
+    let connection = CString::new(connection_path.as_os_str().as_encoded_bytes())?;
+    let socket = CString::new(socket_path.as_os_str().as_encoded_bytes())?;
+    CONNECTION_PATH.store(connection.into_raw(), Ordering::Relaxed);
+    SOCKET_PATH.store(socket.into_raw(), Ordering::Relaxed);
+    unsafe {
+        if libc::signal(libc::SIGTERM, on_terminate as *const () as usize) == libc::SIG_ERR
+            || libc::signal(libc::SIGINT, on_terminate as *const () as usize) == libc::SIG_ERR
+        {
+            return Err(io::Error::last_os_error().into());
+        }
+    }
+    Ok(())
+}
+
+extern "C" fn on_terminate(_signal: c_int) {
+    for slot in [&CONNECTION_PATH, &SOCKET_PATH] {
+        let path = slot.load(Ordering::Relaxed);
+        if !path.is_null() {
+            unsafe { libc::unlink(path) };
+        }
+    }
+    unsafe { libc::_exit(0) }
+}

kernel/relayflowd/src/server/liveness.rs

@@ -120,7 +120,12 @@ pub fn sweep_pass(
         // a no-op. The signal we just emitted was a spurious alert
         // (harmless — at-least-once observability); the row remains
         // sweep-visible for the next real crossing.
-        match registry.latch_stale(&row.flow_key, &row.subscription_id, row.last_event_at_ms, now_ms) {
+        match registry.latch_stale(
+            &row.flow_key,
+            &row.subscription_id,
+            row.last_event_at_ms,
+            now_ms,
+        ) {
             Ok(true) => {}
             Ok(false) => {
                 eprintln!(
@@ -251,7 +256,9 @@ mod tests {
         sweep_pass(dir.path(), &sweep_id_for(1_040_000), "w", 1_040_000).unwrap();
 
         // Next bucket, same row: must NOT re-emit (latched).
-        let leftover = registry.detect_stale(&sweep_id_for(1_080_000), "w", 1_080_000).unwrap();
+        let leftover = registry
+            .detect_stale(&sweep_id_for(1_080_000), "w", 1_080_000)
+            .unwrap();
         assert!(
             leftover.is_empty(),
             "subscription.stale re-emitted after being latched: {leftover:?}"

kernel/relayflowd/src/server/tests.rs

@@ -215,19 +215,21 @@ fn run_resume_adopts_a_real_journal_whose_registry_row_is_missing() {
         )
         .unwrap();
     let run_id = outcome.run_id.clone();
-    assert!(data_dir.join("runs").join(format!("{run_id}.sqlite3")).exists());
+    assert!(
+        data_dir
+            .join("runs")
+            .join(format!("{run_id}.sqlite3"))
+            .exists()
+    );
 
     // Reproduce the crash window: the journal survives, the index entry does
     // not. Removing the registry outright is the same state a SIGKILL between
     // the append and the register leaves behind, and strictly harsher -- the
     // row is not merely stale, there is nothing to consult at all.
     for suffix in ["", "-wal", "-shm"] {
-        let _ = std::fs::remove_file(
-            data_dir.join(format!("relayflowd.sqlite3{suffix}")),
-        );
+        let _ = std::fs::remove_file(data_dir.join(format!("relayflowd.sqlite3{suffix}")));
     }
-    let registry =
-        relayflowd_journal::Registry::open(data_dir.join("relayflowd.sqlite3")).unwrap();
+    let registry = relayflowd_journal::Registry::open(data_dir.join("relayflowd.sqlite3")).unwrap();
     assert!(registry.lookup(&run_id).unwrap().is_none());
 
     let hub = Arc::new(ProtocolHub::default());
@@ -237,9 +239,7 @@ fn run_resume_adopts_a_real_journal_whose_registry_row_is_missing() {
         &hub,
         1,
         &writer,
-        &format!(
-            r#"{{"id":"resume","verb":"run.resume","params":{{"run_id":"{run_id}"}}}}"#
-        ),
+        &format!(r#"{{"id":"resume","verb":"run.resume","params":{{"run_id":"{run_id}"}}}}"#),
     );
 
     assert!(
@@ -275,8 +275,15 @@ fn run_resume_refuses_a_journal_that_never_recorded_its_run() {
     std::fs::create_dir_all(data_dir.join("runs")).unwrap();
     let path = data_dir.join("runs").join(format!("{run_id}.sqlite3"));
     let journal = relayflowd_journal::SqliteJournal::create(&path, run_id, 0).unwrap();
-    assert_eq!(journal.run_id(), run_id, "the meta row is what makes this tempting");
-    assert!(journal.run_spec().is_err(), "and there is no spec to resume");
+    assert_eq!(
+        journal.run_id(),
+        run_id,
+        "the meta row is what makes this tempting"
+    );
+    assert!(
+        journal.run_spec().is_err(),
+        "and there is no spec to resume"
+    );
     drop(journal);
 
     let hub = Arc::new(ProtocolHub::default());
@@ -298,8 +305,7 @@ fn run_resume_refuses_a_journal_that_never_recorded_its_run() {
          failure is not"
     );
 
-    let registry =
-        relayflowd_journal::Registry::open(data_dir.join("relayflowd.sqlite3")).unwrap();
+    let registry = relayflowd_journal::Registry::open(data_dir.join("relayflowd.sqlite3")).unwrap();
     assert!(
         registry.lookup(run_id).unwrap().is_none(),
         "a refused journal must not leave a registry row behind"
@@ -363,8 +369,7 @@ fn run_resume_refuses_a_valid_journal_that_belongs_to_another_run() {
     assert_eq!(error.code, "run_not_found");
 
     // And nothing was adopted under the impostor id.
-    let registry =
-        relayflowd_journal::Registry::open(data_dir.join("relayflowd.sqlite3")).unwrap();
+    let registry = relayflowd_journal::Registry::open(data_dir.join("relayflowd.sqlite3")).unwrap();
     assert!(
         registry.lookup(impostor).unwrap().is_none(),
         "a refused journal must not leave a registry row behind"