From 34196d9e66ff5956e8a3029ba05e087bf063e04b Mon Sep 17 00:00:00 2001 From: Yixiang Zeng Date: Mon, 20 Apr 2026 15:30:29 -0700 Subject: [PATCH 01/25] vcli-core: add now_unix_ms() free-function helper Wall-clock ms convenience for daemon callers that do not hold a Clock handle. Equivalent to SystemClock::new().unix_ms(). Re-exported from the crate root. Co-Authored-By: Claude Opus 4.7 (1M context) --- crates/vcli-core/src/clock.rs | 19 +++++++++++++++++++ crates/vcli-core/src/lib.rs | 2 +- 2 files changed, 20 insertions(+), 1 deletion(-) diff --git a/crates/vcli-core/src/clock.rs b/crates/vcli-core/src/clock.rs index 5cf6c4f..046fc25 100644 --- a/crates/vcli-core/src/clock.rs +++ b/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::*; diff --git a/crates/vcli-core/src/lib.rs b/crates/vcli-core/src/lib.rs index 6d617ad..231a6af 100644 --- a/crates/vcli-core/src/lib.rs +++ b/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; From ed3457b45633f4ff685fa193bf23d69156b78f25 Mon Sep 17 00:00:00 2001 From: Yixiang Zeng Date: Mon, 20 Apr 2026 15:31:08 -0700 Subject: [PATCH 02/25] vcli-store: Store::list_programs with optional state filter Ordered by submitted_at ASC then rowid for stable output. Unit tests cover both the no-filter and filtered paths. Co-Authored-By: Claude Opus 4.7 (1M context) --- crates/vcli-store/src/store.rs | 87 ++++++++++++++++++++++++++++++++++ 1 file changed, 87 insertions(+) diff --git a/crates/vcli-store/src/store.rs b/crates/vcli-store/src/store.rs index bb494a9..f443479 100644 --- a/crates/vcli-store/src/store.rs +++ b/crates/vcli-store/src/store.rs @@ -197,6 +197,50 @@ impl Store { )?; Ok(()) } + + /// List programs, optionally filtered by a single [`ProgramState`]. + /// Ordered by `submitted_at ASC` then by rowid for stability. + /// + /// # Errors + /// Surfaces `SQLite` errors. + /// + /// # Panics + /// Panics if a stored id or state value cannot be parsed (should never + /// happen for rows written by this crate). + pub fn list_programs( + &self, + state_filter: Option, + ) -> StoreResult> { + let state_str = state_filter.map(|s| s.as_str().to_string()); + let mut stmt = self.conn.prepare( + "SELECT id, name, source_json, state, submitted_at, started_at, + finished_at, last_error_code, last_error_msg, labels_json, + body_cursor, body_entered_at + FROM programs + WHERE (?1 IS NULL) OR (state = ?1) + ORDER BY submitted_at ASC, rowid ASC", + )?; + let rows = stmt.query_map(rusqlite::params![state_str], |r| { + let id: String = r.get(0)?; + let state: String = r.get(3)?; + let body_cursor: i64 = r.get(10)?; + Ok(ProgramRow { + id: id.parse().unwrap(), + name: r.get(1)?, + source_json: r.get(2)?, + state: state.parse().unwrap(), + submitted_at: r.get(4)?, + started_at: r.get(5)?, + finished_at: r.get(6)?, + last_error_code: r.get(7)?, + last_error_msg: r.get(8)?, + labels_json: r.get(9)?, + body_cursor: u32::try_from(body_cursor).unwrap_or(0), + body_entered_at: r.get(11)?, + }) + })?; + rows.collect::, _>>().map_err(Into::into) + } } /// Values a caller supplies when inserting a new program row. @@ -366,6 +410,49 @@ mod tests { assert_eq!(s.get_program(id).unwrap().body_cursor, 7); } + #[test] + fn list_programs_returns_all_when_no_filter() { + let d = tempdir().unwrap(); + let (mut s, _) = Store::open(d.path()).unwrap(); + for (i, name) in ["a", "b", "c"].into_iter().enumerate() { + s.insert_program(&NewProgram { + id: ProgramId::new(), + name, + source_json: "{}", + state: ProgramState::Pending, + submitted_at: i as i64, + labels_json: "{}", + }) + .unwrap(); + } + let all = s.list_programs(None).unwrap(); + assert_eq!(all.len(), 3); + assert_eq!(all[0].name, "a"); + } + + #[test] + fn list_programs_filters_by_state() { + let d = tempdir().unwrap(); + let (mut s, _) = Store::open(d.path()).unwrap(); + let running = ProgramId::new(); + let pending = ProgramId::new(); + for (id, name) in [(running, "r"), (pending, "p")] { + s.insert_program(&NewProgram { + id, + name, + source_json: "{}", + state: ProgramState::Pending, + submitted_at: 0, + labels_json: "{}", + }) + .unwrap(); + } + s.update_state(running, ProgramState::Running, 100).unwrap(); + let runs = s.list_programs(Some(ProgramState::Running)).unwrap(); + assert_eq!(runs.len(), 1); + assert_eq!(runs[0].name, "r"); + } + #[test] fn reopen_preserves_rows() { let d = tempdir().unwrap(); From 65d2201c6af52d9553985726ebb02ffd99c4ed71 Mon Sep 17 00:00:00 2001 From: Yixiang Zeng Date: Mon, 20 Apr 2026 15:32:41 -0700 Subject: [PATCH 03/25] vcli-store: fix clippy cast_possible_wrap in list_programs test MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit i64::try_from(i).unwrap() for the usize→i64 cast. Co-Authored-By: Claude Opus 4.7 (1M context) --- crates/vcli-store/src/store.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/crates/vcli-store/src/store.rs b/crates/vcli-store/src/store.rs index f443479..c17bddf 100644 --- a/crates/vcli-store/src/store.rs +++ b/crates/vcli-store/src/store.rs @@ -420,7 +420,7 @@ mod tests { name, source_json: "{}", state: ProgramState::Pending, - submitted_at: i as i64, + submitted_at: i64::try_from(i).unwrap(), labels_json: "{}", }) .unwrap(); From 2d7156743d38d4258e03c576bd7b17673caf2924 Mon Sep 17 00:00:00 2001 From: Yixiang Zeng Date: Mon, 20 Apr 2026 15:33:36 -0700 Subject: [PATCH 04/25] vcli-daemon: empty crate shell + workspace wiring Adds crates/vcli-daemon as a workspace member, a lib + placeholder bin, and the workspace dependencies (tracing-subscriber, tracing-appender, fs2, dirs, signal-hook-tokio, futures-util) the later tasks consume. Also extends tokio feature set with `signal`. Co-Authored-By: Claude Opus 4.7 (1M context) --- Cargo.lock | 480 +++++++++++++++++++++- Cargo.toml | 11 +- crates/vcli-daemon/Cargo.toml | 46 +++ crates/vcli-daemon/src/bin/vcli-daemon.rs | 6 + crates/vcli-daemon/src/lib.rs | 14 + 5 files changed, 539 insertions(+), 18 deletions(-) create mode 100644 crates/vcli-daemon/Cargo.toml create mode 100644 crates/vcli-daemon/src/bin/vcli-daemon.rs create mode 100644 crates/vcli-daemon/src/lib.rs diff --git a/Cargo.lock b/Cargo.lock index cea15e3..2d2b08e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -36,6 +36,15 @@ dependencies = [ "zerocopy", ] +[[package]] +name = "aho-corasick" +version = "1.1.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ddd31a130427c27518df266943a5308ed92d4b226cc639f5a8f1002816174301" +dependencies = [ + "memchr", +] + [[package]] name = "aligned" version = "0.4.3" @@ -133,7 +142,7 @@ dependencies = [ "num-traits", "pastey", "rayon", - "thiserror", + "thiserror 2.0.18", "v_frame", "y4m", ] @@ -382,6 +391,15 @@ dependencies = [ "parking_lot_core", ] +[[package]] +name = "deranged" +version = "0.5.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7cd812cc2bc1d69d4764bd80df88b4317eaef9e773c75226407d9bc0876b211c" +dependencies = [ + "powerfmt", +] + [[package]] name = "digest" version = "0.10.7" @@ -392,6 +410,27 @@ dependencies = [ "crypto-common", ] +[[package]] +name = "dirs" +version = "5.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "44c45a9d03d6676652bcb5e724c7e988de1acad23a711b5217ab9cbecbec2225" +dependencies = [ + "dirs-sys", +] + +[[package]] +name = "dirs-sys" +version = "0.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "520f05a5cbd335fae5a99ff7a6ab8627577660ee5cfd6a94a6a929b52ff0321c" +dependencies = [ + "libc", + "option-ext", + "redox_users", + "windows-sys 0.48.0", +] + [[package]] name = "either" version = "1.15.0" @@ -431,7 +470,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", - "windows-sys", + "windows-sys 0.61.2", ] [[package]] @@ -551,12 +590,52 @@ version = "0.3.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "aa9a19cbb55df58761df49b23516a86d432839add4af60fc256da840f66ed35b" +[[package]] +name = "fs2" +version = "0.4.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9564fc758e15025b46aa6643b1b77d047d1a56a1aea6e01002ac0c7026876213" +dependencies = [ + "libc", + "winapi", +] + [[package]] name = "futures-core" version = "0.3.32" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7e3450815272ef58cec6d564423f6e755e25379b217b0bc688e295ba24df6b1d" +[[package]] +name = "futures-macro" +version = "0.3.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e835b70203e41293343137df5c0664546da5745f82ec9b84d40be8336958447b" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "futures-task" +version = "0.3.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "037711b3d59c33004d3856fbdc83b99d4ff37a24768fa1be9ce3538a1cde4393" + +[[package]] +name = "futures-util" +version = "0.3.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "389ca41296e6190b48053de0321d02a77f32f8a5d2461dd38762c0593805c6d6" +dependencies = [ + "futures-core", + "futures-macro", + "futures-task", + "pin-project-lite", + "slab", +] + [[package]] name = "generic-array" version = "0.14.7" @@ -802,6 +881,12 @@ dependencies = [ "wasm-bindgen", ] +[[package]] +name = "lazy_static" +version = "1.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bbd2bcb4c963f2ddae06a2efc7e9f3591312473c50c6685e1f298068316e66fe" + [[package]] name = "leb128fmt" version = "0.1.0" @@ -836,6 +921,15 @@ version = "0.2.16" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b6d2cec3eae94f9f509c767b45932f1ada8350c4bdb85af2fcab4a3c14807981" +[[package]] +name = "libredox" +version = "0.1.16" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e02f3bb43d335493c96bf3fd3a321600bf6bd07ed34bc64118e9293bdffea46c" +dependencies = [ + "libc", +] + [[package]] name = "libsqlite3-sys" version = "0.28.0" @@ -877,6 +971,15 @@ dependencies = [ "imgref", ] +[[package]] +name = "matchers" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d1525a2a28c7f4fa0fc98bb91ae755d1e2d1505079e05539e35bc876b5d65ae9" +dependencies = [ + "regex-automata", +] + [[package]] name = "matrixmultiply" version = "0.3.10" @@ -921,7 +1024,7 @@ checksum = "50b7e5b27aa02a74bac8c3f23f448f8d87ff11f92d3aac1a6ed369ee08cc56c1" dependencies = [ "libc", "wasi", - "windows-sys", + "windows-sys 0.61.2", ] [[package]] @@ -979,6 +1082,15 @@ version = "0.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0676bb32a98c1a483ce53e500a81ad9c3d5b3f7c920c28c24e9cb0980d0b5bc8" +[[package]] +name = "nu-ansi-term" +version = "0.50.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5" +dependencies = [ + "windows-sys 0.61.2", +] + [[package]] name = "num" version = "0.4.3" @@ -1012,6 +1124,12 @@ dependencies = [ "num-traits", ] +[[package]] +name = "num-conv" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c6673768db2d862beb9b39a78fdcb1a69439615d5794a1be50caa9bc92c81967" + [[package]] name = "num-derive" version = "0.4.2" @@ -1070,6 +1188,12 @@ version = "1.21.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9f7c3e4beb33f85d45ae3e3a1792185706c8e16d043238c593331cc7cd313b50" +[[package]] +name = "option-ext" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "04744f49eae99ab78e0d5c0b603ab218f515ea8cfe5a456d7629ad883a3b6e7d" + [[package]] name = "owned_ttf_parser" version = "0.25.1" @@ -1129,6 +1253,12 @@ dependencies = [ "miniz_oxide", ] +[[package]] +name = "powerfmt" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "439ee305def115ba05938db6eb1644ff94165c5ab5e9420d1c1bcedbba909391" + [[package]] name = "ppv-lite86" version = "0.2.21" @@ -1351,7 +1481,7 @@ dependencies = [ "rand 0.9.4", "rand_chacha 0.9.0", "simd_helpers", - "thiserror", + "thiserror 2.0.18", "v_frame", "wasm-bindgen", ] @@ -1406,6 +1536,28 @@ dependencies = [ "bitflags", ] +[[package]] +name = "redox_users" +version = "0.4.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ba009ff324d1fc1b900bd1fdb31564febe58a8ccc8a6fdbb93b543d33b13ca43" +dependencies = [ + "getrandom 0.2.17", + "libredox", + "thiserror 1.0.69", +] + +[[package]] +name = "regex-automata" +version = "0.4.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6e1dd4122fc1595e8162618945476892eefca7b88c52820e74af6262213cae8f" +dependencies = [ + "aho-corasick", + "memchr", + "regex-syntax", +] + [[package]] name = "regex-syntax" version = "0.8.10" @@ -1442,7 +1594,7 @@ dependencies = [ "errno", "libc", "linux-raw-sys", - "windows-sys", + "windows-sys 0.61.2", ] [[package]] @@ -1551,12 +1703,53 @@ dependencies = [ "digest", ] +[[package]] +name = "sharded-slab" +version = "0.1.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f40ca3c46823713e0d4209592e8d6e826aa57e928f09752619fc696c499637f6" +dependencies = [ + "lazy_static", +] + [[package]] name = "shlex" version = "1.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0fda2ff0d084019ba4d7c6f371c95d8fd75ce3524c3cb8fb653a3023f6323e64" +[[package]] +name = "signal-hook" +version = "0.3.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d881a16cf4426aa584979d30bd82cb33429027e42122b169753d6ef1085ed6e2" +dependencies = [ + "libc", + "signal-hook-registry", +] + +[[package]] +name = "signal-hook-registry" +version = "1.4.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c4db69cba1110affc0e9f7bcd48bbf87b3f4fc7c61fc9155afd4c469eb3d6c1b" +dependencies = [ + "errno", + "libc", +] + +[[package]] +name = "signal-hook-tokio" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "213241f76fb1e37e27de3b6aa1b068a2c333233b59cca6634f634b80a27ecf1e" +dependencies = [ + "futures-core", + "libc", + "signal-hook", + "tokio", +] + [[package]] name = "simba" version = "0.8.1" @@ -1585,6 +1778,12 @@ dependencies = [ "quote", ] +[[package]] +name = "slab" +version = "0.4.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0c790de23124f9ab44544d7ac05d60440adc586479ce501c1d6d7da3cd8c9cf5" + [[package]] name = "smallvec" version = "1.15.1" @@ -1598,7 +1797,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3a766e1110788c36f4fa1c2b71b387a7815aa65f88ce0229841826633d93723e" dependencies = [ "libc", - "windows-sys", + "windows-sys 0.61.2", ] [[package]] @@ -1607,6 +1806,12 @@ version = "1.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6ce2be8dc25455e1f91df71bfa12ad37d7af1092ae736f3a6cd0e37bc7810596" +[[package]] +name = "symlink" +version = "0.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a7973cce6668464ea31f176d85b13c7ab3bba2cb3b77a2ed26abd7801688010a" + [[package]] name = "syn" version = "2.0.117" @@ -1628,7 +1833,16 @@ dependencies = [ "getrandom 0.4.2", "once_cell", "rustix", - "windows-sys", + "windows-sys 0.61.2", +] + +[[package]] +name = "thiserror" +version = "1.0.69" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b6aaf5339b578ea85b50e080feb250a3e8ae8cfcdff9a461c9ec2904bc923f52" +dependencies = [ + "thiserror-impl 1.0.69", ] [[package]] @@ -1637,7 +1851,18 @@ version = "2.0.18" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4288b5bcbc7920c07a1149a35cf9590a2aa808e0bc1eafaade0b80947865fbc4" dependencies = [ - "thiserror-impl", + "thiserror-impl 2.0.18", +] + +[[package]] +name = "thiserror-impl" +version = "1.0.69" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4fee6c4efc90059e10f81e6d42c60a18f76588c3d74cb83a0b242a2b6c7504c1" +dependencies = [ + "proc-macro2", + "quote", + "syn", ] [[package]] @@ -1651,6 +1876,15 @@ dependencies = [ "syn", ] +[[package]] +name = "thread_local" +version = "1.1.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f60246a4944f24f6e018aa17cdeffb7818b76356965d03b07d6a9886e8962185" +dependencies = [ + "cfg-if", +] + [[package]] name = "tiff" version = "0.11.3" @@ -1665,6 +1899,37 @@ dependencies = [ "zune-jpeg", ] +[[package]] +name = "time" +version = "0.3.47" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "743bd48c283afc0388f9b8827b976905fb217ad9e647fae3a379a9283c4def2c" +dependencies = [ + "deranged", + "itoa", + "num-conv", + "powerfmt", + "serde_core", + "time-core", + "time-macros", +] + +[[package]] +name = "time-core" +version = "0.1.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7694e1cfe791f8d31026952abf09c69ca6f6fa4e1a1229e18988f06a04a12dca" + +[[package]] +name = "time-macros" +version = "0.2.27" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2e70e4c5a0e0a8a4823ad65dfe1a6930e4f4d756dcd9dd7939022b5e8c501215" +dependencies = [ + "num-conv", + "time-core", +] + [[package]] name = "tinyvec" version = "1.11.0" @@ -1690,9 +1955,10 @@ dependencies = [ "libc", "mio", "pin-project-lite", + "signal-hook-registry", "socket2", "tokio-macros", - "windows-sys", + "windows-sys 0.61.2", ] [[package]] @@ -1739,6 +2005,19 @@ dependencies = [ "tracing-core", ] +[[package]] +name = "tracing-appender" +version = "0.2.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "050686193eb999b4bb3bc2acfa891a13da00f79734704c4b8b4ef1a10b368a3c" +dependencies = [ + "crossbeam-channel", + "symlink", + "thiserror 2.0.18", + "time", + "tracing-subscriber", +] + [[package]] name = "tracing-attributes" version = "0.1.31" @@ -1757,6 +2036,49 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "db97caf9d906fbde555dd62fa95ddba9eecfd14cb388e4f491a66d74cd5fb79a" dependencies = [ "once_cell", + "valuable", +] + +[[package]] +name = "tracing-log" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ee855f1f400bd0e5c02d150ae5de3840039a3f54b025156404e34c23c03f47c3" +dependencies = [ + "log", + "once_cell", + "tracing-core", +] + +[[package]] +name = "tracing-serde" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "704b1aeb7be0d0a84fc9828cae51dab5970fee5088f83d1dd7ee6f6246fc6ff1" +dependencies = [ + "serde", + "tracing-core", +] + +[[package]] +name = "tracing-subscriber" +version = "0.3.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cb7f578e5945fb242538965c2d0b04418d38ec25c79d160cd279bf0731c8d319" +dependencies = [ + "matchers", + "nu-ansi-term", + "once_cell", + "regex-automata", + "serde", + "serde_json", + "sharded-slab", + "smallvec", + "thread_local", + "tracing", + "tracing-core", + "tracing-log", + "tracing-serde", ] [[package]] @@ -1821,6 +2143,12 @@ dependencies = [ "wasm-bindgen", ] +[[package]] +name = "valuable" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ba73ea9cf16a25df0c8caa16c51acb937d5712a8429db78a3ee29d5dcacd3a65" + [[package]] name = "vcli-capture" version = "0.0.1" @@ -1828,7 +2156,7 @@ dependencies = [ "image", "screencapturekit", "tempfile", - "thiserror", + "thiserror 2.0.18", "tracing", "vcli-core", ] @@ -1841,11 +2169,41 @@ dependencies = [ "ryu", "serde", "serde_json", - "thiserror", + "thiserror 2.0.18", "unicode-normalization", "uuid", ] +[[package]] +name = "vcli-daemon" +version = "0.0.1" +dependencies = [ + "async-trait", + "crossbeam-channel", + "dirs", + "fs2", + "futures-util", + "serde", + "serde_json", + "signal-hook-tokio", + "tempfile", + "thiserror 2.0.18", + "tokio", + "tokio-test", + "tracing", + "tracing-appender", + "tracing-subscriber", + "uuid", + "vcli-capture", + "vcli-core", + "vcli-dsl", + "vcli-input", + "vcli-ipc", + "vcli-perception", + "vcli-runtime", + "vcli-store", +] + [[package]] name = "vcli-dsl" version = "0.0.1" @@ -1853,7 +2211,7 @@ dependencies = [ "proptest", "serde", "serde_json", - "thiserror", + "thiserror 2.0.18", "vcli-core", ] @@ -1868,7 +2226,7 @@ dependencies = [ "serde", "serde_json", "tempfile", - "thiserror", + "thiserror 2.0.18", "vcli-core", ] @@ -1881,7 +2239,7 @@ dependencies = [ "serde", "serde_json", "tempfile", - "thiserror", + "thiserror 2.0.18", "tokio", "tokio-test", "uuid", @@ -1899,7 +2257,7 @@ dependencies = [ "proptest", "serde", "serde_json", - "thiserror", + "thiserror 2.0.18", "vcli-core", ] @@ -1911,7 +2269,7 @@ dependencies = [ "proptest", "serde_json", "tempfile", - "thiserror", + "thiserror 2.0.18", "tracing", "vcli-capture", "vcli-core", @@ -1931,7 +2289,7 @@ dependencies = [ "serde_json", "sha2", "tempfile", - "thiserror", + "thiserror 2.0.18", "vcli-core", ] @@ -2075,12 +2433,43 @@ dependencies = [ "safe_arch", ] +[[package]] +name = "winapi" +version = "0.3.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5c839a674fcd7a98952e593242ea400abe93992746761e38641405d28b00f419" +dependencies = [ + "winapi-i686-pc-windows-gnu", + "winapi-x86_64-pc-windows-gnu", +] + +[[package]] +name = "winapi-i686-pc-windows-gnu" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ac3b87c63620426dd9b991e5ce0329eff545bccbbb34f3be09ff6fb6ab51b7b6" + +[[package]] +name = "winapi-x86_64-pc-windows-gnu" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "712e227841d057c1ee1cd2fb22fa7e5a5461ae8e48fa2ca79ec42cfc1931183f" + [[package]] name = "windows-link" version = "0.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f0805222e57f7521d6a62e36fa9163bc891acd422f971defe97d64e70d0a4fe5" +[[package]] +name = "windows-sys" +version = "0.48.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "677d2418bec65e3338edb076e806bc1ec15693c5d0104683f2efe857f61056a9" +dependencies = [ + "windows-targets", +] + [[package]] name = "windows-sys" version = "0.61.2" @@ -2090,6 +2479,63 @@ dependencies = [ "windows-link", ] +[[package]] +name = "windows-targets" +version = "0.48.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9a2fa6e2155d7247be68c096456083145c183cbbbc2764150dda45a87197940c" +dependencies = [ + "windows_aarch64_gnullvm", + "windows_aarch64_msvc", + "windows_i686_gnu", + "windows_i686_msvc", + "windows_x86_64_gnu", + "windows_x86_64_gnullvm", + "windows_x86_64_msvc", +] + +[[package]] +name = "windows_aarch64_gnullvm" +version = "0.48.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2b38e32f0abccf9987a4e3079dfb67dcd799fb61361e53e2882c3cbaf0d905d8" + +[[package]] +name = "windows_aarch64_msvc" +version = "0.48.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dc35310971f3b2dbbf3f0690a219f40e2d9afcf64f9ab7cc1be722937c26b4bc" + +[[package]] +name = "windows_i686_gnu" +version = "0.48.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a75915e7def60c94dcef72200b9a8e58e5091744960da64ec734a6c6e9b3743e" + +[[package]] +name = "windows_i686_msvc" +version = "0.48.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8f55c233f70c4b27f66c523580f78f1004e8b5a8b659e05a4eb49d4166cca406" + +[[package]] +name = "windows_x86_64_gnu" +version = "0.48.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "53d40abd2583d23e4718fddf1ebec84dbff8381c07cae67ff7768bbf19c6718e" + +[[package]] +name = "windows_x86_64_gnullvm" +version = "0.48.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0b7b52767868a23d5bab768e390dc5f5c55825b6d30b86c844ff2dc7414044cc" + +[[package]] +name = "windows_x86_64_msvc" +version = "0.48.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ed94fce61571a4006852b7389a063ab983c02eb1bb37b47f8272ce92d06d9538" + [[package]] name = "wit-bindgen" version = "0.51.0" diff --git a/Cargo.toml b/Cargo.toml index bda3c0f..4c6bcad 100644 --- a/Cargo.toml +++ b/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" diff --git a/crates/vcli-daemon/Cargo.toml b/crates/vcli-daemon/Cargo.toml new file mode 100644 index 0000000..e80fb0f --- /dev/null +++ b/crates/vcli-daemon/Cargo.toml @@ -0,0 +1,46 @@ +[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 } + +[dev-dependencies] +tempfile = { workspace = true } +tokio-test = { workspace = true } +tokio = { workspace = true, features = ["test-util"] } diff --git a/crates/vcli-daemon/src/bin/vcli-daemon.rs b/crates/vcli-daemon/src/bin/vcli-daemon.rs new file mode 100644 index 0000000..f90f586 --- /dev/null +++ b/crates/vcli-daemon/src/bin/vcli-daemon.rs @@ -0,0 +1,6 @@ +//! Thin binary wrapper. All real work lives in `vcli_daemon::run::run_foreground`. + +fn main() { + // Placeholder until Task 12 wires real argv + run_foreground(). + eprintln!("vcli-daemon: not yet wired — see Task 12"); +} diff --git a/crates/vcli-daemon/src/lib.rs b/crates/vcli-daemon/src/lib.rs new file mode 100644 index 0000000..f0f6944 --- /dev/null +++ b/crates/vcli-daemon/src/lib.rs @@ -0,0 +1,14 @@ +//! vcli-daemon — the binary that wires IPC, the store, and the scheduler. +//! +//! See the v0 design spec at `docs/superpowers/specs/2026-04-16-vcli-design.md` +//! §Threading model, §IPC, and §Restart semantics for the authoritative behavior. +//! +//! The library surface exists so that the binary in `src/bin/vcli-daemon.rs` is +//! a thin `main`, and so that integration tests can exercise the assembly with +//! mock capture + input backends. + +#![forbid(unsafe_code)] +#![deny(rust_2018_idioms)] +#![warn(missing_docs)] +#![warn(clippy::pedantic)] +#![allow(clippy::module_name_repetitions)] From 3aa6fab68544283bfd2400699308d132f99ea935 Mon Sep 17 00:00:00 2001 From: Yixiang Zeng Date: Mon, 20 Apr 2026 15:34:25 -0700 Subject: [PATCH 05/25] vcli-daemon: DaemonError + DaemonResult MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit thiserror-based enum covering startup, pidfile, paths, logging, store, ipc, io, and invalid-program branches. `code()` maps each to a stable vcli_core::ErrorCode (InvalidProgram → InvalidProgram, rest → Internal). Co-Authored-By: Claude Opus 4.7 (1M context) --- crates/vcli-daemon/src/error.rs | 102 ++++++++++++++++++++++++++++++++ crates/vcli-daemon/src/lib.rs | 4 ++ 2 files changed, 106 insertions(+) create mode 100644 crates/vcli-daemon/src/error.rs diff --git a/crates/vcli-daemon/src/error.rs b/crates/vcli-daemon/src/error.rs new file mode 100644 index 0000000..b152706 --- /dev/null +++ b/crates/vcli-daemon/src/error.rs @@ -0,0 +1,102 @@ +//! Top-level daemon error. Maps to `vcli_core::ErrorCode` at the IPC boundary. + +use std::io; +use std::path::PathBuf; + +use thiserror::Error; +use vcli_core::ErrorCode; + +/// Convenience alias. +pub type DaemonResult = Result; + +/// Errors the daemon produces outside the per-request handler path. +#[derive(Debug, Error)] +pub enum DaemonError { + /// PID file already held by a live process. + #[error("another vcli-daemon is already running (pid {pid}, lockfile {path})")] + AlreadyRunning { + /// Owning PID. + pid: u32, + /// Absolute pidfile path. + path: PathBuf, + }, + + /// Could not acquire / write the pidfile. + #[error("pidfile {path}: {source}")] + Pidfile { + /// Absolute path. + path: PathBuf, + /// Underlying IO cause. + #[source] + source: io::Error, + }, + + /// Could not resolve the daemon's data root, socket path, or log dir. + #[error("path resolution: {0}")] + Paths(String), + + /// Tracing subscriber failed to install. + #[error("logging init: {0}")] + Logging(String), + + /// Store open / migrate failed. + #[error("store: {0}")] + Store(#[from] vcli_store::StoreError), + + /// IPC transport setup / serve failure. + #[error("ipc: {0}")] + Ipc(#[from] vcli_ipc::IpcError), + + /// Generic IO error during startup / shutdown. + #[error("io: {0}")] + Io(#[from] io::Error), + + /// DSL validation of a submitted program failed before scheduler touched it. + #[error("invalid program: {0}")] + InvalidProgram(String), +} + +impl DaemonError { + /// Stable error code for IPC responses. + #[must_use] + pub fn code(&self) -> ErrorCode { + match self { + Self::InvalidProgram(_) => ErrorCode::InvalidProgram, + Self::AlreadyRunning { .. } + | Self::Pidfile { .. } + | Self::Paths(_) + | Self::Logging(_) + | Self::Store(_) + | Self::Ipc(_) + | Self::Io(_) => ErrorCode::Internal, + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn invalid_program_maps_to_invalid_program_code() { + let e = DaemonError::InvalidProgram("bad path".into()); + assert_eq!(e.code(), ErrorCode::InvalidProgram); + } + + #[test] + fn paths_maps_to_internal() { + let e = DaemonError::Paths("no home".into()); + assert_eq!(e.code(), ErrorCode::Internal); + } + + #[test] + fn already_running_display_contains_pid_and_path() { + let e = DaemonError::AlreadyRunning { + pid: 42, + path: "/tmp/x.pid".into(), + }; + let s = e.to_string(); + assert!(s.contains("42"), "{s}"); + assert!(s.contains("/tmp/x.pid"), "{s}"); + } +} diff --git a/crates/vcli-daemon/src/lib.rs b/crates/vcli-daemon/src/lib.rs index f0f6944..4ee5db4 100644 --- a/crates/vcli-daemon/src/lib.rs +++ b/crates/vcli-daemon/src/lib.rs @@ -12,3 +12,7 @@ #![warn(missing_docs)] #![warn(clippy::pedantic)] #![allow(clippy::module_name_repetitions)] + +pub mod error; + +pub use error::{DaemonError, DaemonResult}; From 3b46ba02f36490f50ccc4eadfedb34fd28241ad2 Mon Sep 17 00:00:00 2001 From: Yixiang Zeng Date: Mon, 20 Apr 2026 15:35:34 -0700 Subject: [PATCH 06/25] vcli-daemon: Config path resolution for macOS and Linux from_platform_defaults() resolves data_root / log_dir per platform (Library on macOS, XDG on Linux) and socket via vcli_ipc. with_roots() gives tests and alt installs explicit overrides. ensure_dirs() idempotently mkdir -p's data_root and log_dir. Co-Authored-By: Claude Opus 4.7 (1M context) --- crates/vcli-daemon/src/config.rs | 162 +++++++++++++++++++++++++++++++ crates/vcli-daemon/src/lib.rs | 4 + 2 files changed, 166 insertions(+) create mode 100644 crates/vcli-daemon/src/config.rs diff --git a/crates/vcli-daemon/src/config.rs b/crates/vcli-daemon/src/config.rs new file mode 100644 index 0000000..90ac6cc --- /dev/null +++ b/crates/vcli-daemon/src/config.rs @@ -0,0 +1,162 @@ +//! Resolved on-disk layout. One `Config` value, produced at startup, threaded +//! everywhere that needs a path. Every caller is driven by the struct — nothing +//! touches `$HOME` directly so tests stay hermetic. +//! +//! Layout (spec §Persistence → Data layout): +//! `data_root` = `~/Library/Application Support/vcli` (macOS) +//! | `~/.local/share/vcli` (Linux / XDG) +//! db = `/vcli.db` +//! pidfile = `/daemon.pid` +//! socket = `vcli_ipc::default_socket_path()` +//! `log_dir` = `~/Library/Logs/vcli` (macOS) +//! | `~/.cache/vcli/logs` (Linux / XDG) +//! `log_file` = `/daemon.log` (rotated daily, 7-day retention) + +use std::path::{Path, PathBuf}; + +use vcli_ipc::SocketPath; + +use crate::error::{DaemonError, DaemonResult}; + +/// Every path the daemon needs to know. Built once at startup. +#[derive(Debug, Clone)] +pub struct Config { + /// Directory holding `vcli.db`, `daemon.pid`, and the assets tree. + pub data_root: PathBuf, + /// Resolved socket path + its resolution origin (for `vcli health`). + pub socket: SocketPath, + /// Directory for rotating log files. + pub log_dir: PathBuf, +} + +impl Config { + /// Resolve from the platform defaults. macOS uses Apple locations; Linux uses XDG. + /// + /// # Errors + /// [`DaemonError::Paths`] if `$HOME` (or its platform equivalent) cannot be resolved. + pub fn from_platform_defaults() -> DaemonResult { + let data_root = platform_data_root()?; + let log_dir = platform_log_dir()?; + let socket = vcli_ipc::default_socket_path() + .map_err(|e| DaemonError::Paths(format!("socket path: {e}")))?; + Ok(Self { + data_root, + socket, + log_dir, + }) + } + + /// Build a config rooted at an explicit directory — used by tests and + /// by anyone wiring a non-default install. + #[must_use] + pub fn with_roots( + data_root: impl Into, + log_dir: impl Into, + socket: SocketPath, + ) -> Self { + Self { + data_root: data_root.into(), + socket, + log_dir: log_dir.into(), + } + } + + /// Absolute pidfile path (`/daemon.pid`). + #[must_use] + pub fn pidfile_path(&self) -> PathBuf { + self.data_root.join("daemon.pid") + } + + /// Absolute log file path (`/daemon.log`). + #[must_use] + pub fn log_file_path(&self) -> PathBuf { + self.log_dir.join("daemon.log") + } +} + +#[cfg(target_os = "macos")] +fn platform_data_root() -> DaemonResult { + let home = dirs::home_dir().ok_or_else(|| DaemonError::Paths("no home dir".into()))?; + Ok(home.join("Library").join("Application Support").join("vcli")) +} + +#[cfg(target_os = "macos")] +fn platform_log_dir() -> DaemonResult { + let home = dirs::home_dir().ok_or_else(|| DaemonError::Paths("no home dir".into()))?; + Ok(home.join("Library").join("Logs").join("vcli")) +} + +#[cfg(not(target_os = "macos"))] +fn platform_data_root() -> DaemonResult { + let d = dirs::data_local_dir().ok_or_else(|| DaemonError::Paths("no XDG_DATA_HOME".into()))?; + Ok(d.join("vcli")) +} + +#[cfg(not(target_os = "macos"))] +fn platform_log_dir() -> DaemonResult { + let d = dirs::cache_dir().ok_or_else(|| DaemonError::Paths("no XDG_CACHE_HOME".into()))?; + Ok(d.join("vcli").join("logs")) +} + +/// Ensure every directory in `cfg` exists. Idempotent. +/// +/// # Errors +/// IO errors from `fs::create_dir_all`. +pub fn ensure_dirs(cfg: &Config) -> DaemonResult<()> { + mkdir_p(&cfg.data_root)?; + mkdir_p(&cfg.log_dir)?; + Ok(()) +} + +fn mkdir_p(p: &Path) -> DaemonResult<()> { + std::fs::create_dir_all(p).map_err(DaemonError::Io) +} + +#[cfg(test)] +mod tests { + use super::*; + use tempfile::tempdir; + use vcli_ipc::socket_path::SocketPathOrigin; + + fn fake_sock(p: &Path) -> SocketPath { + SocketPath { + path: p.to_path_buf(), + origin: SocketPathOrigin::Override, + } + } + + #[test] + fn with_roots_builds_a_usable_config() { + let d = tempdir().unwrap(); + let cfg = Config::with_roots( + d.path().join("data"), + d.path().join("logs"), + fake_sock(&d.path().join("vcli.sock")), + ); + assert_eq!(cfg.pidfile_path(), d.path().join("data").join("daemon.pid")); + assert_eq!(cfg.log_file_path(), d.path().join("logs").join("daemon.log")); + } + + #[test] + fn ensure_dirs_creates_nested_paths() { + let d = tempdir().unwrap(); + let cfg = Config::with_roots( + d.path().join("a").join("b"), + d.path().join("c").join("d"), + fake_sock(&d.path().join("vcli.sock")), + ); + ensure_dirs(&cfg).unwrap(); + assert!(cfg.data_root.is_dir()); + assert!(cfg.log_dir.is_dir()); + } + + #[test] + #[cfg(target_os = "macos")] + fn platform_defaults_use_library_on_macos() { + let got = platform_data_root().unwrap(); + assert!( + got.ends_with("Library/Application Support/vcli"), + "{got:?}" + ); + } +} diff --git a/crates/vcli-daemon/src/lib.rs b/crates/vcli-daemon/src/lib.rs index 4ee5db4..776df73 100644 --- a/crates/vcli-daemon/src/lib.rs +++ b/crates/vcli-daemon/src/lib.rs @@ -16,3 +16,7 @@ pub mod error; pub use error::{DaemonError, DaemonResult}; + +pub mod config; + +pub use config::{ensure_dirs, Config}; From 25ea88730a9ff034a45343af255bf1ce3a3f400c Mon Sep 17 00:00:00 2001 From: Yixiang Zeng Date: Mon, 20 Apr 2026 15:36:37 -0700 Subject: [PATCH 07/25] vcli-daemon: PidFile with advisory fs2 lock PidFile::acquire opens , takes fs2 exclusive lock, writes current PID. Collision returns DaemonError::AlreadyRunning with the owning PID read from the file. release() consumes self, unlocks + unlinks; Drop covers the panic / fast-exit path the same way. Stale PIDs after SIGKILL are handled automatically: the kernel releases the flock on fd close, so the next daemon acquires cleanly without needing a kill-check. Co-Authored-By: Claude Opus 4.7 (1M context) --- crates/vcli-daemon/src/lib.rs | 4 + crates/vcli-daemon/src/pidfile.rs | 191 ++++++++++++++++++++++++++++++ 2 files changed, 195 insertions(+) create mode 100644 crates/vcli-daemon/src/pidfile.rs diff --git a/crates/vcli-daemon/src/lib.rs b/crates/vcli-daemon/src/lib.rs index 776df73..dc2a25c 100644 --- a/crates/vcli-daemon/src/lib.rs +++ b/crates/vcli-daemon/src/lib.rs @@ -20,3 +20,7 @@ pub use error::{DaemonError, DaemonResult}; pub mod config; pub use config::{ensure_dirs, Config}; + +pub mod pidfile; + +pub use pidfile::PidFile; diff --git a/crates/vcli-daemon/src/pidfile.rs b/crates/vcli-daemon/src/pidfile.rs new file mode 100644 index 0000000..25bd2de --- /dev/null +++ b/crates/vcli-daemon/src/pidfile.rs @@ -0,0 +1,191 @@ +//! Advisory pidfile. On acquire: +//! 1. Open (create if missing) `` read-write. +//! 2. `fs2::try_lock_exclusive` — if another process holds the lock, read +//! its contents and return [`DaemonError::AlreadyRunning`]. +//! 3. Rewind, truncate, write current PID + `\n`. +//! +//! On drop / explicit `release()`: unlock + unlink (best effort). +//! +//! The advisory lock is flock(2)-based via `fs2`; it's unaffected by file +//! copies and released if the holder dies without cleanup (kernel releases +//! the lock on fd close). That covers the SIGKILL case: a stale PID file on +//! disk has no lock, so the next daemon acquires it cleanly. + +use std::fs::{File, OpenOptions}; +use std::io::{Read, Seek, SeekFrom, Write}; +use std::path::{Path, PathBuf}; + +use fs2::FileExt; + +use crate::error::{DaemonError, DaemonResult}; + +/// Owning handle on the pidfile. Lock released when dropped. +pub struct PidFile { + path: PathBuf, + file: Option, +} + +impl std::fmt::Debug for PidFile { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("PidFile") + .field("path", &self.path) + .finish_non_exhaustive() + } +} + +impl PidFile { + /// Try to acquire the lock and write the current PID. + /// + /// # Errors + /// - [`DaemonError::AlreadyRunning`] if another process holds the lock. + /// - [`DaemonError::Pidfile`] on any underlying IO failure. + pub fn acquire(path: impl AsRef) -> DaemonResult { + let path = path.as_ref().to_path_buf(); + if let Some(parent) = path.parent() { + std::fs::create_dir_all(parent).map_err(|e| DaemonError::Pidfile { + path: path.clone(), + source: e, + })?; + } + let mut file = OpenOptions::new() + .read(true) + .write(true) + .create(true) + .truncate(false) + .open(&path) + .map_err(|e| DaemonError::Pidfile { + path: path.clone(), + source: e, + })?; + + if file.try_lock_exclusive().is_err() { + let mut s = String::new(); + let _ = file.read_to_string(&mut s); + let pid: u32 = s.trim().parse().unwrap_or(0); + return Err(DaemonError::AlreadyRunning { pid, path }); + } + + file.set_len(0).map_err(|e| DaemonError::Pidfile { + path: path.clone(), + source: e, + })?; + file.seek(SeekFrom::Start(0)) + .map_err(|e| DaemonError::Pidfile { + path: path.clone(), + source: e, + })?; + writeln!(file, "{}", std::process::id()).map_err(|e| DaemonError::Pidfile { + path: path.clone(), + source: e, + })?; + file.flush().map_err(|e| DaemonError::Pidfile { + path: path.clone(), + source: e, + })?; + + Ok(Self { + path, + file: Some(file), + }) + } + + /// The path this lock is anchored to. + #[must_use] + pub fn path(&self) -> &Path { + &self.path + } + + /// PID recorded inside the file (always this process on a live handle). + #[must_use] + pub fn pid(&self) -> u32 { + std::process::id() + } + + /// Explicitly release (unlock + unlink). Drop does the same, but in an + /// ordered shutdown we want errors logged rather than swallowed. + /// + /// # Errors + /// IO errors during unlink. + pub fn release(mut self) -> DaemonResult<()> { + if let Some(file) = self.file.take() { + let _ = fs2::FileExt::unlock(&file); + drop(file); + } + std::fs::remove_file(&self.path).or_else(|e| { + if e.kind() == std::io::ErrorKind::NotFound { + Ok(()) + } else { + Err(DaemonError::Pidfile { + path: self.path.clone(), + source: e, + }) + } + }) + } +} + +impl Drop for PidFile { + fn drop(&mut self) { + if let Some(file) = self.file.take() { + let _ = fs2::FileExt::unlock(&file); + } + let _ = std::fs::remove_file(&self.path); + } +} + +#[cfg(test)] +mod tests { + use super::*; + use tempfile::tempdir; + + #[test] + fn acquire_creates_pidfile_with_current_pid() { + let d = tempdir().unwrap(); + let p = d.path().join("daemon.pid"); + let lock = PidFile::acquire(&p).unwrap(); + assert!(p.exists()); + let s = std::fs::read_to_string(&p).unwrap(); + assert_eq!(s.trim().parse::().unwrap(), std::process::id()); + drop(lock); + } + + #[test] + fn second_acquire_in_same_process_returns_already_running() { + let d = tempdir().unwrap(); + let p = d.path().join("daemon.pid"); + let _first = PidFile::acquire(&p).unwrap(); + let err = PidFile::acquire(&p).unwrap_err(); + match err { + DaemonError::AlreadyRunning { pid, .. } => assert_eq!(pid, std::process::id()), + other => panic!("wrong err: {other:?}"), + } + } + + #[test] + fn release_unlinks_pidfile() { + let d = tempdir().unwrap(); + let p = d.path().join("daemon.pid"); + let lock = PidFile::acquire(&p).unwrap(); + lock.release().unwrap(); + assert!(!p.exists()); + } + + #[test] + fn drop_unlinks_pidfile() { + let d = tempdir().unwrap(); + let p = d.path().join("daemon.pid"); + { + let _lock = PidFile::acquire(&p).unwrap(); + assert!(p.exists()); + } + assert!(!p.exists()); + } + + #[test] + fn parent_dir_is_created_on_acquire() { + let d = tempdir().unwrap(); + let p = d.path().join("nested").join("dir").join("daemon.pid"); + let _lock = PidFile::acquire(&p).unwrap(); + assert!(p.exists()); + } +} From 0e3739820392d07f2a59f6205d1625df93077b71 Mon Sep 17 00:00:00 2001 From: Yixiang Zeng Date: Mon, 20 Apr 2026 15:37:30 -0700 Subject: [PATCH 08/25] vcli-daemon: tracing subscriber with daily rotation logging::init(log_dir) installs a JSON tracing subscriber whose file writer is tracing-appender's daily roller (`daemon.log.YYYY-MM-DD`). Respects RUST_LOG via EnvFilter; defaults to info. Returns a LogGuard the caller must keep alive to avoid losing buffered records. Co-Authored-By: Claude Opus 4.7 (1M context) --- crates/vcli-daemon/src/lib.rs | 4 ++ crates/vcli-daemon/src/logging.rs | 71 +++++++++++++++++++++++++++++++ 2 files changed, 75 insertions(+) create mode 100644 crates/vcli-daemon/src/logging.rs diff --git a/crates/vcli-daemon/src/lib.rs b/crates/vcli-daemon/src/lib.rs index dc2a25c..8da0dbd 100644 --- a/crates/vcli-daemon/src/lib.rs +++ b/crates/vcli-daemon/src/lib.rs @@ -24,3 +24,7 @@ pub use config::{ensure_dirs, Config}; pub mod pidfile; pub use pidfile::PidFile; + +pub mod logging; + +pub use logging::{init as init_logging, LogGuard}; diff --git a/crates/vcli-daemon/src/logging.rs b/crates/vcli-daemon/src/logging.rs new file mode 100644 index 0000000..8ed097a --- /dev/null +++ b/crates/vcli-daemon/src/logging.rs @@ -0,0 +1,71 @@ +//! tracing-subscriber setup with a daily-rolling JSON file + stderr fallback. +//! +//! Returns a `WorkerGuard` the caller must keep alive for the lifetime of the +//! process — dropping it flushes any buffered records. Emits JSON into +//! `/daemon.log.`; `tracing-appender` handles rotation and +//! keeps the rolling logic out of the daemon. + +use std::path::Path; + +use tracing_appender::rolling; +use tracing_subscriber::{fmt, prelude::*, EnvFilter}; + +use crate::error::{DaemonError, DaemonResult}; + +/// Non-drop-until-shutdown handle to the background log writer thread. +pub struct LogGuard(#[allow(dead_code)] tracing_appender::non_blocking::WorkerGuard); + +/// Install the tracing subscriber. Respects `RUST_LOG` via `EnvFilter`; defaults +/// to `info` if unset. +/// +/// # Errors +/// Returns [`DaemonError::Logging`] if the subscriber cannot be installed +/// (typically: a subscriber is already set in the current process). +pub fn init(log_dir: &Path) -> DaemonResult { + let file_appender = rolling::daily(log_dir, "daemon.log"); + let (nb, guard) = tracing_appender::non_blocking(file_appender); + + let env = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info")); + + let subscriber = tracing_subscriber::registry().with(env).with( + fmt::layer() + .json() + .with_writer(nb) + .with_current_span(false) + .with_span_list(false), + ); + + subscriber + .try_init() + .map_err(|e| DaemonError::Logging(format!("{e}")))?; + + Ok(LogGuard(guard)) +} + +#[cfg(test)] +mod tests { + use super::*; + use tempfile::tempdir; + + /// Smoke test: constructing the appender does not panic and produces a file + /// on first log write. We cannot call `init()` from a test (global + /// subscriber), so this exercises the appender alone. + #[test] + fn daily_appender_writes_to_log_dir() { + use std::io::Write; + let d = tempdir().unwrap(); + let app = rolling::daily(d.path(), "daemon.log"); + let (mut nb, guard) = tracing_appender::non_blocking(app); + nb.write_all(b"hello\n").unwrap(); + nb.flush().unwrap(); + drop(guard); + let mut found = false; + for entry in std::fs::read_dir(d.path()).unwrap() { + let e = entry.unwrap(); + if e.file_name().to_string_lossy().starts_with("daemon.log") { + found = true; + } + } + assert!(found, "expected a daemon.log.* file"); + } +} From f57741a118955f94d808a14c76f21c7672180729 Mon Sep 17 00:00:00 2001 From: Yixiang Zeng Date: Mon, 20 Apr 2026 15:38:22 -0700 Subject: [PATCH 09/25] =?UTF-8?q?vcli-daemon:=20CommandChannel=20tokio?= =?UTF-8?q?=E2=86=94scheduler=20bridge?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit new_channels() returns a bundle of endpoints: a bounded crossbeam cmd_tx for tokio→scheduler commands, an unbounded event_rx/sched_event_tx pair for scheduler→tokio events, and a tokio broadcast::Sender (capacity 1024) for fanning persisted events to IPC subscribers. Also adds build.rs with the same macOS libswift_Concurrency rpath vcli-runtime uses — required because vcli-daemon pulls in vcli-capture transitively. Co-Authored-By: Claude Opus 4.7 (1M context) --- crates/vcli-daemon/build.rs | 13 +++++ crates/vcli-daemon/src/bridge.rs | 97 ++++++++++++++++++++++++++++++++ crates/vcli-daemon/src/lib.rs | 4 ++ 3 files changed, 114 insertions(+) create mode 100644 crates/vcli-daemon/build.rs create mode 100644 crates/vcli-daemon/src/bridge.rs diff --git a/crates/vcli-daemon/build.rs b/crates/vcli-daemon/build.rs new file mode 100644 index 0000000..885803f --- /dev/null +++ b/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}"); + } +} diff --git a/crates/vcli-daemon/src/bridge.rs b/crates/vcli-daemon/src/bridge.rs new file mode 100644 index 0000000..ddc1c03 --- /dev/null +++ b/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, + /// Broadcast fanning out persisted events to IPC subscribers. + pub event_tx: broadcast::Sender, +} + +/// 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, + Receiver, + Sender, +) { + let (cmd_tx, cmd_rx) = bounded::(CMD_CAPACITY); + let (sched_event_tx, event_rx) = unbounded::(); + let (bcast_tx, _) = broadcast::channel::(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); + } +} diff --git a/crates/vcli-daemon/src/lib.rs b/crates/vcli-daemon/src/lib.rs index 8da0dbd..9747605 100644 --- a/crates/vcli-daemon/src/lib.rs +++ b/crates/vcli-daemon/src/lib.rs @@ -28,3 +28,7 @@ pub use pidfile::PidFile; pub mod logging; pub use logging::{init as init_logging, LogGuard}; + +pub mod bridge; + +pub use bridge::{new_channels, CommandChannel, SchedulerCommand, EVENT_BROADCAST_CAPACITY}; From ab0faa28973293edff6e61647466774a08c49ac4 Mon Sep 17 00:00:00 2001 From: Yixiang Zeng Date: Mon, 20 Apr 2026 15:39:38 -0700 Subject: [PATCH 10/25] =?UTF-8?q?vcli-daemon:=20EventSink=20=E2=80=94=20pe?= =?UTF-8?q?rsist=20then=20broadcast?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit spawn_event_pump drains the scheduler's crossbeam Receiver on a tokio spawn_blocking task. For program-scoped events it calls Store::append_event before broadcasting; daemon-scoped events (DaemonStarted/Stopped/Pressure/StreamDropped/TickFrameSkipped/ CapturePermissionMissing) bypass the DB and broadcast only (Decision 1.7). Co-Authored-By: Claude Opus 4.7 (1M context) --- crates/vcli-daemon/src/lib.rs | 4 + crates/vcli-daemon/src/persist.rs | 131 ++++++++++++++++++++++++++++++ 2 files changed, 135 insertions(+) create mode 100644 crates/vcli-daemon/src/persist.rs diff --git a/crates/vcli-daemon/src/lib.rs b/crates/vcli-daemon/src/lib.rs index 9747605..a7cbe82 100644 --- a/crates/vcli-daemon/src/lib.rs +++ b/crates/vcli-daemon/src/lib.rs @@ -32,3 +32,7 @@ pub use logging::{init as init_logging, LogGuard}; pub mod bridge; pub use bridge::{new_channels, CommandChannel, SchedulerCommand, EVENT_BROADCAST_CAPACITY}; + +pub mod persist; + +pub use persist::{program_id_of, spawn_event_pump}; diff --git a/crates/vcli-daemon/src/persist.rs b/crates/vcli-daemon/src/persist.rs new file mode 100644 index 0000000..66c380e --- /dev/null +++ b/crates/vcli-daemon/src/persist.rs @@ -0,0 +1,131 @@ +//! Event pipeline. The scheduler pushes `vcli_core::Event`s into +//! `crossbeam_channel::Receiver`. A dedicated tokio task drains that +//! receiver, calls `store.append_event` for every program-scoped variant +//! (`DaemonStarted` / `DaemonStopped` / `DaemonPressure` / `StreamDropped` +//! etc. are broadcast-only — no program id, no DB row), then forwards the +//! event into the `broadcast::Sender`. +//! +//! Ordering invariant (Decision 1.7): an event is in the DB before any +//! subscriber can see it. + +use std::sync::{Arc, Mutex}; + +use crossbeam_channel::Receiver; +use tokio::sync::broadcast; +use tracing::{error, trace}; +use vcli_core::{Event, EventData, ProgramId}; +use vcli_store::Store; + +/// Drain `event_rx` on a background tokio task. Returns a `JoinHandle` so the +/// run loop can await shutdown. +/// +/// # Panics +/// The background task panics if the store mutex is poisoned. +pub fn spawn_event_pump( + store: Arc>, + event_rx: Receiver, + broadcast_tx: broadcast::Sender, +) -> tokio::task::JoinHandle<()> { + tokio::task::spawn_blocking(move || { + while let Ok(ev) = event_rx.recv() { + if let Some(pid) = program_id_of(&ev.data) { + let mut guard = store.lock().expect("store mutex poisoned"); + if let Err(e) = guard.append_event(pid, &ev) { + error!(error = %e, "failed to append event"); + } + } + let _ = broadcast_tx.send(ev); + trace!("event pump iteration"); + } + }) +} + +/// Extract the program id from an event payload, if any. Returns `None` for +/// daemon-scoped events that are broadcast-only. +#[must_use] +pub fn program_id_of(d: &EventData) -> Option { + match d { + EventData::ProgramSubmitted { program_id, .. } + | EventData::ProgramStateChanged { program_id, .. } + | EventData::ProgramCompleted { program_id, .. } + | EventData::ProgramFailed { program_id, .. } + | EventData::ProgramResumed { program_id, .. } + | EventData::WatchFired { program_id, .. } + | EventData::ActionDispatched { program_id, .. } + | EventData::ActionDeferred { program_id, .. } => Some(*program_id), + EventData::DaemonStarted { .. } + | EventData::DaemonStopped + | EventData::DaemonPressure { .. } + | EventData::StreamDropped { .. } + | EventData::TickFrameSkipped { .. } + | EventData::CapturePermissionMissing { .. } => None, + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crossbeam_channel::unbounded; + use tempfile::tempdir; + use vcli_core::EventData; + use vcli_store::{NewProgram, Store}; + + #[tokio::test] + async fn program_event_is_appended_before_broadcast() { + let d = tempdir().unwrap(); + let (mut store, _) = Store::open(d.path()).unwrap(); + let pid = ProgramId::new(); + store + .insert_program(&NewProgram { + id: pid, + name: "x", + source_json: "{}", + state: vcli_core::ProgramState::Pending, + submitted_at: 0, + labels_json: "{}", + }) + .unwrap(); + let store = Arc::new(Mutex::new(store)); + let (sched_tx, sched_rx) = unbounded::(); + let (bcast_tx, mut bcast_rx) = broadcast::channel::(16); + + let pump = spawn_event_pump(store.clone(), sched_rx, bcast_tx); + + let ev = Event { + at: 7, + data: EventData::ProgramCompleted { + program_id: pid, + emit: None, + }, + }; + sched_tx.send(ev.clone()).unwrap(); + let got = bcast_rx.recv().await.unwrap(); + assert_eq!(got, ev); + let rows = store.lock().unwrap().stream_events(0, 10).unwrap(); + assert_eq!(rows.len(), 1); + assert_eq!(rows[0].type_tag, "program.completed"); + + drop(sched_tx); + pump.await.unwrap(); + } + + #[test] + fn daemon_started_has_no_program_id() { + let d = EventData::DaemonStarted { + version: "0.0.1".into(), + }; + assert!(program_id_of(&d).is_none()); + } + + #[test] + fn state_changed_has_program_id() { + let p = ProgramId::new(); + let d = EventData::ProgramStateChanged { + program_id: p, + from: vcli_core::ProgramState::Waiting, + to: vcli_core::ProgramState::Running, + reason: "trigger".into(), + }; + assert_eq!(program_id_of(&d), Some(p)); + } +} From d55729b4e50d106e543f87cb241a60b6c4333f83 Mon Sep 17 00:00:00 2001 From: Yixiang Zeng Date: Mon, 20 Apr 2026 15:41:36 -0700 Subject: [PATCH 11/25] vcli-daemon: DaemonHandler scaffold with Health + Shutdown MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit DaemonHandler carries Arc>, CommandChannel, started_at Instant, and an oneshot shutdown_tx. handle() dispatches to handle_health / handle_shutdown; every other op returns Internal "not yet wired" until the follow-up 8b–8g subtasks land each op. Tests live on a test_support::fresh_handler() fixture that keeps the tempdir alive and exposes the bridge's cmd_rx for later subtasks. Co-Authored-By: Claude Opus 4.7 (1M context) --- crates/vcli-daemon/src/handler.rs | 158 ++++++++++++++++++++++++++++++ crates/vcli-daemon/src/lib.rs | 4 + 2 files changed, 162 insertions(+) create mode 100644 crates/vcli-daemon/src/handler.rs diff --git a/crates/vcli-daemon/src/handler.rs b/crates/vcli-daemon/src/handler.rs new file mode 100644 index 0000000..1abe4f7 --- /dev/null +++ b/crates/vcli-daemon/src/handler.rs @@ -0,0 +1,158 @@ +//! `DaemonHandler` implements `vcli_ipc::Handler`. One `impl` method per +//! [`RequestOp`] variant. The handler owns clones of the bridge endpoints and +//! an `Arc>` (sync — reached via `spawn_blocking` when we need to +//! call `SQLite` from inside an async method). + +use std::sync::{Arc, Mutex}; +use std::time::Instant; + +use async_trait::async_trait; +use tokio::sync::oneshot; +use vcli_core::{ErrorCode, ErrorPayload}; +use vcli_ipc::{ + Handler, IpcResult, RequestId, RequestOp, Response, StreamFrame, StreamKind, StreamSender, +}; +use vcli_store::Store; + +use crate::bridge::{CommandChannel, SchedulerCommand}; + +/// Shared boundary between the tokio handler and the scheduler/store. +#[derive(Clone)] +pub struct DaemonHandler { + /// Async-side store handle. All DB ops run inside `spawn_blocking`. + pub store: Arc>, + /// Command + broadcast endpoints. + pub bridge: CommandChannel, + /// Wall-clock start time (for `health.uptime_ms`). + pub started_at: Instant, + /// Graceful-shutdown trigger, set when a client sends `Shutdown`. The + /// `run_foreground` task awaits the receiver. + pub shutdown_tx: Arc>>>, +} + +impl DaemonHandler { + /// Fire the shutdown oneshot if still armed. Idempotent. + /// + /// # Panics + /// Panics if the internal mutex is poisoned. + pub fn trigger_shutdown(&self) { + if let Some(tx) = self.shutdown_tx.lock().unwrap().take() { + let _ = tx.send(()); + } + } + + fn handle_health(&self, id: RequestId) -> Response { + let uptime_ms = u64::try_from(self.started_at.elapsed().as_millis()).unwrap_or(u64::MAX); + Response::ok( + id, + serde_json::json!({ + "version": env!("CARGO_PKG_VERSION"), + "uptime_ms": uptime_ms, + "socket_origin": "resolved", + }), + ) + } + + fn handle_shutdown(&self, id: RequestId) -> Response { + self.trigger_shutdown(); + let _ = self.bridge.cmd_tx.send(SchedulerCommand::Shutdown); + Response::ok(id, serde_json::json!({ "bye": true })) + } +} + +#[async_trait] +impl Handler for DaemonHandler { + async fn handle(&self, id: RequestId, op: RequestOp) -> IpcResult { + let resp = match op { + RequestOp::Health => self.handle_health(id), + RequestOp::Shutdown => self.handle_shutdown(id), + other => Response::err( + id, + ErrorPayload::simple(ErrorCode::Internal, format!("op not yet wired: {other:?}")), + ), + }; + Ok(resp) + } + + async fn handle_stream( + &self, + id: RequestId, + _op: RequestOp, + tx: StreamSender, + ) -> IpcResult<()> { + let _ = tx + .send(StreamFrame::end_of_stream(id, StreamKind::Events)) + .await; + Ok(()) + } +} + +#[cfg(test)] +pub(crate) mod test_support { + use super::*; + use tempfile::TempDir; + + /// Bundle returned by `fresh_handler` that keeps the tempdir alive and + /// exposes a few handles follow-up tasks consume directly. + pub struct Fixture { + // `dir` is held purely for Drop — the TempDir must outlive the Store. + // Why: `#[allow(dead_code)]` — required storage for the RAII tempdir + // even though no test reads the field. + #[allow(dead_code)] + pub dir: TempDir, + pub handler: DaemonHandler, + pub shutdown_rx: oneshot::Receiver<()>, + // Consumed by Task 8d+ tests; kept in the bundle so the channel isn't + // dropped before later tasks wire assertions against it. + // Why: `#[allow(dead_code)]` — alive now so the cmd_tx end inside + // `handler.bridge` doesn't disconnect, which would change test + // semantics when 8d starts asserting on dispatched commands. + #[allow(dead_code)] + pub cmd_rx: crossbeam_channel::Receiver, + } + + pub fn fresh_handler() -> Fixture { + let dir = tempfile::tempdir().unwrap(); + let (store, _) = Store::open(dir.path()).unwrap(); + let (bridge, cmd_rx, _event_rx, _sched_event_tx) = crate::bridge::new_channels(); + let (stx, srx) = oneshot::channel(); + let handler = DaemonHandler { + store: Arc::new(Mutex::new(store)), + bridge, + started_at: Instant::now(), + shutdown_tx: Arc::new(Mutex::new(Some(stx))), + }; + Fixture { + dir, + handler, + shutdown_rx: srx, + cmd_rx, + } + } +} + +#[cfg(test)] +mod tests { + use super::test_support::fresh_handler; + use super::*; + + #[tokio::test] + async fn health_returns_version_and_uptime() { + let f = fresh_handler(); + let id = RequestId::new(); + let resp = f.handler.handle(id, RequestOp::Health).await.unwrap(); + let body = serde_json::to_value(&resp).unwrap(); + assert_eq!(body["id"], id.to_string()); + assert_eq!(body["ok"], true); + assert!(body["result"]["version"].as_str().is_some()); + assert!(body["result"]["uptime_ms"].as_u64().is_some()); + } + + #[tokio::test] + async fn shutdown_triggers_oneshot_and_cmd() { + let f = fresh_handler(); + let id = RequestId::new(); + let _resp = f.handler.handle(id, RequestOp::Shutdown).await.unwrap(); + assert!(f.shutdown_rx.await.is_ok()); + } +} diff --git a/crates/vcli-daemon/src/lib.rs b/crates/vcli-daemon/src/lib.rs index a7cbe82..7468ddc 100644 --- a/crates/vcli-daemon/src/lib.rs +++ b/crates/vcli-daemon/src/lib.rs @@ -36,3 +36,7 @@ pub use bridge::{new_channels, CommandChannel, SchedulerCommand, EVENT_BROADCAST pub mod persist; pub use persist::{program_id_of, spawn_event_pump}; + +pub mod handler; + +pub use handler::DaemonHandler; From a3dfea991eec4893b2e451fcba9c2b57a3a7fccd Mon Sep 17 00:00:00 2001 From: Yixiang Zeng Date: Mon, 20 Apr 2026 15:42:52 -0700 Subject: [PATCH 12/25] vcli-daemon: DaemonHandler::Submit validates, persists, and enqueues handle_submit runs vcli_dsl::validate_value on the incoming JSON. On success it canonicalizes and writes a Pending row via spawn_blocking, then pushes SchedulerCommand::SubmitValidated into cmd_tx. DslError already exposes to_payload(), so error paths map straight to the wire. Co-Authored-By: Claude Opus 4.7 (1M context) --- crates/vcli-daemon/src/handler.rs | 121 ++++++++++++++++++++++++++++-- 1 file changed, 114 insertions(+), 7 deletions(-) diff --git a/crates/vcli-daemon/src/handler.rs b/crates/vcli-daemon/src/handler.rs index 1abe4f7..9b5d21e 100644 --- a/crates/vcli-daemon/src/handler.rs +++ b/crates/vcli-daemon/src/handler.rs @@ -8,7 +8,8 @@ use std::time::Instant; use async_trait::async_trait; use tokio::sync::oneshot; -use vcli_core::{ErrorCode, ErrorPayload}; +use tracing::error; +use vcli_core::{ErrorCode, ErrorPayload, ProgramId}; use vcli_ipc::{ Handler, IpcResult, RequestId, RequestOp, Response, StreamFrame, StreamKind, StreamSender, }; @@ -41,6 +42,80 @@ impl DaemonHandler { } } + async fn handle_submit( + &self, + id: RequestId, + program_json: serde_json::Value, + ) -> Response { + let program = match vcli_dsl::validate_value(&program_json) { + Ok(v) => v.program, + Err(e) => { + return Response::err(id, e.to_payload()); + } + }; + + let program_id = program.id.unwrap_or_else(ProgramId::new); + let name = program.name.clone(); + let canonical_bytes = match vcli_core::canonicalize(&program_json) { + Ok(b) => b, + Err(e) => { + return Response::err( + id, + ErrorPayload::simple(ErrorCode::Internal, format!("canonicalize: {e}")), + ); + } + }; + let canonical = String::from_utf8(canonical_bytes) + .expect("canonicalize produces UTF-8 by construction"); + + let store = self.store.clone(); + let pid = program_id; + let submitted_at = vcli_core::clock::now_unix_ms(); + let name_for_insert = name.clone(); + let canonical_str = canonical.clone(); + let insert_result = tokio::task::spawn_blocking(move || { + let mut s = store.lock().unwrap(); + s.insert_program(&vcli_store::NewProgram { + id: pid, + name: &name_for_insert, + source_json: &canonical_str, + state: vcli_core::ProgramState::Pending, + submitted_at, + labels_json: "{}", + }) + }) + .await; + let insert = match insert_result { + Ok(v) => v, + Err(e) => { + error!(error = %e, "submit join error"); + return Response::err( + id, + ErrorPayload::simple(ErrorCode::Internal, format!("{e}")), + ); + } + }; + if let Err(e) = insert { + return Response::err(id, ErrorPayload::simple(ErrorCode::Internal, format!("{e}"))); + } + + if let Err(e) = self.bridge.cmd_tx.send(SchedulerCommand::SubmitValidated { + program_id: pid, + program, + }) { + error!(error = %e, "cmd_tx full"); + return Response::err( + id, + ErrorPayload::simple(ErrorCode::DaemonBusy, "scheduler command queue full"), + ); + } + + Response::ok( + id, + serde_json::json!({ "program_id": pid.to_string(), "name": name }), + ) + } + fn handle_health(&self, id: RequestId) -> Response { let uptime_ms = u64::try_from(self.started_at.elapsed().as_millis()).unwrap_or(u64::MAX); Response::ok( @@ -66,6 +141,7 @@ impl Handler for DaemonHandler { let resp = match op { RequestOp::Health => self.handle_health(id), RequestOp::Shutdown => self.handle_shutdown(id), + RequestOp::Submit { program } => self.handle_submit(id, program).await, other => Response::err( id, ErrorPayload::simple(ErrorCode::Internal, format!("op not yet wired: {other:?}")), @@ -102,12 +178,6 @@ pub(crate) mod test_support { pub dir: TempDir, pub handler: DaemonHandler, pub shutdown_rx: oneshot::Receiver<()>, - // Consumed by Task 8d+ tests; kept in the bundle so the channel isn't - // dropped before later tasks wire assertions against it. - // Why: `#[allow(dead_code)]` — alive now so the cmd_tx end inside - // `handler.bridge` doesn't disconnect, which would change test - // semantics when 8d starts asserting on dispatched commands. - #[allow(dead_code)] pub cmd_rx: crossbeam_channel::Receiver, } @@ -148,6 +218,43 @@ mod tests { assert!(body["result"]["uptime_ms"].as_u64().is_some()); } + #[tokio::test] + async fn submit_validates_and_enqueues() { + let f = fresh_handler(); + let program = serde_json::json!({ + "version": "0.1", + "name": "noop", + "trigger": { "kind": "on_submit" }, + "predicates": {}, + "watches": [], + "body": [], + }); + let id = RequestId::new(); + let resp = f + .handler + .handle( + id, + RequestOp::Submit { + program: program.clone(), + }, + ) + .await + .unwrap(); + let body = serde_json::to_value(&resp).unwrap(); + assert_eq!(body["ok"], true); + let pid: ProgramId = body["result"]["program_id"] + .as_str() + .unwrap() + .parse() + .unwrap(); + let row = f.handler.store.lock().unwrap().get_program(pid).unwrap(); + assert_eq!(row.name, "noop"); + match f.cmd_rx.try_recv().unwrap() { + SchedulerCommand::SubmitValidated { program_id, .. } => assert_eq!(program_id, pid), + other => panic!("wrong cmd: {other:?}"), + } + } + #[tokio::test] async fn shutdown_triggers_oneshot_and_cmd() { let f = fresh_handler(); From df00780b7114d32ba31992b461de8a27e4adce25 Mon Sep 17 00:00:00 2001 From: Yixiang Zeng Date: Mon, 20 Apr 2026 15:43:41 -0700 Subject: [PATCH 13/25] vcli-daemon: DaemonHandler::List and Status handle_list parses the optional state filter, queries Store::list_programs off the async executor via spawn_blocking, and emits {items: [...]}. handle_status returns the single-row view or maps StoreError::UnknownProgram to ErrorCode::UnknownProgram. Co-Authored-By: Claude Opus 4.7 (1M context) --- crates/vcli-daemon/src/handler.rs | 140 ++++++++++++++++++++++++++++++ 1 file changed, 140 insertions(+) diff --git a/crates/vcli-daemon/src/handler.rs b/crates/vcli-daemon/src/handler.rs index 9b5d21e..39e1827 100644 --- a/crates/vcli-daemon/src/handler.rs +++ b/crates/vcli-daemon/src/handler.rs @@ -116,6 +116,70 @@ impl DaemonHandler { ) } + async fn handle_list(&self, id: RequestId, state: Option) -> Response { + let filter = state.and_then(|s| s.parse::().ok()); + let store = self.store.clone(); + let rows = tokio::task::spawn_blocking(move || { + let s = store.lock().unwrap(); + s.list_programs(filter) + }) + .await + .unwrap_or_else(|e| { + Err(vcli_store::StoreError::Io { + path: "".into(), + source: std::io::Error::other(format!("{e}")), + }) + }); + match rows { + Ok(rows) => { + let items: Vec<_> = rows + .into_iter() + .map(|r| { + serde_json::json!({ + "id": r.id.to_string(), + "name": r.name, + "state": r.state.as_str(), + "submitted_at": r.submitted_at, + }) + }) + .collect(); + Response::ok(id, serde_json::json!({ "items": items })) + } + Err(e) => Response::err(id, ErrorPayload::simple(ErrorCode::Internal, format!("{e}"))), + } + } + + async fn handle_status(&self, id: RequestId, program_id: ProgramId) -> Response { + let store = self.store.clone(); + let row = tokio::task::spawn_blocking(move || { + let s = store.lock().unwrap(); + s.get_program(program_id) + }) + .await; + match row { + Ok(Ok(r)) => Response::ok( + id, + serde_json::json!({ + "id": r.id.to_string(), + "name": r.name, + "state": r.state.as_str(), + "body_cursor": r.body_cursor, + "submitted_at": r.submitted_at, + "started_at": r.started_at, + "finished_at": r.finished_at, + }), + ), + Ok(Err(vcli_store::StoreError::UnknownProgram(_))) => Response::err( + id, + ErrorPayload::simple(ErrorCode::UnknownProgram, "not found"), + ), + Ok(Err(e)) => { + Response::err(id, ErrorPayload::simple(ErrorCode::Internal, format!("{e}"))) + } + Err(e) => Response::err(id, ErrorPayload::simple(ErrorCode::Internal, format!("{e}"))), + } + } + fn handle_health(&self, id: RequestId) -> Response { let uptime_ms = u64::try_from(self.started_at.elapsed().as_millis()).unwrap_or(u64::MAX); Response::ok( @@ -142,6 +206,8 @@ impl Handler for DaemonHandler { RequestOp::Health => self.handle_health(id), RequestOp::Shutdown => self.handle_shutdown(id), RequestOp::Submit { program } => self.handle_submit(id, program).await, + RequestOp::List { state } => self.handle_list(id, state).await, + RequestOp::Status { program_id } => self.handle_status(id, program_id).await, other => Response::err( id, ErrorPayload::simple(ErrorCode::Internal, format!("op not yet wired: {other:?}")), @@ -218,6 +284,80 @@ mod tests { assert!(body["result"]["uptime_ms"].as_u64().is_some()); } + #[tokio::test] + async fn list_returns_all_programs_when_no_filter() { + let f = fresh_handler(); + for name in ["a", "b"] { + let id = ProgramId::new(); + f.handler + .store + .lock() + .unwrap() + .insert_program(&vcli_store::NewProgram { + id, + name, + source_json: "{}", + state: vcli_core::ProgramState::Pending, + submitted_at: 0, + labels_json: "{}", + }) + .unwrap(); + } + let resp = f + .handler + .handle(RequestId::new(), RequestOp::List { state: None }) + .await + .unwrap(); + let body = serde_json::to_value(&resp).unwrap(); + let items = body["result"]["items"].as_array().unwrap(); + assert_eq!(items.len(), 2); + } + + #[tokio::test] + async fn status_returns_row_for_known_id() { + let f = fresh_handler(); + let pid = ProgramId::new(); + f.handler + .store + .lock() + .unwrap() + .insert_program(&vcli_store::NewProgram { + id: pid, + name: "s", + source_json: "{}", + state: vcli_core::ProgramState::Pending, + submitted_at: 0, + labels_json: "{}", + }) + .unwrap(); + let resp = f + .handler + .handle(RequestId::new(), RequestOp::Status { program_id: pid }) + .await + .unwrap(); + let body = serde_json::to_value(&resp).unwrap(); + assert_eq!(body["result"]["name"], "s"); + assert_eq!(body["result"]["state"], "pending"); + } + + #[tokio::test] + async fn status_returns_unknown_program_for_missing_id() { + let f = fresh_handler(); + let resp = f + .handler + .handle( + RequestId::new(), + RequestOp::Status { + program_id: ProgramId::new(), + }, + ) + .await + .unwrap(); + let body = serde_json::to_value(&resp).unwrap(); + assert_eq!(body["ok"], false); + assert_eq!(body["error"]["code"], "unknown_program"); + } + #[tokio::test] async fn submit_validates_and_enqueues() { let f = fresh_handler(); From 4d33caac9882eaf5d4c278010824e66443669176 Mon Sep 17 00:00:00 2001 From: Yixiang Zeng Date: Mon, 20 Apr 2026 15:44:26 -0700 Subject: [PATCH 14/25] vcli-daemon: DaemonHandler::Cancel and Start Both ops push the matching SchedulerCommand onto cmd_tx and surface a DaemonBusy error if the bounded queue is full. Scheduler owns the state transition; the handler is a pure forwarder. Co-Authored-By: Claude Opus 4.7 (1M context) --- crates/vcli-daemon/src/handler.rs | 63 +++++++++++++++++++++++++++++++ 1 file changed, 63 insertions(+) diff --git a/crates/vcli-daemon/src/handler.rs b/crates/vcli-daemon/src/handler.rs index 39e1827..e1160d7 100644 --- a/crates/vcli-daemon/src/handler.rs +++ b/crates/vcli-daemon/src/handler.rs @@ -116,6 +116,33 @@ impl DaemonHandler { ) } + fn handle_cancel(&self, id: RequestId, pid: ProgramId) -> Response { + match self.bridge.cmd_tx.send(SchedulerCommand::Cancel { + program_id: pid, + reason: "user".into(), + }) { + Ok(()) => Response::ok(id, serde_json::json!({ "program_id": pid.to_string() })), + Err(_) => Response::err( + id, + ErrorPayload::simple(ErrorCode::DaemonBusy, "cmd queue full"), + ), + } + } + + fn handle_start(&self, id: RequestId, pid: ProgramId) -> Response { + match self + .bridge + .cmd_tx + .send(SchedulerCommand::Start { program_id: pid }) + { + Ok(()) => Response::ok(id, serde_json::json!({ "program_id": pid.to_string() })), + Err(_) => Response::err( + id, + ErrorPayload::simple(ErrorCode::DaemonBusy, "cmd queue full"), + ), + } + } + async fn handle_list(&self, id: RequestId, state: Option) -> Response { let filter = state.and_then(|s| s.parse::().ok()); let store = self.store.clone(); @@ -208,6 +235,8 @@ impl Handler for DaemonHandler { RequestOp::Submit { program } => self.handle_submit(id, program).await, RequestOp::List { state } => self.handle_list(id, state).await, RequestOp::Status { program_id } => self.handle_status(id, program_id).await, + RequestOp::Cancel { program_id } => self.handle_cancel(id, program_id), + RequestOp::Start { program_id } => self.handle_start(id, program_id), other => Response::err( id, ErrorPayload::simple(ErrorCode::Internal, format!("op not yet wired: {other:?}")), @@ -284,6 +313,40 @@ mod tests { assert!(body["result"]["uptime_ms"].as_u64().is_some()); } + #[tokio::test] + async fn cancel_sends_command_and_returns_ok() { + let f = fresh_handler(); + let pid = ProgramId::new(); + let resp = f + .handler + .handle(RequestId::new(), RequestOp::Cancel { program_id: pid }) + .await + .unwrap(); + let body = serde_json::to_value(&resp).unwrap(); + assert_eq!(body["ok"], true); + match f.cmd_rx.try_recv().unwrap() { + SchedulerCommand::Cancel { program_id, .. } => assert_eq!(program_id, pid), + other => panic!("wrong cmd: {other:?}"), + } + } + + #[tokio::test] + async fn start_sends_command_and_returns_ok() { + let f = fresh_handler(); + let pid = ProgramId::new(); + let resp = f + .handler + .handle(RequestId::new(), RequestOp::Start { program_id: pid }) + .await + .unwrap(); + let body = serde_json::to_value(&resp).unwrap(); + assert_eq!(body["ok"], true); + match f.cmd_rx.try_recv().unwrap() { + SchedulerCommand::Start { program_id } => assert_eq!(program_id, pid), + other => panic!("wrong cmd: {other:?}"), + } + } + #[tokio::test] async fn list_returns_all_programs_when_no_filter() { let f = fresh_handler(); From 02ecbd7d294fa7d970ba6e093492949e935a8b8a Mon Sep 17 00:00:00 2001 From: Yixiang Zeng Date: Mon, 20 Apr 2026 15:45:23 -0700 Subject: [PATCH 15/25] vcli-daemon: DaemonHandler::Resume handle_resume runs Store::resume_program (DB state transition) then re-reads and re-validates the source_json via vcli_dsl so we hand the scheduler a fresh vcli_core::Program. StoreError::NotResumable and UnknownProgram are mapped to matching ErrorCode variants so clients can distinguish "wrong state" from "missing id" cleanly. Co-Authored-By: Claude Opus 4.7 (1M context) --- crates/vcli-daemon/src/handler.rs | 152 ++++++++++++++++++++++++++++++ 1 file changed, 152 insertions(+) diff --git a/crates/vcli-daemon/src/handler.rs b/crates/vcli-daemon/src/handler.rs index e1160d7..217e590 100644 --- a/crates/vcli-daemon/src/handler.rs +++ b/crates/vcli-daemon/src/handler.rs @@ -116,6 +116,58 @@ impl DaemonHandler { ) } + async fn handle_resume( + &self, + id: RequestId, + pid: ProgramId, + from_start: bool, + ) -> Response { + let store = self.store.clone(); + let now_ms = vcli_core::clock::now_unix_ms(); + let resume_result = tokio::task::spawn_blocking(move || { + let mut s = store.lock().unwrap(); + let outcome = s.resume_program(pid, from_start, now_ms)?; + let row = s.get_program(pid)?; + let value: serde_json::Value = serde_json::from_str(&row.source_json)?; + let program = vcli_dsl::validate_value(&value) + .map(|v| v.program) + .map_err(|e| vcli_store::StoreError::Io { + path: "".into(), + source: std::io::Error::new( + std::io::ErrorKind::InvalidData, + format!("{e}"), + ), + })?; + Ok::<_, vcli_store::StoreError>((outcome, program)) + }) + .await; + + match resume_result { + Ok(Ok((out, program))) => { + let _ = self.bridge.cmd_tx.send(SchedulerCommand::ResumeRunning { + program_id: pid, + from_step: out.from_step, + program, + }); + Response::ok( + id, + serde_json::json!({ "program_id": pid.to_string(), "from_step": out.from_step }), + ) + } + Ok(Err(vcli_store::StoreError::NotResumable(m))) => { + Response::err(id, ErrorPayload::simple(ErrorCode::NotResumable, m)) + } + Ok(Err(vcli_store::StoreError::UnknownProgram(_))) => Response::err( + id, + ErrorPayload::simple(ErrorCode::UnknownProgram, "not found"), + ), + Ok(Err(e)) => { + Response::err(id, ErrorPayload::simple(ErrorCode::Internal, format!("{e}"))) + } + Err(e) => Response::err(id, ErrorPayload::simple(ErrorCode::Internal, format!("{e}"))), + } + } + fn handle_cancel(&self, id: RequestId, pid: ProgramId) -> Response { match self.bridge.cmd_tx.send(SchedulerCommand::Cancel { program_id: pid, @@ -237,6 +289,10 @@ impl Handler for DaemonHandler { RequestOp::Status { program_id } => self.handle_status(id, program_id).await, RequestOp::Cancel { program_id } => self.handle_cancel(id, program_id), RequestOp::Start { program_id } => self.handle_start(id, program_id), + RequestOp::Resume { + program_id, + from_start, + } => self.handle_resume(id, program_id, from_start).await, other => Response::err( id, ErrorPayload::simple(ErrorCode::Internal, format!("op not yet wired: {other:?}")), @@ -313,6 +369,102 @@ mod tests { assert!(body["result"]["uptime_ms"].as_u64().is_some()); } + fn minimal_valid_program_json() -> String { + serde_json::json!({ + "version": "0.1", + "name": "r", + "trigger": { "kind": "on_submit" }, + "predicates": {}, + "watches": [], + "body": [], + }) + .to_string() + } + + #[tokio::test] + async fn resume_transitions_store_and_sends_command() { + let f = fresh_handler(); + let pid = ProgramId::new(); + let path = f.dir.path().to_path_buf(); + let src = minimal_valid_program_json(); + { + let mut s = f.handler.store.lock().unwrap(); + s.insert_program(&vcli_store::NewProgram { + id: pid, + name: "r", + source_json: &src, + state: vcli_core::ProgramState::Pending, + submitted_at: 0, + labels_json: "{}", + }) + .unwrap(); + s.update_state(pid, vcli_core::ProgramState::Running, 1) + .unwrap(); + s.set_body_cursor(pid, 3).unwrap(); + } + // Trigger a recovery cycle by reopening the same DB. + let (_, _) = vcli_store::Store::open(&path).unwrap(); + + let resp = f + .handler + .handle( + RequestId::new(), + RequestOp::Resume { + program_id: pid, + from_start: false, + }, + ) + .await + .unwrap(); + let body = serde_json::to_value(&resp).unwrap(); + assert_eq!(body["ok"], true); + assert_eq!(body["result"]["from_step"], 3); + match f.cmd_rx.try_recv().unwrap() { + SchedulerCommand::ResumeRunning { + program_id, + from_step, + .. + } => { + assert_eq!(program_id, pid); + assert_eq!(from_step, 3); + } + other => panic!("wrong cmd: {other:?}"), + } + } + + #[tokio::test] + async fn resume_rejects_non_resumable_program() { + let f = fresh_handler(); + let pid = ProgramId::new(); + f.handler + .store + .lock() + .unwrap() + .insert_program(&vcli_store::NewProgram { + id: pid, + name: "r", + source_json: "{}", + state: vcli_core::ProgramState::Pending, + submitted_at: 0, + labels_json: "{}", + }) + .unwrap(); + let resp = f + .handler + .handle( + RequestId::new(), + RequestOp::Resume { + program_id: pid, + from_start: false, + }, + ) + .await + .unwrap(); + let body = serde_json::to_value(&resp).unwrap(); + assert_eq!(body["ok"], false); + assert_eq!(body["error"]["code"], "not_resumable"); + } + #[tokio::test] async fn cancel_sends_command_and_returns_ok() { let f = fresh_handler(); From 099b4c9e33fd4d9e2526a503562420fc231527dc Mon Sep 17 00:00:00 2001 From: Yixiang Zeng Date: Mon, 20 Apr 2026 15:47:01 -0700 Subject: [PATCH 16/25] vcli-daemon: DaemonHandler streaming for Events, Logs, and Trace stream_events handles both firehose and per-program filtering plus follow-mode broadcast receives with lag-recovery via a stream.dropped frame. Non-follow mode pages the stored history out of SQLite and returns. Trace is a stubbed no-op pending runtime trace support. Co-Authored-By: Claude Opus 4.7 (1M context) --- crates/vcli-daemon/src/handler.rs | 145 ++++++++++++++++++++++++++++-- 1 file changed, 137 insertions(+), 8 deletions(-) diff --git a/crates/vcli-daemon/src/handler.rs b/crates/vcli-daemon/src/handler.rs index 217e590..b659097 100644 --- a/crates/vcli-daemon/src/handler.rs +++ b/crates/vcli-daemon/src/handler.rs @@ -7,9 +7,9 @@ use std::sync::{Arc, Mutex}; use std::time::Instant; use async_trait::async_trait; -use tokio::sync::oneshot; -use tracing::error; -use vcli_core::{ErrorCode, ErrorPayload, ProgramId}; +use tokio::sync::{broadcast, oneshot}; +use tracing::{debug, error}; +use vcli_core::{ErrorCode, ErrorPayload, Event, ProgramId}; use vcli_ipc::{ Handler, IpcResult, RequestId, RequestOp, Response, StreamFrame, StreamKind, StreamSender, }; @@ -116,6 +116,70 @@ impl DaemonHandler { ) } + async fn stream_events( + &self, + id: RequestId, + filter_program: Option, + follow: bool, + tx: StreamSender, + ) -> IpcResult<()> { + let mut rx = self.bridge.event_tx.subscribe(); + if !follow { + let store = self.store.clone(); + let history = tokio::task::spawn_blocking(move || { + let s = store.lock().unwrap(); + s.stream_events(0, 10_000) + }) + .await + .ok() + .and_then(Result::ok) + .unwrap_or_default(); + for row in history { + if filter_program.is_some_and(|p| p != row.program_id) { + continue; + } + if let Ok(ev) = serde_json::from_str::(&row.data_json) { + let frame = StreamFrame::event(id, ev); + if tx.send(frame).await.is_err() { + return Ok(()); + } + } + } + return Ok(()); + } + loop { + match rx.recv().await { + Ok(ev) => { + if let Some(p) = filter_program { + if crate::persist::program_id_of(&ev.data) != Some(p) { + continue; + } + } + if tx.send(StreamFrame::event(id, ev)).await.is_err() { + return Ok(()); + } + } + Err(broadcast::error::RecvError::Lagged(n)) => { + let now = vcli_core::clock::now_unix_ms(); + let dropped = StreamFrame::event( + id, + Event { + at: now, + data: vcli_core::EventData::StreamDropped { + count: u32::try_from(n).unwrap_or(u32::MAX), + since: now, + }, + }, + ); + if tx.send(dropped).await.is_err() { + return Ok(()); + } + } + Err(broadcast::error::RecvError::Closed) => return Ok(()), + } + } + } + async fn handle_resume( &self, id: RequestId, @@ -304,13 +368,28 @@ impl Handler for DaemonHandler { async fn handle_stream( &self, id: RequestId, - _op: RequestOp, + op: RequestOp, tx: StreamSender, ) -> IpcResult<()> { - let _ = tx - .send(StreamFrame::end_of_stream(id, StreamKind::Events)) - .await; - Ok(()) + match op { + RequestOp::Events { follow } => self.stream_events(id, None, follow, tx).await, + RequestOp::Logs { + program_id, + follow, + } => self.stream_events(id, Some(program_id), follow, tx).await, + RequestOp::Trace { program_id: _ } => { + // v0 minimum: empty trace; server writes end_of_stream on return. + let _ = tx; + Ok(()) + } + other => { + debug!("stream op {other:?} not supported"); + let _ = tx + .send(StreamFrame::end_of_stream(id, StreamKind::Events)) + .await; + Ok(()) + } + } } } @@ -369,6 +448,56 @@ mod tests { assert!(body["result"]["uptime_ms"].as_u64().is_some()); } + #[tokio::test] + async fn events_stream_drains_history_then_closes_when_not_following() { + use tokio::sync::mpsc; + let f = fresh_handler(); + let pid = ProgramId::new(); + { + let mut s = f.handler.store.lock().unwrap(); + s.insert_program(&vcli_store::NewProgram { + id: pid, + name: "x", + source_json: "{}", + state: vcli_core::ProgramState::Pending, + submitted_at: 0, + labels_json: "{}", + }) + .unwrap(); + let ev = Event { + at: 7, + data: vcli_core::EventData::ProgramCompleted { + program_id: pid, + emit: None, + }, + }; + s.append_event(pid, &ev).unwrap(); + } + + let (tx, mut rx) = mpsc::channel::(8); + let sender = StreamSender(tx); + let handler = f.handler.clone(); + let task = tokio::spawn(async move { + handler + .handle_stream( + RequestId::new(), + RequestOp::Events { follow: false }, + sender, + ) + .await + }); + + let frame = tokio::time::timeout(std::time::Duration::from_millis(500), rx.recv()) + .await + .expect("stream_events should deliver history in <500ms") + .expect("channel closed before delivering frame"); + assert!(frame.event.is_some()); + + let _ = tokio::time::timeout(std::time::Duration::from_millis(500), task) + .await + .expect("handler should return after draining history"); + } + fn minimal_valid_program_json() -> String { serde_json::json!({ "version": "0.1", From ade7e197a8bdfcb6fe70ef8dd3e800abbf8cdd9f Mon Sep 17 00:00:00 2001 From: Yixiang Zeng Date: Mon, 20 Apr 2026 15:47:34 -0700 Subject: [PATCH 17/25] vcli-daemon: DaemonHandler::Gc wraps Store::gc_all handle_gc computes a retention cutoff (now - RETENTION_DAYS) on the async executor, calls Store::gc_all via spawn_blocking, and surfaces the GcReport fields (programs/assets/blobs/orphan_blobs deleted). Co-Authored-By: Claude Opus 4.7 (1M context) --- crates/vcli-daemon/src/handler.rs | 41 +++++++++++++++++++++++++++++++ 1 file changed, 41 insertions(+) diff --git a/crates/vcli-daemon/src/handler.rs b/crates/vcli-daemon/src/handler.rs index b659097..e993511 100644 --- a/crates/vcli-daemon/src/handler.rs +++ b/crates/vcli-daemon/src/handler.rs @@ -180,6 +180,32 @@ impl DaemonHandler { } } + async fn handle_gc(&self, id: RequestId) -> Response { + let store = self.store.clone(); + let now = vcli_core::clock::now_unix_ms(); + let cutoff = now - i64::from(vcli_store::RETENTION_DAYS) * 24 * 60 * 60 * 1000; + let report = tokio::task::spawn_blocking(move || { + let mut s = store.lock().unwrap(); + s.gc_all(cutoff) + }) + .await; + match report { + Ok(Ok(r)) => Response::ok( + id, + serde_json::json!({ + "programs_deleted": r.programs_deleted, + "assets_deleted": r.assets_deleted, + "blobs_deleted": r.blobs_deleted, + "orphan_blobs_deleted": r.orphan_blobs_deleted, + }), + ), + Ok(Err(e)) => { + Response::err(id, ErrorPayload::simple(ErrorCode::Internal, format!("{e}"))) + } + Err(e) => Response::err(id, ErrorPayload::simple(ErrorCode::Internal, format!("{e}"))), + } + } + async fn handle_resume( &self, id: RequestId, @@ -357,6 +383,7 @@ impl Handler for DaemonHandler { program_id, from_start, } => self.handle_resume(id, program_id, from_start).await, + RequestOp::Gc => self.handle_gc(id).await, other => Response::err( id, ErrorPayload::simple(ErrorCode::Internal, format!("op not yet wired: {other:?}")), @@ -448,6 +475,20 @@ mod tests { assert!(body["result"]["uptime_ms"].as_u64().is_some()); } + #[tokio::test] + async fn gc_returns_report_shape() { + let f = fresh_handler(); + let resp = f + .handler + .handle(RequestId::new(), RequestOp::Gc) + .await + .unwrap(); + let body = serde_json::to_value(&resp).unwrap(); + assert_eq!(body["ok"], true); + assert!(body["result"]["programs_deleted"].as_u64().is_some()); + assert!(body["result"]["assets_deleted"].as_u64().is_some()); + } + #[tokio::test] async fn events_stream_drains_history_then_closes_when_not_following() { use tokio::sync::mpsc; From 4f71ff2d5592bb1ea0046f384c68aad6e510c120 Mon Sep 17 00:00:00 2001 From: Yixiang Zeng Date: Mon, 20 Apr 2026 15:48:12 -0700 Subject: [PATCH 18/25] =?UTF-8?q?vcli-daemon:=20startup=20=E2=80=94=20reco?= =?UTF-8?q?very=20events=20+=20waiting=20reload=20+=20daemon.started?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit emit_recovery_events writes (program.state_changed, program.failed) pairs for every RecoveredProgram returned by Store::open — so IPC subscribers see a clean transition from running into failed(daemon_restart). emit_daemon_started posts the version banner. reload_waiting_programs walks rows in Waiting state and pushes each through SchedulerCommand::SubmitValidated so triggers are re-armed. Co-Authored-By: Claude Opus 4.7 (1M context) --- crates/vcli-daemon/src/lib.rs | 4 + crates/vcli-daemon/src/startup.rs | 131 ++++++++++++++++++++++++++++++ 2 files changed, 135 insertions(+) create mode 100644 crates/vcli-daemon/src/startup.rs diff --git a/crates/vcli-daemon/src/lib.rs b/crates/vcli-daemon/src/lib.rs index 7468ddc..be0b927 100644 --- a/crates/vcli-daemon/src/lib.rs +++ b/crates/vcli-daemon/src/lib.rs @@ -40,3 +40,7 @@ pub use persist::{program_id_of, spawn_event_pump}; pub mod handler; pub use handler::DaemonHandler; + +pub mod startup; + +pub use startup::{emit_daemon_started, emit_recovery_events, reload_waiting_programs}; diff --git a/crates/vcli-daemon/src/startup.rs b/crates/vcli-daemon/src/startup.rs new file mode 100644 index 0000000..97087ee --- /dev/null +++ b/crates/vcli-daemon/src/startup.rs @@ -0,0 +1,131 @@ +//! Startup tasks: emit events for orphan recovery, reload waiting programs, +//! publish `daemon.started`. Runs once on every boot; idempotent given the +//! `RecoveredProgram` set from `Store::open` is fresh each call. + +use std::sync::{Arc, Mutex}; + +use crossbeam_channel::Sender; +use tracing::info; + +use vcli_core::{Event, EventData, ProgramState}; +use vcli_store::{RecoveredProgram, Store}; + +/// Emit one `program.state_changed` + one `program.failed` for every program +/// that `Store::open` rewrote from `running` → `failed(daemon_restart)`. +/// The events are sent on `sched_event_tx` (scheduler → pump side), so the +/// persistence pipeline sees them first. +pub fn emit_recovery_events(recovered: &[RecoveredProgram], sched_event_tx: &Sender) { + let now = vcli_core::clock::now_unix_ms(); + for r in recovered { + let ev1 = Event { + at: now, + data: EventData::ProgramStateChanged { + program_id: r.id, + from: ProgramState::Running, + to: ProgramState::Failed, + reason: "daemon_restart".into(), + }, + }; + let ev2 = Event { + at: now, + data: EventData::ProgramFailed { + program_id: r.id, + reason: "daemon restarted during execution".into(), + step: Some(format!("body[{}]", r.body_cursor)), + emit: None, + }, + }; + let _ = sched_event_tx.send(ev1); + let _ = sched_event_tx.send(ev2); + } + info!(count = recovered.len(), "emitted recovery events"); +} + +/// Emit a `daemon.started` event. +pub fn emit_daemon_started(sched_event_tx: &Sender) { + let ev = Event { + at: vcli_core::clock::now_unix_ms(), + data: EventData::DaemonStarted { + version: env!("CARGO_PKG_VERSION").into(), + }, + }; + let _ = sched_event_tx.send(ev); +} + +/// Reload programs currently in `waiting` state: enqueue a scheduler command +/// per program so the scheduler re-installs its trigger. This is the second +/// half of spec §Restart semantics step 5. +/// +/// # Panics +/// Panics if the store mutex is poisoned. +pub fn reload_waiting_programs( + store: &Arc>, + cmd_tx: &crossbeam_channel::Sender, +) -> usize { + let rows = { + let s = store.lock().unwrap(); + s.list_programs(Some(ProgramState::Waiting)) + .unwrap_or_default() + }; + let mut sent = 0; + for row in &rows { + let Ok(value) = serde_json::from_str::(&row.source_json) else { + continue; + }; + let Ok(program) = vcli_dsl::validate_value(&value).map(|v| v.program) else { + continue; + }; + if cmd_tx + .send(crate::bridge::SchedulerCommand::SubmitValidated { + program_id: row.id, + program, + }) + .is_ok() + { + sent += 1; + } + } + info!(count = sent, "reloaded waiting programs"); + sent +} + +#[cfg(test)] +mod tests { + use super::*; + use crossbeam_channel::unbounded; + use tempfile::tempdir; + + #[test] + fn emits_two_events_per_recovered_program() { + let (tx, rx) = unbounded::(); + let recovered = vec![RecoveredProgram { + id: vcli_core::ProgramId::new(), + body_cursor: 2, + }]; + emit_recovery_events(&recovered, &tx); + let first = rx.recv().unwrap(); + let second = rx.recv().unwrap(); + assert!(matches!(first.data, EventData::ProgramStateChanged { .. })); + assert!(matches!(second.data, EventData::ProgramFailed { .. })); + } + + #[test] + fn emit_daemon_started_sends_one_event() { + let (tx, rx) = unbounded::(); + emit_daemon_started(&tx); + assert!(matches!( + rx.recv().unwrap().data, + EventData::DaemonStarted { .. } + )); + } + + #[test] + fn reload_waiting_does_nothing_when_none_exist() { + let d = tempdir().unwrap(); + let (store, _) = Store::open(d.path()).unwrap(); + let store = Arc::new(Mutex::new(store)); + let (cmd_tx, _cmd_rx) = unbounded::(); + let n = reload_waiting_programs(&store, &cmd_tx); + assert_eq!(n, 0); + } +} From 3141cd618e7a03cc4fbe86dda229bf8e9a66288e Mon Sep 17 00:00:00 2001 From: Yixiang Zeng Date: Mon, 20 Apr 2026 15:49:04 -0700 Subject: [PATCH 19/25] vcli-daemon: signal handlers + daemon.stopped emission MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit install_signal_handlers spawns a signal-hook-tokio task that watches SIGTERM + SIGINT and fires the shared shutdown oneshot. shutdown_tx is an Arc>> shared with DaemonHandler so either the OS or an IPC Shutdown request wins — whichever races first. emit_daemon_stopped posts the terminal event on the scheduler→pump channel so the DB and every subscriber see the shutdown. Co-Authored-By: Claude Opus 4.7 (1M context) --- crates/vcli-daemon/Cargo.toml | 1 + crates/vcli-daemon/src/lib.rs | 4 ++ crates/vcli-daemon/src/shutdown.rs | 70 ++++++++++++++++++++++++++++++ 3 files changed, 75 insertions(+) create mode 100644 crates/vcli-daemon/src/shutdown.rs diff --git a/crates/vcli-daemon/Cargo.toml b/crates/vcli-daemon/Cargo.toml index e80fb0f..a9eaeb5 100644 --- a/crates/vcli-daemon/Cargo.toml +++ b/crates/vcli-daemon/Cargo.toml @@ -39,6 +39,7 @@ dirs = { workspace = true } futures-util = { workspace = true } signal-hook-tokio = { workspace = true } uuid = { workspace = true } +libc = { workspace = true } [dev-dependencies] tempfile = { workspace = true } diff --git a/crates/vcli-daemon/src/lib.rs b/crates/vcli-daemon/src/lib.rs index be0b927..f4455d2 100644 --- a/crates/vcli-daemon/src/lib.rs +++ b/crates/vcli-daemon/src/lib.rs @@ -44,3 +44,7 @@ pub use handler::DaemonHandler; pub mod startup; pub use startup::{emit_daemon_started, emit_recovery_events, reload_waiting_programs}; + +pub mod shutdown; + +pub use shutdown::{emit_daemon_stopped, install_signal_handlers}; diff --git a/crates/vcli-daemon/src/shutdown.rs b/crates/vcli-daemon/src/shutdown.rs new file mode 100644 index 0000000..63a574c --- /dev/null +++ b/crates/vcli-daemon/src/shutdown.rs @@ -0,0 +1,70 @@ +//! Signal handling. On Unix, SIGTERM and SIGINT both trigger graceful shutdown +//! by firing the oneshot the IPC server is listening on. The second signal is +//! not special — `tokio::select!` races the loop already. A third signal +//! should escalate via the OS (we don't install SIGKILL handling). + +use std::sync::{Arc, Mutex}; + +use futures_util::stream::StreamExt; +use tokio::sync::oneshot; +use tracing::{info, warn}; + +/// Install a signal handler on the current tokio runtime that fires +/// `shutdown_tx` on the first SIGTERM or SIGINT. `shutdown_tx` is shared with +/// `DaemonHandler` so an IPC `shutdown` also counts. +/// +/// # Errors +/// Returns `std::io::Error` if `signal_hook_tokio::Signals::new` fails. +/// +/// # Panics +/// The background signal task panics if the shutdown mutex is poisoned. +pub async fn install_signal_handlers( + shutdown_tx: Arc>>>, +) -> std::io::Result<()> { + use signal_hook_tokio::Signals; + let mut signals = Signals::new([libc::SIGTERM, libc::SIGINT])?; + tokio::spawn(async move { + if let Some(sig) = signals.next().await { + info!(signal = ?sig, "received shutdown signal"); + if let Some(tx) = shutdown_tx.lock().unwrap().take() { + let _ = tx.send(()); + } else { + warn!("shutdown already triggered"); + } + } + }); + Ok(()) +} + +/// Emit `daemon.stopped` on the scheduler event channel so IPC subscribers and +/// the DB see the transition. +pub fn emit_daemon_stopped(sched_event_tx: &crossbeam_channel::Sender) { + use vcli_core::{Event, EventData}; + let _ = sched_event_tx.send(Event { + at: vcli_core::clock::now_unix_ms(), + data: EventData::DaemonStopped, + }); +} + +#[cfg(test)] +mod tests { + use super::*; + use crossbeam_channel::unbounded; + + #[test] + fn emit_daemon_stopped_sends_one_event() { + let (tx, rx) = unbounded::(); + emit_daemon_stopped(&tx); + assert!(matches!( + rx.recv().unwrap().data, + vcli_core::EventData::DaemonStopped + )); + } + + #[tokio::test] + async fn install_handler_is_idempotent_without_signal() { + let (tx, _rx) = oneshot::channel(); + let slot = Arc::new(Mutex::new(Some(tx))); + install_signal_handlers(slot.clone()).await.unwrap(); + } +} From 62cc6c05f7d3b350f42a809280b53b7b2bd7534f Mon Sep 17 00:00:00 2001 From: Yixiang Zeng Date: Mon, 20 Apr 2026 15:50:10 -0700 Subject: [PATCH 20/25] vcli-daemon: run_foreground wires store, bridge, scheduler, IPC MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Orchestrates the full startup sequence in the order spec §Restart semantics Phase A requires: ensure_dirs → logging → pidfile → open store (restart recovery) → channels → event pump → recovery events → daemon.started → reload waiting programs → spawn scheduler thread → build handler + shutdown oneshot → signal handlers → bind socket LAST (Decision 1.2, observable readiness). Shutdown path drains the scheduler, emits daemon.stopped, closes the pump, and releases the pidfile. The RuntimeFactory closure seam lets tests inject MockCapture and MockInputSink; the binary wires the real macOS backends in Task 12. Co-Authored-By: Claude Opus 4.7 (1M context) --- crates/vcli-daemon/src/lib.rs | 4 ++ crates/vcli-daemon/src/run.rs | 129 ++++++++++++++++++++++++++++++++++ 2 files changed, 133 insertions(+) create mode 100644 crates/vcli-daemon/src/run.rs diff --git a/crates/vcli-daemon/src/lib.rs b/crates/vcli-daemon/src/lib.rs index f4455d2..e4dcf35 100644 --- a/crates/vcli-daemon/src/lib.rs +++ b/crates/vcli-daemon/src/lib.rs @@ -48,3 +48,7 @@ pub use startup::{emit_daemon_started, emit_recovery_events, reload_waiting_prog pub mod shutdown; pub use shutdown::{emit_daemon_stopped, install_signal_handlers}; + +pub mod run; + +pub use run::{run_foreground, RuntimeBackends, RuntimeFactory}; diff --git a/crates/vcli-daemon/src/run.rs b/crates/vcli-daemon/src/run.rs new file mode 100644 index 0000000..81547c9 --- /dev/null +++ b/crates/vcli-daemon/src/run.rs @@ -0,0 +1,129 @@ +//! Daemon assembly. `run_foreground` is the function the binary calls. All +//! meaningful side effects (opening the store, binding the socket, spawning +//! the scheduler thread, installing signals) happen inside here, in exactly +//! the order spec §Restart semantics Phase A requires. + +use std::sync::{Arc, Mutex}; +use std::thread; +use std::time::Instant; + +use tokio::sync::oneshot; +use tracing::{error, info}; + +use vcli_capture::Capture; +use vcli_input::InputSink; +use vcli_ipc::IpcServer; +use vcli_perception::Perception; +use vcli_store::Store; + +use crate::bridge::{new_channels, SchedulerCommand}; +use crate::config::{ensure_dirs, Config}; +use crate::error::{DaemonError, DaemonResult}; +use crate::handler::DaemonHandler; +use crate::persist::spawn_event_pump; +use crate::pidfile::PidFile; +use crate::shutdown::{emit_daemon_stopped, install_signal_handlers}; +use crate::startup::{emit_daemon_started, emit_recovery_events, reload_waiting_programs}; + +/// Bundle of backend implementations the daemon will hand to the scheduler. +pub struct RuntimeBackends { + /// Capture backend. + pub capture: Box, + /// Input sink. + pub input: Arc, + /// Perception façade. + pub perception: Perception, + /// Runtime clock (usually `vcli_runtime::SystemRuntimeClock`). The runtime + /// crate exposes its own `RuntimeClock` trait because Rust 1.75 lacks + /// trait-object upcasting, so we cannot pass a `vcli_core::Clock` here. + pub clock: Arc, +} + +/// Function that produces `RuntimeBackends` at startup. Lets tests inject +/// mocks without entering `run_foreground` through platform backends. +pub type RuntimeFactory = Box DaemonResult + Send>; + +/// Foreground entrypoint. Blocks until a signal / IPC shutdown drains the +/// process. Use this from `fn main` after argv parsing. +/// +/// # Errors +/// Startup failures (path resolution, already-running, store open, socket bind). +pub async fn run_foreground(cfg: Config, factory: RuntimeFactory) -> DaemonResult<()> { + ensure_dirs(&cfg)?; + let _log_guard = crate::logging::init(&cfg.log_dir)?; + info!( + data_root = %cfg.data_root.display(), + socket = %cfg.socket.path.display(), + "starting vcli-daemon" + ); + + let pid = PidFile::acquire(cfg.pidfile_path())?; + info!(pid = pid.pid(), pidfile = %pid.path().display(), "pidfile acquired"); + + let (store, recovered) = Store::open(&cfg.data_root)?; + let store = Arc::new(Mutex::new(store)); + + let (bridge, cmd_rx, event_rx, sched_event_tx) = new_channels(); + + let pump = spawn_event_pump(store.clone(), event_rx, bridge.event_tx.clone()); + + emit_recovery_events(&recovered, &sched_event_tx); + emit_daemon_started(&sched_event_tx); + + let _ = reload_waiting_programs(&store, &bridge.cmd_tx); + + let RuntimeBackends { + capture, + input, + perception, + clock, + } = factory()?; + let sched_event_tx_for_thread = sched_event_tx.clone(); + let scheduler_join = thread::Builder::new() + .name("vcli-scheduler".into()) + .spawn(move || { + let scheduler = vcli_runtime::Scheduler::new( + vcli_runtime::SchedulerConfig::default(), + capture, + input, + perception, + clock, + cmd_rx, + sched_event_tx_for_thread, + ); + scheduler.run_until_shutdown(); + }) + .map_err(DaemonError::Io)?; + + let (shutdown_tx, shutdown_rx) = oneshot::channel::<()>(); + let shutdown_slot = Arc::new(Mutex::new(Some(shutdown_tx))); + let handler = DaemonHandler { + store: store.clone(), + bridge: bridge.clone(), + started_at: Instant::now(), + shutdown_tx: shutdown_slot.clone(), + }; + + install_signal_handlers(shutdown_slot.clone()) + .await + .map_err(DaemonError::Io)?; + + let server = IpcServer::bind(&cfg.socket.path, Arc::new(handler))?; + info!(socket = %cfg.socket.path.display(), "listening"); + + let serve_result = server.serve(shutdown_rx).await; + if let Err(e) = serve_result { + error!(error = %e, "ipc server terminated with error"); + } + + let _ = bridge.cmd_tx.send(SchedulerCommand::Shutdown); + emit_daemon_stopped(&sched_event_tx); + drop(sched_event_tx); + if let Err(e) = scheduler_join.join() { + error!("scheduler thread panicked: {e:?}"); + } + let _ = pump.await; + pid.release()?; + info!("vcli-daemon exited cleanly"); + Ok(()) +} From fc731983933325348ad0f864609f3cf10454042f Mon Sep 17 00:00:00 2001 From: Yixiang Zeng Date: Mon, 20 Apr 2026 15:51:02 -0700 Subject: [PATCH 21/25] vcli-daemon: binary main wires Config + runtime factory MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replaces the placeholder main with --version / --help plus the real run_foreground entrypoint. Builds a multi-thread tokio runtime, resolves platform defaults into a Config, and hands run_foreground the default_runtime_factory — which, for v0, returns MockCapture + MockInputSink + default Perception + SystemRuntimeClock. Real macOS backends land in a future lane. Exit code policy: 0 on clean shutdown, 1 on any DaemonError. Co-Authored-By: Claude Opus 4.7 (1M context) --- crates/vcli-daemon/src/bin/vcli-daemon.rs | 75 +++++++++++++++++++++-- 1 file changed, 71 insertions(+), 4 deletions(-) diff --git a/crates/vcli-daemon/src/bin/vcli-daemon.rs b/crates/vcli-daemon/src/bin/vcli-daemon.rs index f90f586..ed411e0 100644 --- a/crates/vcli-daemon/src/bin/vcli-daemon.rs +++ b/crates/vcli-daemon/src/bin/vcli-daemon.rs @@ -1,6 +1,73 @@ -//! Thin binary wrapper. All real work lives in `vcli_daemon::run::run_foreground`. +//! 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`. -fn main() { - // Placeholder until Task 12 wires real argv + run_foreground(). - eprintln!("vcli-daemon: not yet wired — see Task 12"); +use std::process::ExitCode; + +use vcli_daemon::{run_foreground, Config, DaemonError, RuntimeBackends}; + +fn main() -> ExitCode { + let args: Vec = 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 { + let capture: Box = Box::new(vcli_capture::MockCapture::empty()); + let input: std::sync::Arc = + std::sync::Arc::new(vcli_input::MockInputSink::new()); + let perception = vcli_perception::Perception::default(); + let clock: std::sync::Arc = + std::sync::Arc::new(vcli_runtime::SystemRuntimeClock::new()); + Ok(RuntimeBackends { + capture, + input, + perception, + clock, + }) } From b2b8e1bb763bd0477061e58f867614b81eb1f90a Mon Sep 17 00:00:00 2001 From: Yixiang Zeng Date: Mon, 20 Apr 2026 15:51:43 -0700 Subject: [PATCH 22/25] vcli-daemon: integration test for orphan program recovery Plants a Running row, starts the daemon, waits for the socket to appear, then sends Shutdown over IPC. After the daemon drains cleanly, reopens the store and asserts both the row state transitioned to Failed and the event log contains program.state_changed and program.failed. Co-Authored-By: Claude Opus 4.7 (1M context) --- .../tests/startup_orphan_recovery.rs | 88 +++++++++++++++++++ 1 file changed, 88 insertions(+) create mode 100644 crates/vcli-daemon/tests/startup_orphan_recovery.rs diff --git a/crates/vcli-daemon/tests/startup_orphan_recovery.rs b/crates/vcli-daemon/tests/startup_orphan_recovery.rs new file mode 100644 index 0000000..be472b3 --- /dev/null +++ b/crates/vcli-daemon/tests/startup_orphan_recovery.rs @@ -0,0 +1,88 @@ +//! Startup integration: plant a `running` row in a store before the daemon +//! starts, then drive `run_foreground` until the socket is listening and +//! shutdown cleanly, asserting: +//! * the row is now `failed(daemon_restart)` +//! * the event log contains `program.state_changed` and `program.failed`. + +use std::sync::Arc; +use std::time::Duration; + +use tempfile::tempdir; +use tokio::time::timeout; + +use vcli_core::ProgramState; +use vcli_daemon::{run_foreground, Config, RuntimeBackends}; +use vcli_ipc::socket_path::SocketPathOrigin; +use vcli_store::{NewProgram, Store}; + +fn noop_backends() -> Result { + Ok(RuntimeBackends { + capture: Box::new(vcli_capture::MockCapture::empty()), + input: Arc::new(vcli_input::MockInputSink::new()), + perception: vcli_perception::Perception::default(), + clock: Arc::new(vcli_runtime::SystemRuntimeClock::new()), + }) +} + +#[tokio::test] +async fn running_programs_transition_to_failed_on_startup() { + let dir = tempdir().unwrap(); + let data_root = dir.path().join("data"); + let log_dir = dir.path().join("logs"); + let sock = dir.path().join("vcli.sock"); + + let pid = vcli_core::ProgramId::new(); + { + let (mut store, _) = Store::open(&data_root).unwrap(); + store + .insert_program(&NewProgram { + id: pid, + name: "orphan", + source_json: "{}", + state: ProgramState::Pending, + submitted_at: 0, + labels_json: "{}", + }) + .unwrap(); + store.update_state(pid, ProgramState::Running, 1).unwrap(); + store.set_body_cursor(pid, 2).unwrap(); + } + + let cfg = Config::with_roots( + &data_root, + &log_dir, + vcli_ipc::SocketPath { + path: sock.clone(), + origin: SocketPathOrigin::Override, + }, + ); + + let factory: vcli_daemon::RuntimeFactory = Box::new(noop_backends); + let run = tokio::spawn(run_foreground(cfg, factory)); + + for _ in 0..50 { + if sock.exists() { + break; + } + tokio::time::sleep(Duration::from_millis(20)).await; + } + assert!(sock.exists(), "daemon never bound the socket"); + + let mut client = vcli_ipc::IpcClient::connect(&sock).await.unwrap(); + let resp = client.request(vcli_ipc::RequestOp::Shutdown).await.unwrap(); + assert!(matches!(resp.body, vcli_ipc::ResponseBody::Ok { .. })); + + timeout(Duration::from_secs(2), run) + .await + .unwrap() + .unwrap() + .unwrap(); + + let (store, _) = Store::open(&data_root).unwrap(); + let row = store.get_program(pid).unwrap(); + assert_eq!(row.state, ProgramState::Failed); + let events = store.stream_events(0, 100).unwrap(); + let types: Vec<&str> = events.iter().map(|e| e.type_tag.as_str()).collect(); + assert!(types.contains(&"program.state_changed")); + assert!(types.contains(&"program.failed")); +} From 2c476454b1bdc916e8b25cf63bced43f885f1c89 Mon Sep 17 00:00:00 2001 From: Yixiang Zeng Date: Mon, 20 Apr 2026 15:52:10 -0700 Subject: [PATCH 23/25] vcli-daemon: integration test for submit over IPC End-to-end: starts the daemon with mock backends, connects via IpcClient, sends Submit with a minimal valid program, and asserts the response carries a program_id. Then issues Shutdown so the daemon drains cleanly inside the test timeout. Co-Authored-By: Claude Opus 4.7 (1M context) --- crates/vcli-daemon/tests/submit_and_run.rs | 73 ++++++++++++++++++++++ 1 file changed, 73 insertions(+) create mode 100644 crates/vcli-daemon/tests/submit_and_run.rs diff --git a/crates/vcli-daemon/tests/submit_and_run.rs b/crates/vcli-daemon/tests/submit_and_run.rs new file mode 100644 index 0000000..9088d7b --- /dev/null +++ b/crates/vcli-daemon/tests/submit_and_run.rs @@ -0,0 +1,73 @@ +//! End-to-end: start the daemon with mock backends, submit a minimal program +//! over IPC, assert the daemon returns a program_id, then shut down. + +use std::sync::Arc; +use std::time::Duration; + +use tempfile::tempdir; +use tokio::time::timeout; + +use vcli_daemon::{run_foreground, Config, RuntimeBackends}; +use vcli_ipc::socket_path::SocketPathOrigin; +use vcli_ipc::{IpcClient, RequestOp, ResponseBody}; + +fn mocks() -> Result { + Ok(RuntimeBackends { + capture: Box::new(vcli_capture::MockCapture::empty()), + input: Arc::new(vcli_input::MockInputSink::new()), + perception: vcli_perception::Perception::default(), + clock: Arc::new(vcli_runtime::SystemRuntimeClock::new()), + }) +} + +#[tokio::test] +async fn submit_creates_program_row_and_returns_id() { + let dir = tempdir().unwrap(); + let sock = dir.path().join("vcli.sock"); + let cfg = Config::with_roots( + dir.path().join("data"), + dir.path().join("logs"), + vcli_ipc::SocketPath { + path: sock.clone(), + origin: SocketPathOrigin::Override, + }, + ); + + let factory: vcli_daemon::RuntimeFactory = Box::new(mocks); + let run = tokio::spawn(run_foreground(cfg, factory)); + + for _ in 0..50 { + if sock.exists() { + break; + } + tokio::time::sleep(Duration::from_millis(20)).await; + } + assert!(sock.exists()); + + let mut client = IpcClient::connect(&sock).await.unwrap(); + let program = serde_json::json!({ + "version": "0.1", + "name": "noop", + "trigger": { "kind": "on_submit" }, + "predicates": {}, + "watches": [], + "body": [], + }); + let resp = client.request(RequestOp::Submit { program }).await.unwrap(); + match resp.body { + ResponseBody::Ok { result, .. } => { + let pid = result["program_id"].as_str().unwrap(); + assert!(!pid.is_empty()); + } + ResponseBody::Err { error, .. } => panic!("submit failed: {error:?}"), + } + + let shut = client.request(RequestOp::Shutdown).await.unwrap(); + assert!(matches!(shut.body, ResponseBody::Ok { .. })); + + timeout(Duration::from_secs(2), run) + .await + .unwrap() + .unwrap() + .unwrap(); +} From 9d3e0aa8ced1e0712a2e96b310df07f97dba5386 Mon Sep 17 00:00:00 2001 From: Yixiang Zeng Date: Mon, 20 Apr 2026 15:52:36 -0700 Subject: [PATCH 24/25] vcli-daemon: integration test for graceful shutdown cleanup MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Starts the daemon, connects, issues Shutdown, and after the daemon drains asserts both the socket file and the pidfile are unlinked — observable readiness in reverse (Decision 1.2). Co-Authored-By: Claude Opus 4.7 (1M context) --- crates/vcli-daemon/tests/graceful_shutdown.rs | 58 +++++++++++++++++++ 1 file changed, 58 insertions(+) create mode 100644 crates/vcli-daemon/tests/graceful_shutdown.rs diff --git a/crates/vcli-daemon/tests/graceful_shutdown.rs b/crates/vcli-daemon/tests/graceful_shutdown.rs new file mode 100644 index 0000000..dea61b2 --- /dev/null +++ b/crates/vcli-daemon/tests/graceful_shutdown.rs @@ -0,0 +1,58 @@ +//! On shutdown, the socket file and pidfile should both be unlinked. + +use std::sync::Arc; +use std::time::Duration; + +use tempfile::tempdir; + +use vcli_daemon::{run_foreground, Config, RuntimeBackends}; +use vcli_ipc::socket_path::SocketPathOrigin; +use vcli_ipc::{IpcClient, RequestOp, ResponseBody}; + +fn mocks() -> Result { + Ok(RuntimeBackends { + capture: Box::new(vcli_capture::MockCapture::empty()), + input: Arc::new(vcli_input::MockInputSink::new()), + perception: vcli_perception::Perception::default(), + clock: Arc::new(vcli_runtime::SystemRuntimeClock::new()), + }) +} + +#[tokio::test] +async fn shutdown_unlinks_socket_file() { + let dir = tempdir().unwrap(); + let sock = dir.path().join("vcli.sock"); + let cfg = Config::with_roots( + dir.path().join("data"), + dir.path().join("logs"), + vcli_ipc::SocketPath { + path: sock.clone(), + origin: SocketPathOrigin::Override, + }, + ); + + let factory: vcli_daemon::RuntimeFactory = Box::new(mocks); + let run = tokio::spawn(run_foreground(cfg, factory)); + + for _ in 0..50 { + if sock.exists() { + break; + } + tokio::time::sleep(Duration::from_millis(20)).await; + } + assert!(sock.exists()); + + let mut client = IpcClient::connect(&sock).await.unwrap(); + let r = client.request(RequestOp::Shutdown).await.unwrap(); + assert!(matches!(r.body, ResponseBody::Ok { .. })); + + tokio::time::timeout(Duration::from_secs(2), run) + .await + .unwrap() + .unwrap() + .unwrap(); + + assert!(!sock.exists(), "socket should be unlinked after shutdown"); + let pidfile = dir.path().join("data").join("daemon.pid"); + assert!(!pidfile.exists(), "pidfile should be removed after shutdown"); +} From 97a6bdcafa8080ed4f7e44d3e81d3d8d6e12c25c Mon Sep 17 00:00:00 2001 From: Yixiang Zeng Date: Mon, 20 Apr 2026 15:53:41 -0700 Subject: [PATCH 25/25] vcli-daemon: rustfmt pass + Cargo.lock for libc dep Mechanical rustfmt reflow of config.rs, handler.rs, and graceful_shutdown.rs (no behavior changes). Also commits the Cargo.lock update from adding libc to vcli-daemon's deps in the shutdown lane commit. Co-Authored-By: Claude Opus 4.7 (1M context) --- Cargo.lock | 1 + crates/vcli-daemon/src/config.rs | 15 ++-- crates/vcli-daemon/src/handler.rs | 78 +++++++++---------- crates/vcli-daemon/tests/graceful_shutdown.rs | 5 +- 4 files changed, 53 insertions(+), 46 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 2d2b08e..e395a5f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2183,6 +2183,7 @@ dependencies = [ "dirs", "fs2", "futures-util", + "libc", "serde", "serde_json", "signal-hook-tokio", diff --git a/crates/vcli-daemon/src/config.rs b/crates/vcli-daemon/src/config.rs index 90ac6cc..e9ec09e 100644 --- a/crates/vcli-daemon/src/config.rs +++ b/crates/vcli-daemon/src/config.rs @@ -77,7 +77,10 @@ impl Config { #[cfg(target_os = "macos")] fn platform_data_root() -> DaemonResult { let home = dirs::home_dir().ok_or_else(|| DaemonError::Paths("no home dir".into()))?; - Ok(home.join("Library").join("Application Support").join("vcli")) + Ok(home + .join("Library") + .join("Application Support") + .join("vcli")) } #[cfg(target_os = "macos")] @@ -134,7 +137,10 @@ mod tests { fake_sock(&d.path().join("vcli.sock")), ); assert_eq!(cfg.pidfile_path(), d.path().join("data").join("daemon.pid")); - assert_eq!(cfg.log_file_path(), d.path().join("logs").join("daemon.log")); + assert_eq!( + cfg.log_file_path(), + d.path().join("logs").join("daemon.log") + ); } #[test] @@ -154,9 +160,6 @@ mod tests { #[cfg(target_os = "macos")] fn platform_defaults_use_library_on_macos() { let got = platform_data_root().unwrap(); - assert!( - got.ends_with("Library/Application Support/vcli"), - "{got:?}" - ); + assert!(got.ends_with("Library/Application Support/vcli"), "{got:?}"); } } diff --git a/crates/vcli-daemon/src/handler.rs b/crates/vcli-daemon/src/handler.rs index e993511..b497c0a 100644 --- a/crates/vcli-daemon/src/handler.rs +++ b/crates/vcli-daemon/src/handler.rs @@ -42,11 +42,7 @@ impl DaemonHandler { } } - async fn handle_submit( - &self, - id: RequestId, - program_json: serde_json::Value, - ) -> Response { + async fn handle_submit(&self, id: RequestId, program_json: serde_json::Value) -> Response { let program = match vcli_dsl::validate_value(&program_json) { Ok(v) => v.program, Err(e) => { @@ -96,7 +92,10 @@ impl DaemonHandler { } }; if let Err(e) = insert { - return Response::err(id, ErrorPayload::simple(ErrorCode::Internal, format!("{e}"))); + return Response::err( + id, + ErrorPayload::simple(ErrorCode::Internal, format!("{e}")), + ); } if let Err(e) = self.bridge.cmd_tx.send(SchedulerCommand::SubmitValidated { @@ -199,19 +198,18 @@ impl DaemonHandler { "orphan_blobs_deleted": r.orphan_blobs_deleted, }), ), - Ok(Err(e)) => { - Response::err(id, ErrorPayload::simple(ErrorCode::Internal, format!("{e}"))) - } - Err(e) => Response::err(id, ErrorPayload::simple(ErrorCode::Internal, format!("{e}"))), + Ok(Err(e)) => Response::err( + id, + ErrorPayload::simple(ErrorCode::Internal, format!("{e}")), + ), + Err(e) => Response::err( + id, + ErrorPayload::simple(ErrorCode::Internal, format!("{e}")), + ), } } - async fn handle_resume( - &self, - id: RequestId, - pid: ProgramId, - from_start: bool, - ) -> Response { + async fn handle_resume(&self, id: RequestId, pid: ProgramId, from_start: bool) -> Response { let store = self.store.clone(); let now_ms = vcli_core::clock::now_unix_ms(); let resume_result = tokio::task::spawn_blocking(move || { @@ -223,10 +221,7 @@ impl DaemonHandler { .map(|v| v.program) .map_err(|e| vcli_store::StoreError::Io { path: "".into(), - source: std::io::Error::new( - std::io::ErrorKind::InvalidData, - format!("{e}"), - ), + source: std::io::Error::new(std::io::ErrorKind::InvalidData, format!("{e}")), })?; Ok::<_, vcli_store::StoreError>((outcome, program)) }) @@ -251,10 +246,14 @@ impl DaemonHandler { id, ErrorPayload::simple(ErrorCode::UnknownProgram, "not found"), ), - Ok(Err(e)) => { - Response::err(id, ErrorPayload::simple(ErrorCode::Internal, format!("{e}"))) - } - Err(e) => Response::err(id, ErrorPayload::simple(ErrorCode::Internal, format!("{e}"))), + Ok(Err(e)) => Response::err( + id, + ErrorPayload::simple(ErrorCode::Internal, format!("{e}")), + ), + Err(e) => Response::err( + id, + ErrorPayload::simple(ErrorCode::Internal, format!("{e}")), + ), } } @@ -314,7 +313,10 @@ impl DaemonHandler { .collect(); Response::ok(id, serde_json::json!({ "items": items })) } - Err(e) => Response::err(id, ErrorPayload::simple(ErrorCode::Internal, format!("{e}"))), + Err(e) => Response::err( + id, + ErrorPayload::simple(ErrorCode::Internal, format!("{e}")), + ), } } @@ -342,10 +344,14 @@ impl DaemonHandler { id, ErrorPayload::simple(ErrorCode::UnknownProgram, "not found"), ), - Ok(Err(e)) => { - Response::err(id, ErrorPayload::simple(ErrorCode::Internal, format!("{e}"))) - } - Err(e) => Response::err(id, ErrorPayload::simple(ErrorCode::Internal, format!("{e}"))), + Ok(Err(e)) => Response::err( + id, + ErrorPayload::simple(ErrorCode::Internal, format!("{e}")), + ), + Err(e) => Response::err( + id, + ErrorPayload::simple(ErrorCode::Internal, format!("{e}")), + ), } } @@ -392,18 +398,12 @@ impl Handler for DaemonHandler { Ok(resp) } - async fn handle_stream( - &self, - id: RequestId, - op: RequestOp, - tx: StreamSender, - ) -> IpcResult<()> { + async fn handle_stream(&self, id: RequestId, op: RequestOp, tx: StreamSender) -> IpcResult<()> { match op { RequestOp::Events { follow } => self.stream_events(id, None, follow, tx).await, - RequestOp::Logs { - program_id, - follow, - } => self.stream_events(id, Some(program_id), follow, tx).await, + RequestOp::Logs { program_id, follow } => { + self.stream_events(id, Some(program_id), follow, tx).await + } RequestOp::Trace { program_id: _ } => { // v0 minimum: empty trace; server writes end_of_stream on return. let _ = tx; diff --git a/crates/vcli-daemon/tests/graceful_shutdown.rs b/crates/vcli-daemon/tests/graceful_shutdown.rs index dea61b2..d4a56ab 100644 --- a/crates/vcli-daemon/tests/graceful_shutdown.rs +++ b/crates/vcli-daemon/tests/graceful_shutdown.rs @@ -54,5 +54,8 @@ async fn shutdown_unlinks_socket_file() { assert!(!sock.exists(), "socket should be unlinked after shutdown"); let pidfile = dir.path().join("data").join("daemon.pid"); - assert!(!pidfile.exists(), "pidfile should be removed after shutdown"); + assert!( + !pidfile.exists(), + "pidfile should be removed after shutdown" + ); }