vcli-daemon: long-running process that wires IPC + store + runtime (plan 4)

Cargo.toml

@@ -3,6 +3,7 @@ resolver = "2"
 members = [
     "crates/vcli-core",
     "crates/vcli-capture",
+    "crates/vcli-daemon",
     "crates/vcli-dsl",
     "crates/vcli-input",
     "crates/vcli-ipc",
@@ -29,10 +30,18 @@ ryu = "1"
 unicode-normalization = "0.1"
 
 # Added by vcli-ipc lane
-tokio = { version = "1", features = ["net", "io-util", "macros", "rt-multi-thread", "sync", "time"] }
+tokio = { version = "1", features = ["net", "io-util", "macros", "rt-multi-thread", "sync", "time", "signal"] }
 async-trait = "0.1"
 futures-core = "0.3"
 
+# Added by vcli-daemon lane
+tracing-subscriber = { version = "0.3", features = ["fmt", "env-filter", "json"] }
+tracing-appender = "0.2"
+fs2 = "0.4"
+dirs = "5"
+signal-hook-tokio = { version = "0.3", features = ["futures-v0_3"] }
+futures-util = "0.3"
+
 rusqlite = { version = "0.31", features = ["bundled"] }
 sha2 = "0.10"
 hex = "0.4"

crates/vcli-core/src/clock.rs

@@ -109,6 +109,25 @@ impl Clock for TestClock {
     }
 }
 
+/// Convenience for callers that just want wall-clock ms and do not hold a
+/// `Clock` handle. Equivalent to `SystemClock::new().unix_ms()`. Non-test code
+/// outside the scheduler uses this for event timestamps.
+#[must_use]
+pub fn now_unix_ms() -> UnixMs {
+    SystemClock::new().unix_ms()
+}
+
+#[cfg(test)]
+mod now_unix_ms_tests {
+    use super::*;
+
+    #[test]
+    fn now_unix_ms_is_positive_and_recent() {
+        let t = now_unix_ms();
+        assert!(t > 1_700_000_000_000, "got {t}");
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

crates/vcli-core/src/lib.rs

@@ -21,7 +21,7 @@ pub use frame::{Frame, FrameFormat};
 
 pub mod clock;
 
-pub use clock::{Clock, SystemClock, TestClock, UnixMs};
+pub use clock::{now_unix_ms, Clock, SystemClock, TestClock, UnixMs};
 
 pub mod region;
 

crates/vcli-daemon/Cargo.toml

@@ -0,0 +1,47 @@
+[package]
+name = "vcli-daemon"
+version.workspace = true
+edition.workspace = true
+rust-version.workspace = true
+license.workspace = true
+repository.workspace = true
+authors.workspace = true
+description = "The vcli daemon binary — tokio reactor + scheduler bridge + socket server."
+
+[lib]
+path = "src/lib.rs"
+
+[[bin]]
+name = "vcli-daemon"
+path = "src/bin/vcli-daemon.rs"
+
+[dependencies]
+vcli-core       = { path = "../vcli-core" }
+vcli-dsl        = { path = "../vcli-dsl" }
+vcli-store      = { path = "../vcli-store" }
+vcli-ipc        = { path = "../vcli-ipc" }
+vcli-capture    = { path = "../vcli-capture" }
+vcli-input      = { path = "../vcli-input" }
+vcli-perception = { path = "../vcli-perception" }
+vcli-runtime    = { path = "../vcli-runtime" }
+
+serde              = { workspace = true }
+serde_json         = { workspace = true }
+thiserror          = { workspace = true }
+async-trait        = { workspace = true }
+tokio              = { workspace = true }
+tracing            = { workspace = true }
+tracing-subscriber = { workspace = true }
+tracing-appender   = { workspace = true }
+crossbeam-channel  = { workspace = true }
+fs2                = { workspace = true }
+dirs               = { workspace = true }
+futures-util       = { workspace = true }
+signal-hook-tokio  = { workspace = true }
+uuid               = { workspace = true }
+libc               = { workspace = true }
+
+[dev-dependencies]
+tempfile    = { workspace = true }
+tokio-test  = { workspace = true }
+tokio       = { workspace = true, features = ["test-util"] }

crates/vcli-daemon/build.rs

@@ -0,0 +1,13 @@
+fn main() {
+    #[cfg(target_os = "macos")]
+    {
+        // vcli-daemon pulls in vcli-capture → screencapturekit transitively.
+        // screencapturekit's Swift bridge references libswift_Concurrency.dylib.
+        // vcli-capture sets this rpath for its own artifacts; we repeat it here
+        // so vcli-daemon's test + binary output can load the Swift dylib too.
+        let xcode_swift55_path =
+            "/Applications/Xcode.app/Contents/Developer/Toolchains/XcodeDefault.xctoolchain\
+             /usr/lib/swift-5.5/macosx";
+        println!("cargo:rustc-link-arg=-Wl,-rpath,{xcode_swift55_path}");
+    }
+}

crates/vcli-daemon/src/bin/vcli-daemon.rs

@@ -0,0 +1,73 @@
+//! vcli-daemon entrypoint. Minimal argv surface:
+//!   * `--version` → print crate version, exit 0
+//!   * `--help`    → print usage, exit 0
+//!
+//! All real work lives in `vcli_daemon::run_foreground`.
+
+use std::process::ExitCode;
+
+use vcli_daemon::{run_foreground, Config, DaemonError, RuntimeBackends};
+
+fn main() -> ExitCode {
+    let args: Vec<String> = std::env::args().collect();
+    if args.iter().any(|a| a == "--version" || a == "-V") {
+        println!("vcli-daemon {}", env!("CARGO_PKG_VERSION"));
+        return ExitCode::SUCCESS;
+    }
+    if args.iter().any(|a| a == "--help" || a == "-h") {
+        println!("usage: vcli-daemon");
+        println!("  no flags — runs in the foreground until SIGTERM/SIGINT");
+        return ExitCode::SUCCESS;
+    }
+
+    let rt = match tokio::runtime::Builder::new_multi_thread()
+        .enable_all()
+        .build()
+    {
+        Ok(r) => r,
+        Err(e) => {
+            eprintln!("vcli-daemon: failed to build tokio runtime: {e}");
+            return ExitCode::from(1);
+        }
+    };
+
+    let cfg = match Config::from_platform_defaults() {
+        Ok(c) => c,
+        Err(e) => {
+            eprintln!("vcli-daemon: {e}");
+            return ExitCode::from(1);
+        }
+    };
+
+    let factory: vcli_daemon::RuntimeFactory = Box::new(default_runtime_factory);
+
+    match rt.block_on(run_foreground(cfg, factory)) {
+        Ok(()) => ExitCode::SUCCESS,
+        Err(DaemonError::AlreadyRunning { pid, .. }) => {
+            eprintln!("vcli-daemon: already running (pid {pid})");
+            ExitCode::from(1)
+        }
+        Err(e) => {
+            eprintln!("vcli-daemon: {e}");
+            ExitCode::from(1)
+        }
+    }
+}
+
+/// Real-backend factory used in production. For v0 the daemon binary falls
+/// back to mock backends on every platform — a future lane wires the real
+/// macOS capture + input.
+fn default_runtime_factory() -> Result<RuntimeBackends, DaemonError> {
+    let capture: Box<dyn vcli_capture::Capture> = Box::new(vcli_capture::MockCapture::empty());
+    let input: std::sync::Arc<dyn vcli_input::InputSink> =
+        std::sync::Arc::new(vcli_input::MockInputSink::new());
+    let perception = vcli_perception::Perception::default();
+    let clock: std::sync::Arc<dyn vcli_runtime::RuntimeClock> =
+        std::sync::Arc::new(vcli_runtime::SystemRuntimeClock::new());
+    Ok(RuntimeBackends {
+        capture,
+        input,
+        perception,
+        clock,
+    })
+}

crates/vcli-daemon/src/bridge.rs

@@ -0,0 +1,97 @@
+//! Tokio ↔ scheduler bridge.
+//!
+//! The scheduler thread is pure sync; the tokio reactor is async. We route
+//! data between them via two `crossbeam_channel`s:
+//!   * `cmd_tx` (tokio → sched): bounded, backpressure-free; handlers send
+//!     [`SchedulerCommand`] values.
+//!   * `event_rx` (sched → tokio): unbounded. A dedicated tokio task drains
+//!     it and pushes each [`vcli_core::Event`] through the persistence sink
+//!     and then into the [`tokio::sync::broadcast`] channel for connected
+//!     IPC subscribers.
+//!
+//! `event_tx` (broadcast side) is cloned per IPC subscription; each streaming
+//! handler subscribes, reads frames, and hangs up when the client drops. The
+//! capacity of the broadcast (Decision 1.7) is 1024; overflowing clients
+//! receive a `stream.dropped` notification inside the handler.
+
+use crossbeam_channel::{bounded, unbounded, Receiver, Sender};
+use tokio::sync::broadcast;
+use vcli_core::Event;
+
+pub use vcli_runtime::SchedulerCommand;
+
+/// Capacity of the broadcast channel fanned out to IPC subscribers.
+pub const EVENT_BROADCAST_CAPACITY: usize = 1024;
+
+/// Capacity of the command channel tokio → scheduler.
+pub const CMD_CAPACITY: usize = 256;
+
+/// All the channel endpoints together. Cheap to clone; references passed into
+/// the handler and startup modules.
+#[derive(Clone)]
+pub struct CommandChannel {
+    /// Tokio → scheduler commands.
+    pub cmd_tx: Sender<SchedulerCommand>,
+    /// Broadcast fanning out persisted events to IPC subscribers.
+    pub event_tx: broadcast::Sender<Event>,
+}
+
+/// Construct a matched set of endpoints. The `cmd_rx` and `event_rx` returned
+/// are handed to the scheduler thread + the event-pump task respectively.
+#[must_use]
+pub fn new_channels() -> (
+    CommandChannel,
+    Receiver<SchedulerCommand>,
+    Receiver<Event>,
+    Sender<Event>,
+) {
+    let (cmd_tx, cmd_rx) = bounded::<SchedulerCommand>(CMD_CAPACITY);
+    let (sched_event_tx, event_rx) = unbounded::<Event>();
+    let (bcast_tx, _) = broadcast::channel::<Event>(EVENT_BROADCAST_CAPACITY);
+    (
+        CommandChannel {
+            cmd_tx,
+            event_tx: bcast_tx,
+        },
+        cmd_rx,
+        event_rx,
+        sched_event_tx,
+    )
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use vcli_core::{EventData, ProgramId};
+
+    #[test]
+    fn command_channel_is_cloneable_and_reaches_scheduler() {
+        let (chan, cmd_rx, _event_rx, _sched_event_tx) = new_channels();
+        let chan2 = chan.clone();
+        chan2
+            .cmd_tx
+            .send(SchedulerCommand::Cancel {
+                program_id: ProgramId::new(),
+                reason: "test".into(),
+            })
+            .unwrap();
+        match cmd_rx.recv().unwrap() {
+            SchedulerCommand::Cancel { .. } => {}
+            other => panic!("unexpected cmd: {other:?}"),
+        }
+    }
+
+    #[tokio::test]
+    async fn broadcast_fanout_reaches_multiple_subscribers() {
+        let (chan, _cmd_rx, _event_rx, _sched_event_tx) = new_channels();
+        let mut rx1 = chan.event_tx.subscribe();
+        let mut rx2 = chan.event_tx.subscribe();
+        let ev = Event {
+            at: 1,
+            data: EventData::DaemonStopped,
+        };
+        chan.event_tx.send(ev.clone()).unwrap();
+        assert_eq!(rx1.recv().await.unwrap(), ev);
+        assert_eq!(rx2.recv().await.unwrap(), ev);
+    }
+}