diff --git a/Cargo.lock b/Cargo.lock index 287771d..cea15e3 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -318,6 +318,15 @@ dependencies = [ "cfg-if", ] +[[package]] +name = "crossbeam-channel" +version = "0.5.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "82b8f8f868b36967f9606790d1903570de9ceaf870a7bf9fbbd3016d636a2cb2" +dependencies = [ + "crossbeam-utils", +] + [[package]] name = "crossbeam-deque" version = "0.8.6" @@ -1894,6 +1903,23 @@ dependencies = [ "vcli-core", ] +[[package]] +name = "vcli-runtime" +version = "0.0.1" +dependencies = [ + "crossbeam-channel", + "proptest", + "serde_json", + "tempfile", + "thiserror", + "tracing", + "vcli-capture", + "vcli-core", + "vcli-dsl", + "vcli-input", + "vcli-perception", +] + [[package]] name = "vcli-store" version = "0.0.1" diff --git a/Cargo.toml b/Cargo.toml index ff522c6..bda3c0f 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -7,6 +7,7 @@ members = [ "crates/vcli-input", "crates/vcli-ipc", "crates/vcli-perception", + "crates/vcli-runtime", "crates/vcli-store", ] @@ -38,6 +39,9 @@ hex = "0.4" tracing = "0.1" +# Added by vcli-runtime lane +crossbeam-channel = "0.5" + # Platform-specific (used by vcli-input + vcli-capture) core-graphics = { version = "0.24", features = ["highsierra"] } core-foundation = "0.10" diff --git a/crates/vcli-runtime/Cargo.toml b/crates/vcli-runtime/Cargo.toml new file mode 100644 index 0000000..eb2701e --- /dev/null +++ b/crates/vcli-runtime/Cargo.toml @@ -0,0 +1,77 @@ +[package] +name = "vcli-runtime" +version.workspace = true +edition.workspace = true +rust-version.workspace= true +license.workspace = true +repository.workspace = true +authors.workspace = true +description = "Deterministic 10 fps scheduler that wires Capture → Perception → Arbiter → Input for vcli programs." + +[dependencies] +vcli-core = { path = "../vcli-core" } +vcli-capture = { path = "../vcli-capture" } +vcli-input = { path = "../vcli-input" } +vcli-perception = { path = "../vcli-perception" } + +crossbeam-channel = { workspace = true } +thiserror = { workspace = true } +tracing = { workspace = true } +serde_json = { workspace = true } + +[dev-dependencies] +vcli-dsl = { path = "../vcli-dsl" } +proptest = { workspace = true } +tempfile = { workspace = true } + +[[test]] +name = "one_shot_watch" +path = "tests/scenarios/one_shot_watch.rs" + +[[test]] +name = "persistent_watch" +path = "tests/scenarios/persistent_watch.rs" + +[[test]] +name = "until_predicate" +path = "tests/scenarios/until_predicate.rs" + +[[test]] +name = "watch_timeout" +path = "tests/scenarios/watch_timeout.rs" + +[[test]] +name = "while_true_retry" +path = "tests/scenarios/while_true_retry.rs" + +[[test]] +name = "wait_for_timeout" +path = "tests/scenarios/wait_for_timeout.rs" + +[[test]] +name = "assert_failure" +path = "tests/scenarios/assert_failure.rs" + +[[test]] +name = "input_postcondition" +path = "tests/scenarios/input_postcondition.rs" + +[[test]] +name = "novelty_timeout" +path = "tests/scenarios/novelty_timeout.rs" + +[[test]] +name = "action_conflict" +path = "tests/scenarios/action_conflict.rs" + +[[test]] +name = "predicate_dedup" +path = "tests/scenarios/predicate_dedup.rs" + +[[test]] +name = "elapsed_ms_since_true" +path = "tests/scenarios/elapsed_ms_since_true.rs" + +[[test]] +name = "daemon_restart_marker" +path = "tests/scenarios/daemon_restart_marker.rs" diff --git a/crates/vcli-runtime/build.rs b/crates/vcli-runtime/build.rs new file mode 100644 index 0000000..988afba --- /dev/null +++ b/crates/vcli-runtime/build.rs @@ -0,0 +1,13 @@ +fn main() { + #[cfg(target_os = "macos")] + { + // vcli-runtime 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-runtime'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-runtime/src/arbiter.rs b/crates/vcli-runtime/src/arbiter.rs new file mode 100644 index 0000000..ab43246 --- /dev/null +++ b/crates/vcli-runtime/src/arbiter.rs @@ -0,0 +1,155 @@ +//! Action arbitration. Spec §403. + +use vcli_core::program::Priority; +use vcli_core::ProgramId; + +/// One pending action for the arbiter to consider. +#[derive(Debug, Clone)] +pub struct Candidate { + /// Source program. + pub program_id: ProgramId, + /// Static priority from `program.priority`. + pub priority: Priority, + /// The action / step payload (opaque to the arbiter). + pub payload: T, +} + +/// Outcome of arbitration for one candidate. +#[derive(Debug, Clone)] +pub struct Decision { + /// Source program. + pub program_id: ProgramId, + /// Payload (passed through unchanged). + pub payload: T, + /// True if this payload should dispatch. + pub dispatch: bool, + /// If not dispatched, the conflicting winner's id. + pub loser_of: Option, +} + +/// Resolve conflicts: at most one dispatch per tick. Ordering: +/// 1. highest `Priority` wins +/// 2. tiebreak by lexicographic `ProgramId` (larger wins, for determinism) +/// +/// If `candidates.len() <= 1` everyone dispatches. +pub fn resolve(candidates: Vec>) -> Vec> { + if candidates.len() <= 1 { + return candidates + .into_iter() + .map(|c| Decision { + program_id: c.program_id, + payload: c.payload, + dispatch: true, + loser_of: None, + }) + .collect(); + } + let winner = candidates + .iter() + .max_by(|a, b| { + a.priority + .cmp(&b.priority) + .then_with(|| a.program_id.to_string().cmp(&b.program_id.to_string())) + }) + .map(|c| c.program_id) + .expect("len > 1 handled above"); + candidates + .into_iter() + .map(|c| { + let is_winner = c.program_id == winner; + Decision { + program_id: c.program_id, + payload: c.payload, + dispatch: is_winner, + loser_of: if is_winner { None } else { Some(winner) }, + } + }) + .collect() +} + +#[cfg(test)] +mod tests { + use super::*; + + fn id(n: u8) -> ProgramId { + let s = format!("{n:02x}345678-1234-4567-8910-111213141516"); + s.parse().unwrap() + } + + #[test] + fn single_candidate_always_dispatches() { + let out = resolve(vec![Candidate { + program_id: id(1), + priority: Priority(0), + payload: "a", + }]); + assert_eq!(out.len(), 1); + assert!(out[0].dispatch); + assert!(out[0].loser_of.is_none()); + } + + #[test] + fn higher_priority_wins() { + let out = resolve(vec![ + Candidate { + program_id: id(1), + priority: Priority(0), + payload: "a", + }, + Candidate { + program_id: id(2), + priority: Priority(5), + payload: "b", + }, + ]); + let dispatched: Vec<_> = out.iter().filter(|d| d.dispatch).collect(); + assert_eq!(dispatched.len(), 1); + assert_eq!(dispatched[0].program_id, id(2)); + assert_eq!( + out.iter().find(|d| !d.dispatch).unwrap().loser_of, + Some(id(2)) + ); + } + + #[test] + fn tie_breaks_by_program_id_lex_desc() { + let out = resolve(vec![ + Candidate { + program_id: id(1), + priority: Priority(0), + payload: "a", + }, + Candidate { + program_id: id(2), + priority: Priority(0), + payload: "b", + }, + ]); + let winner = out.iter().find(|d| d.dispatch).unwrap().program_id; + assert_eq!(winner, id(2)); + } + + #[test] + fn three_way_drops_two_losers() { + let out = resolve(vec![ + Candidate { + program_id: id(1), + priority: Priority(3), + payload: "a", + }, + Candidate { + program_id: id(2), + priority: Priority(3), + payload: "b", + }, + Candidate { + program_id: id(3), + priority: Priority(3), + payload: "c", + }, + ]); + let dispatched: Vec<_> = out.iter().filter(|d| d.dispatch).collect(); + assert_eq!(dispatched.len(), 1); + assert_eq!(dispatched[0].program_id, id(3)); + } +} diff --git a/crates/vcli-runtime/src/body.rs b/crates/vcli-runtime/src/body.rs new file mode 100644 index 0000000..722a082 --- /dev/null +++ b/crates/vcli-runtime/src/body.rs @@ -0,0 +1,518 @@ +//! Body step executor: runs exactly one step per tick; advances `body_cursor` +//! on success. + +use std::collections::BTreeMap; +use std::sync::Arc; + +use vcli_core::geom::Point; +use vcli_core::step::{OnFail, OnTimeout, Step, Target}; +use vcli_core::{Frame, Predicate, ProgramId, UnixMs}; +use vcli_input::InputSink; +use vcli_perception::Perception; + +use crate::error::RuntimeError; +use crate::expr; + +/// Outcome of one body-step attempt. +#[derive(Debug)] +pub enum StepOutcome { + /// Step completed — advance `body_cursor` by 1. + Advanced, + /// Step is still waiting (e.g. `wait_for`, `sleep_ms` not yet elapsed). Do not advance. + Stalled, + /// Body finished (cursor already past end). + BodyComplete, + /// Program must transition to `failed`. + Failed(RuntimeError), +} + +/// Pending `sleep_ms` / `wait_for` deadline. The scheduler stashes this in +/// `RunningProgram` and the body executor consults it on subsequent ticks. +#[derive(Debug, Clone)] +pub enum BodyDefer { + /// Sleep until `wake_at_ms`. + Sleep { + /// Wall-clock ms after which the sleep is done. + wake_at_ms: UnixMs, + }, + /// Wait for predicate up to `deadline_ms`. + WaitFor { + /// Predicate name. + predicate: String, + /// Deadline. + deadline_ms: UnixMs, + /// Timeout behaviour. + on_timeout: OnTimeout, + }, +} + +/// Per-program body-executor state. +#[derive(Default, Clone, Debug)] +pub struct BodyState { + /// Active deferral. + pub deferred: Option, +} + +/// Execute one body step. The scheduler holds the borrow on `RunningProgram`, +/// so mutable fields (cursor + defer) live on `BodyState`; callers pass them +/// in. +#[allow(clippy::too_many_arguments)] +#[allow(clippy::too_many_lines)] +pub fn step_once( + program_id: ProgramId, + body: &[Step], + cursor: u32, + state: &mut BodyState, + predicates: &BTreeMap, + frame: &Frame, + now_ms: UnixMs, + assets: &BTreeMap>, + perception: &Perception, + input: &Arc, +) -> StepOutcome { + if let Some(def) = state.deferred.clone() { + match def { + BodyDefer::Sleep { wake_at_ms } => { + if now_ms >= wake_at_ms { + state.deferred = None; + return StepOutcome::Advanced; + } + return StepOutcome::Stalled; + } + BodyDefer::WaitFor { + predicate, + deadline_ms, + on_timeout, + } => { + let r = match perception.evaluate_named( + &predicate, + predicates, + frame, + now_ms, + assets, + Some(program_id), + ) { + Ok(r) => r, + Err(e) => return StepOutcome::Failed(RuntimeError::Perception(e.to_string())), + }; + if r.truthy { + state.deferred = None; + return StepOutcome::Advanced; + } + if now_ms >= deadline_ms { + state.deferred = None; + return match on_timeout { + OnTimeout::Continue | OnTimeout::Retry => StepOutcome::Advanced, + OnTimeout::Fail => StepOutcome::Failed(RuntimeError::WaitForTimeout { + predicate, + waited_ms: 0, + }), + }; + } + return StepOutcome::Stalled; + } + } + } + + let Some(step) = body.get(cursor as usize) else { + return StepOutcome::BodyComplete; + }; + + match step { + Step::SleepMs { ms } => { + state.deferred = Some(BodyDefer::Sleep { + wake_at_ms: now_ms.saturating_add(UnixMs::from(*ms)), + }); + StepOutcome::Stalled + } + Step::WaitFor { + predicate, + timeout_ms, + on_timeout, + } => { + state.deferred = Some(BodyDefer::WaitFor { + predicate: predicate.clone(), + deadline_ms: now_ms.saturating_add(UnixMs::from(*timeout_ms)), + on_timeout: *on_timeout, + }); + StepOutcome::Stalled + } + Step::Assert { predicate, on_fail } => { + let r = match perception.evaluate_named( + predicate, + predicates, + frame, + now_ms, + assets, + Some(program_id), + ) { + Ok(r) => r, + Err(e) => return StepOutcome::Failed(RuntimeError::Perception(e.to_string())), + }; + if r.truthy { + StepOutcome::Advanced + } else { + match on_fail { + OnFail::Continue => StepOutcome::Advanced, + OnFail::Fail => StepOutcome::Failed(RuntimeError::AssertFailed { + predicate: predicate.clone(), + }), + } + } + } + Step::Move { at } => dispatch_at( + at, + predicates, + frame, + now_ms, + assets, + perception, + program_id, + |p| { + input + .mouse_move(p) + .map_err(|e| RuntimeError::Input(e.to_string())) + }, + ), + Step::Click { at, button } => dispatch_at( + at, + predicates, + frame, + now_ms, + assets, + perception, + program_id, + |p| { + input + .click(p, *button, &[], 0) + .map_err(|e| RuntimeError::Input(e.to_string())) + }, + ), + Step::Scroll { at, dx, dy } => { + let _ = (dx, dy); + dispatch_at( + at, + predicates, + frame, + now_ms, + assets, + perception, + program_id, + |p| { + input + .mouse_move(p) + .map_err(|e| RuntimeError::Input(e.to_string())) + }, + ) + } + Step::Type { text } => match input.type_text(text) { + Ok(()) => StepOutcome::Advanced, + Err(e) => StepOutcome::Failed(RuntimeError::Input(e.to_string())), + }, + Step::Key { key, modifiers } => match input.key_combo(modifiers, key) { + Ok(()) => StepOutcome::Advanced, + Err(e) => StepOutcome::Failed(RuntimeError::Input(e.to_string())), + }, + } +} + +#[allow(clippy::too_many_arguments)] +fn resolve_target( + target: &Target, + predicates: &BTreeMap, + frame: &Frame, + now_ms: UnixMs, + assets: &BTreeMap>, + perception: &Perception, + program_id: ProgramId, +) -> Result { + match target { + Target::Absolute(p) => Ok(*p), + Target::Expression(s) => { + let e = expr::parse(s)?; + let r = perception + .evaluate_named( + e.predicate, + predicates, + frame, + now_ms, + assets, + Some(program_id), + ) + .map_err(|err| RuntimeError::Perception(err.to_string()))?; + match e.accessor { + expr::Accessor::MatchCenter => expr::resolve_center(&r), + expr::Accessor::MatchBbox => { + let bx = expr::resolve_bbox(&r)?; + Ok(Point { x: bx.x, y: bx.y }) + } + } + } + } +} + +#[allow(clippy::too_many_arguments)] +fn dispatch_at( + target: &Target, + predicates: &BTreeMap, + frame: &Frame, + now_ms: UnixMs, + assets: &BTreeMap>, + perception: &Perception, + program_id: ProgramId, + f: F, +) -> StepOutcome +where + F: FnOnce(Point) -> Result<(), RuntimeError>, +{ + let point = match resolve_target( + target, predicates, frame, now_ms, assets, perception, program_id, + ) { + Ok(p) => p, + Err(e) => return StepOutcome::Failed(e), + }; + match f(point) { + Ok(()) => StepOutcome::Advanced, + Err(e) => StepOutcome::Failed(e), + } +} + +/// Dispatch a single action step (Move/Click/Type/Key/Scroll) — used for watch +/// firings, which don't carry body-only control-flow steps. +/// +/// # Errors +/// +/// Returns `RuntimeError::Input` on OS dispatch failure, `RuntimeError::Perception` +/// if an expression target can't evaluate, or `RuntimeError::Internal` if a +/// body-only step slips through (validator normally rejects these in watches). +#[allow(clippy::too_many_arguments)] +pub fn dispatch_action( + step: &Step, + predicates: &BTreeMap, + frame: &Frame, + now_ms: UnixMs, + assets: &BTreeMap>, + perception: &Perception, + program_id: ProgramId, + input: &Arc, +) -> Result<(), RuntimeError> { + match step { + Step::Move { at } => { + let p = resolve_target( + at, predicates, frame, now_ms, assets, perception, program_id, + )?; + input + .mouse_move(p) + .map_err(|e| RuntimeError::Input(e.to_string())) + } + Step::Click { at, button } => { + let p = resolve_target( + at, predicates, frame, now_ms, assets, perception, program_id, + )?; + input + .click(p, *button, &[], 0) + .map_err(|e| RuntimeError::Input(e.to_string())) + } + Step::Scroll { at, .. } => { + let p = resolve_target( + at, predicates, frame, now_ms, assets, perception, program_id, + )?; + input + .mouse_move(p) + .map_err(|e| RuntimeError::Input(e.to_string())) + } + Step::Type { text } => input + .type_text(text) + .map_err(|e| RuntimeError::Input(e.to_string())), + Step::Key { key, modifiers } => input + .key_combo(modifiers, key) + .map_err(|e| RuntimeError::Input(e.to_string())), + Step::SleepMs { .. } | Step::WaitFor { .. } | Step::Assert { .. } => Err( + RuntimeError::Internal("body-only step used in watch".into()), + ), + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::Arc; + use vcli_core::action::{Button, Modifier}; + use vcli_core::frame::FrameFormat; + use vcli_core::geom::Rect; + + fn blank_frame() -> Frame { + Frame::new( + FrameFormat::Rgba8, + Rect { + x: 0, + y: 0, + w: 1, + h: 1, + }, + 4, + Arc::from(vec![0u8, 0, 0, 0]), + 0, + ) + } + + fn id() -> ProgramId { + "12345678-1234-4567-8910-111213141516".parse().unwrap() + } + + struct NopSink; + impl InputSink for NopSink { + fn mouse_move(&self, _: Point) -> Result<(), vcli_input::error::InputError> { + Ok(()) + } + fn click( + &self, + _: Point, + _: Button, + _: &[Modifier], + _: u32, + ) -> Result<(), vcli_input::error::InputError> { + Ok(()) + } + fn double_click(&self, _: Point, _: Button) -> Result<(), vcli_input::error::InputError> { + Ok(()) + } + fn drag( + &self, + _: Point, + _: &[vcli_input::sink::DragSegment], + _: Button, + ) -> Result<(), vcli_input::error::InputError> { + Ok(()) + } + fn type_text(&self, _: &str) -> Result<(), vcli_input::error::InputError> { + Ok(()) + } + fn key_combo(&self, _: &[Modifier], _: &str) -> Result<(), vcli_input::error::InputError> { + Ok(()) + } + } + + #[test] + fn absolute_click_advances() { + let preds = BTreeMap::new(); + let assets = BTreeMap::new(); + let p = Perception::new(); + let input: Arc = Arc::new(NopSink); + let mut st = BodyState::default(); + let body = vec![Step::Click { + at: Target::Absolute(Point { x: 1, y: 2 }), + button: Button::Left, + }]; + let out = step_once( + id(), + &body, + 0, + &mut st, + &preds, + &blank_frame(), + 0, + &assets, + &p, + &input, + ); + assert!(matches!(out, StepOutcome::Advanced)); + } + + #[test] + fn sleep_ms_stalls_then_advances() { + let preds = BTreeMap::new(); + let assets = BTreeMap::new(); + let p = Perception::new(); + let input: Arc = Arc::new(NopSink); + let mut st = BodyState::default(); + let body = vec![Step::SleepMs { ms: 100 }]; + let out = step_once( + id(), + &body, + 0, + &mut st, + &preds, + &blank_frame(), + 0, + &assets, + &p, + &input, + ); + assert!(matches!(out, StepOutcome::Stalled)); + let out2 = step_once( + id(), + &body, + 0, + &mut st, + &preds, + &blank_frame(), + 200, + &assets, + &p, + &input, + ); + assert!(matches!(out2, StepOutcome::Advanced)); + } + + #[test] + fn assert_fail_propagates_error() { + use vcli_core::predicate::{PredicateKind, Rgb}; + let mut preds = BTreeMap::new(); + preds.insert( + "blue".into(), + PredicateKind::ColorAt { + point: Point { x: 0, y: 0 }, + rgb: Rgb([0, 0, 255]), + tolerance: 0, + }, + ); + let assets = BTreeMap::new(); + let p = Perception::new(); + let input: Arc = Arc::new(NopSink); + let mut st = BodyState::default(); + let body = vec![Step::Assert { + predicate: "blue".into(), + on_fail: OnFail::Fail, + }]; + let out = step_once( + id(), + &body, + 0, + &mut st, + &preds, + &blank_frame(), + 0, + &assets, + &p, + &input, + ); + assert!(matches!( + out, + StepOutcome::Failed(RuntimeError::AssertFailed { .. }) + )); + } + + #[test] + fn cursor_past_end_is_body_complete() { + let preds = BTreeMap::new(); + let assets = BTreeMap::new(); + let p = Perception::new(); + let input: Arc = Arc::new(NopSink); + let mut st = BodyState::default(); + let body: Vec = vec![]; + let out = step_once( + id(), + &body, + 0, + &mut st, + &preds, + &blank_frame(), + 0, + &assets, + &p, + &input, + ); + assert!(matches!(out, StepOutcome::BodyComplete)); + } +} diff --git a/crates/vcli-runtime/src/clock.rs b/crates/vcli-runtime/src/clock.rs new file mode 100644 index 0000000..5f83a5f --- /dev/null +++ b/crates/vcli-runtime/src/clock.rs @@ -0,0 +1,128 @@ +//! Scheduler-facing clock. Combines a wall-clock read with a blocking sleep so +//! tests can substitute a manual clock that advances time without actually +//! sleeping. +//! +//! Not a supertrait of `vcli_core::Clock` — Rust 1.75 lacks trait-object +//! upcasting. Implementors forward `unix_ms` to an inner `vcli_core::Clock` or +//! compute it directly. + +use std::sync::{Arc, Condvar, Mutex}; +use std::time::Duration; + +use vcli_core::{Clock, SystemClock, UnixMs}; + +/// Scheduler-facing clock. `Send + Sync` because the scheduler holds it behind +/// `Arc`. +pub trait RuntimeClock: Send + Sync { + /// Wall-clock reading for event timestamps. + fn unix_ms(&self) -> UnixMs; + /// Block for `ms` milliseconds. `ManualClock` short-circuits under tests. + fn sleep_ms(&self, ms: u32); +} + +/// Production clock: wraps `vcli_core::SystemClock` + `std::thread::sleep`. +#[derive(Debug, Default)] +pub struct SystemRuntimeClock { + inner: SystemClock, +} + +impl SystemRuntimeClock { + /// Fresh instance. + #[must_use] + pub fn new() -> Self { + Self { + inner: SystemClock::new(), + } + } +} + +impl RuntimeClock for SystemRuntimeClock { + fn unix_ms(&self) -> UnixMs { + self.inner.unix_ms() + } + fn sleep_ms(&self, ms: u32) { + std::thread::sleep(Duration::from_millis(u64::from(ms))); + } +} + +/// Deterministic clock for tests. `unix_ms()` returns whatever the test set +/// via `set_unix_ms`; `sleep_ms` advances the clock by the same amount instead +/// of blocking. +#[derive(Debug, Clone)] +pub struct ManualClock { + inner: Arc<(Mutex, Condvar)>, +} + +impl ManualClock { + /// Create at `start_ms`. + #[must_use] + pub fn new(start_ms: UnixMs) -> Self { + Self { + inner: Arc::new((Mutex::new(start_ms), Condvar::new())), + } + } + + /// Jump the clock to `ms`. + /// + /// # Panics + /// + /// Panics if the internal mutex is poisoned. + pub fn set_unix_ms(&self, ms: UnixMs) { + let (lock, cv) = &*self.inner; + *lock.lock().unwrap() = ms; + cv.notify_all(); + } + + /// Advance the clock by `delta_ms`. + /// + /// # Panics + /// + /// Panics if the internal mutex is poisoned. + pub fn advance_ms(&self, delta_ms: u32) { + let (lock, cv) = &*self.inner; + let mut g = lock.lock().unwrap(); + *g = g.saturating_add(UnixMs::from(delta_ms)); + cv.notify_all(); + } +} + +impl RuntimeClock for ManualClock { + fn unix_ms(&self) -> UnixMs { + *self.inner.0.lock().unwrap() + } + fn sleep_ms(&self, ms: u32) { + self.advance_ms(ms); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn manual_clock_advance_adds_time() { + let c = ManualClock::new(1_000); + c.advance_ms(250); + assert_eq!(c.unix_ms(), 1_250); + } + + #[test] + fn manual_clock_sleep_advances_instead_of_blocks() { + let c = ManualClock::new(0); + let t0 = std::time::Instant::now(); + c.sleep_ms(10_000); + assert!( + t0.elapsed() < Duration::from_millis(100), + "sleep must not block" + ); + assert_eq!(c.unix_ms(), 10_000); + } + + #[test] + fn system_runtime_clock_is_monotonic_within_a_tick() { + let c = SystemRuntimeClock::new(); + let t1 = c.unix_ms(); + let t2 = c.unix_ms(); + assert!(t2 >= t1); + } +} diff --git a/crates/vcli-runtime/src/command.rs b/crates/vcli-runtime/src/command.rs new file mode 100644 index 0000000..a1cebe6 --- /dev/null +++ b/crates/vcli-runtime/src/command.rs @@ -0,0 +1,93 @@ +//! `SchedulerCommand` — messages sent from the daemon's tokio reactor into the +//! synchronous scheduler thread. Shape is pinned by spec §Threading model and +//! by the sibling vcli-daemon plan. + +use vcli_core::{Program, ProgramId}; + +/// Commands consumed on every tick before evaluation. +#[derive(Debug, Clone)] +pub enum SchedulerCommand { + /// Program was already validated by `vcli-dsl` and persisted by the + /// daemon. The scheduler inserts it at state `Pending` and fires the + /// trigger next tick. + SubmitValidated { + /// Daemon-assigned id. + program_id: ProgramId, + /// Parsed program. + program: Program, + }, + /// Terminate a program with `Cancelled`. + Cancel { + /// Target. + program_id: ProgramId, + /// Human reason (propagated into `program.state_changed.reason`). + reason: String, + }, + /// Force-start a `Waiting` program (bypasses its trigger — used by + /// `vcli resume`). + Start { + /// Target. + program_id: ProgramId, + }, + /// Re-insert a program at `Running` with a prior `body_cursor`. + ResumeRunning { + /// Target. + program_id: ProgramId, + /// Step index to resume at (`0` = from start). + from_step: u32, + /// Full program (daemon reloaded it from `SQLite`). + program: Program, + }, + /// Drain and exit `run_until_shutdown`. + Shutdown, +} + +#[cfg(test)] +mod tests { + use super::*; + use std::collections::BTreeMap; + use vcli_core::{program::DslVersion, trigger::Trigger}; + + fn sample_program() -> Program { + Program { + version: DslVersion(DslVersion::V0_1.to_string()), + name: "x".into(), + id: None, + trigger: Trigger::OnSubmit, + predicates: BTreeMap::new(), + watches: vec![], + body: vec![], + on_complete: None, + on_fail: None, + timeout_ms: None, + labels: BTreeMap::new(), + priority: vcli_core::Priority::default(), + } + } + + #[test] + fn submit_variant_roundtrips_basic_shape() { + let id: ProgramId = "12345678-1234-4567-8910-111213141516".parse().unwrap(); + let c = SchedulerCommand::SubmitValidated { + program_id: id, + program: sample_program(), + }; + match c { + SchedulerCommand::SubmitValidated { program_id, .. } => assert_eq!(program_id, id), + _ => panic!("wrong variant"), + } + } + + #[test] + fn cancel_keeps_reason_untouched() { + let id: ProgramId = "00000000-0000-4000-8000-000000000000".parse().unwrap(); + let c = SchedulerCommand::Cancel { + program_id: id, + reason: "user".into(), + }; + match c { + SchedulerCommand::Cancel { reason, .. } => assert_eq!(reason, "user"), + _ => panic!("wrong variant"), + } + } +} diff --git a/crates/vcli-runtime/src/error.rs b/crates/vcli-runtime/src/error.rs new file mode 100644 index 0000000..fb71c8b --- /dev/null +++ b/crates/vcli-runtime/src/error.rs @@ -0,0 +1,135 @@ +//! Typed errors with a stable `code()` string suitable for IPC wire values. + +use thiserror::Error; + +/// Stable error-code string embedded in `program.failed.reason` and returned +/// to callers via IPC. Matches spec §555 (CLI error codes) and §713 (internal +/// surfaces). +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ErrorCode { + /// `capture.grab_screen()` or `grab_window()` failed. + CaptureFailed, + /// `InputSink` returned an error or the kill switch fired. + InputFailed, + /// A predicate evaluator raised an error (unknown name, asset missing, decode). + PerceptionFailed, + /// `wait_for` ran out of budget. + WaitForTimeout, + /// An `assert` predicate was not truthy. + AssertFailed, + /// A program's `timeout_ms` elapsed before completion. + ProgramTimeout, + /// Input postcondition never observed (`novelty_timeout` reached). + NoveltyTimeout, + /// An expression failed to resolve (`$pred.match.center` on a non-match). + ExpressionUnresolved, + /// Daemon restart marker (transitioned by `Scheduler::on_startup`). + DaemonRestart, + /// Catch-all for programmer errors (unreachable panics demoted to errors). + Internal, +} + +impl ErrorCode { + /// Canonical wire-value string. + #[must_use] + pub fn as_str(self) -> &'static str { + match self { + Self::CaptureFailed => "capture_failed", + Self::InputFailed => "input_failed", + Self::PerceptionFailed => "perception_failed", + Self::WaitForTimeout => "wait_for_timeout", + Self::AssertFailed => "assert_failed", + Self::ProgramTimeout => "program_timeout", + Self::NoveltyTimeout => "novelty_timeout", + Self::ExpressionUnresolved => "expression_unresolved", + Self::DaemonRestart => "daemon_restart", + Self::Internal => "internal", + } + } +} + +/// Runtime-layer error. Carries a [`ErrorCode`] plus a human reason string. +#[derive(Debug, Error)] +pub enum RuntimeError { + /// Capture backend failure. + #[error("capture_failed: {0}")] + Capture(String), + /// Input dispatch failure. + #[error("input_failed: {0}")] + Input(String), + /// Perception evaluator failure. + #[error("perception_failed: {0}")] + Perception(String), + /// `wait_for` timed out. + #[error("wait_for_timeout: predicate={predicate} after {waited_ms}ms")] + WaitForTimeout { + /// Predicate name. + predicate: String, + /// Budget actually consumed. + waited_ms: u32, + }, + /// `assert` predicate not truthy. + #[error("assert_failed: predicate={predicate}")] + AssertFailed { + /// Predicate name. + predicate: String, + }, + /// Program-level timeout fired. + #[error("program_timeout: after {elapsed_ms}ms")] + ProgramTimeout { + /// How long we ran before tripping. + elapsed_ms: u32, + }, + /// Postcondition not observed. + #[error("novelty_timeout: no postcondition change within {timeout_ms}ms")] + NoveltyTimeout { + /// Budget configured on the Step. + timeout_ms: u32, + }, + /// Expression unresolved. + #[error("expression_unresolved: {0}")] + ExpressionUnresolved(String), + /// Internal invariant violation. + #[error("internal: {0}")] + Internal(String), +} + +impl RuntimeError { + /// Wire-stable code. Propagated to `program.failed.reason`. + #[must_use] + pub fn code(&self) -> ErrorCode { + match self { + Self::Capture(_) => ErrorCode::CaptureFailed, + Self::Input(_) => ErrorCode::InputFailed, + Self::Perception(_) => ErrorCode::PerceptionFailed, + Self::WaitForTimeout { .. } => ErrorCode::WaitForTimeout, + Self::AssertFailed { .. } => ErrorCode::AssertFailed, + Self::ProgramTimeout { .. } => ErrorCode::ProgramTimeout, + Self::NoveltyTimeout { .. } => ErrorCode::NoveltyTimeout, + Self::ExpressionUnresolved(_) => ErrorCode::ExpressionUnresolved, + Self::Internal(_) => ErrorCode::Internal, + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn code_strings_are_spec_stable() { + assert_eq!(ErrorCode::CaptureFailed.as_str(), "capture_failed"); + assert_eq!(ErrorCode::WaitForTimeout.as_str(), "wait_for_timeout"); + assert_eq!(ErrorCode::NoveltyTimeout.as_str(), "novelty_timeout"); + } + + #[test] + fn runtime_error_maps_to_code() { + let e = RuntimeError::WaitForTimeout { + predicate: "p".into(), + waited_ms: 1000, + }; + assert_eq!(e.code(), ErrorCode::WaitForTimeout); + assert_eq!(e.code().as_str(), "wait_for_timeout"); + } +} diff --git a/crates/vcli-runtime/src/event.rs b/crates/vcli-runtime/src/event.rs new file mode 100644 index 0000000..f0a7e11 --- /dev/null +++ b/crates/vcli-runtime/src/event.rs @@ -0,0 +1,70 @@ +//! Event emitter: wraps a `crossbeam_channel::Sender` and stamps every +//! emission with `clock.unix_ms()`. + +use std::sync::Arc; + +use crossbeam_channel::Sender; +use vcli_core::{Event, EventData}; + +use crate::clock::RuntimeClock; + +/// Stamp + send. The daemon owns the receiver side. +#[derive(Clone)] +pub struct EventEmitter { + tx: Sender, + clock: Arc, +} + +impl EventEmitter { + /// Constructor. + #[must_use] + pub fn new(tx: Sender, clock: Arc) -> Self { + Self { tx, clock } + } + + /// Emit `data` stamped with `clock.unix_ms()`. Returns `false` only if the + /// receiver has been dropped (daemon shut down before us). The scheduler + /// ignores the return value and keeps running; the daemon drains + /// remaining events after joining the scheduler thread. + pub fn emit(&self, data: EventData) -> bool { + let ev = Event { + at: self.clock.unix_ms(), + data, + }; + self.tx.send(ev).is_ok() + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::clock::ManualClock; + use crossbeam_channel::unbounded; + use vcli_core::ProgramId; + + fn sample_id() -> ProgramId { + "12345678-1234-4567-8910-111213141516".parse().unwrap() + } + + #[test] + fn emit_stamps_with_clock_now() { + let (tx, rx) = unbounded::(); + let clock: Arc = Arc::new(ManualClock::new(12_345)); + let em = EventEmitter::new(tx, clock); + assert!(em.emit(EventData::ProgramSubmitted { + program_id: sample_id(), + name: "x".into() + })); + let ev = rx.recv().unwrap(); + assert_eq!(ev.at, 12_345); + } + + #[test] + fn emit_returns_false_when_receiver_dropped() { + let (tx, rx) = unbounded::(); + let clock: Arc = Arc::new(ManualClock::new(0)); + let em = EventEmitter::new(tx, clock); + drop(rx); + assert!(!em.emit(EventData::DaemonStopped)); + } +} diff --git a/crates/vcli-runtime/src/expr.rs b/crates/vcli-runtime/src/expr.rs new file mode 100644 index 0000000..9baccb8 --- /dev/null +++ b/crates/vcli-runtime/src/expr.rs @@ -0,0 +1,155 @@ +//! Expression resolver: `$.match.center` and `$.match.bbox`. + +use vcli_core::geom::{Point, Rect}; +use vcli_core::predicate::PredicateResult; + +use crate::error::RuntimeError; + +/// Parts of an expression. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ParsedExpr<'a> { + /// Referenced predicate name. + pub predicate: &'a str, + /// Accessor path (`.match.center` / `.match.bbox`). + pub accessor: Accessor, +} + +/// Supported accessors. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Accessor { + /// `.match.center` + MatchCenter, + /// `.match.bbox` + MatchBbox, +} + +/// Parse an expression string. Returns [`RuntimeError::ExpressionUnresolved`] +/// on malformed input. +/// +/// # Errors +/// +/// Returns `ExpressionUnresolved` for non-`$...` strings or unknown accessors. +pub fn parse(s: &str) -> Result, RuntimeError> { + let rest = s + .strip_prefix('$') + .ok_or_else(|| RuntimeError::ExpressionUnresolved(format!("expected '$' prefix: {s}")))?; + let (pred, tail) = rest.split_once('.').ok_or_else(|| { + RuntimeError::ExpressionUnresolved(format!("expected '.': {s}")) + })?; + if pred.is_empty() { + return Err(RuntimeError::ExpressionUnresolved(format!( + "empty predicate name: {s}" + ))); + } + let accessor = match tail { + "match.center" => Accessor::MatchCenter, + "match.bbox" => Accessor::MatchBbox, + other => { + return Err(RuntimeError::ExpressionUnresolved(format!( + "unknown accessor '{other}'" + ))); + } + }; + Ok(ParsedExpr { + predicate: pred, + accessor, + }) +} + +/// Resolve a parsed expression's center against a predicate result. +/// +/// Uses `(w-1)/2` / `(h-1)/2` offsets so a 40-wide rect centers on pixel 19, +/// not 20 — spec §222's inclusive-pixel definition. +/// +/// # Errors +/// +/// Returns `ExpressionUnresolved` if the result has no `match_data` (the +/// predicate was truthy but non-spatial, e.g. `color_at`). +pub fn resolve_center(r: &PredicateResult) -> Result { + let md = r + .match_data + .as_ref() + .ok_or_else(|| RuntimeError::ExpressionUnresolved("predicate has no match_data".into()))?; + let w_minus_1 = u32::try_from(md.bbox.w.max(1) - 1).unwrap_or(0); + let h_minus_1 = u32::try_from(md.bbox.h.max(1) - 1).unwrap_or(0); + Ok(Point { + x: md.bbox.x + i32::try_from(w_minus_1 / 2).unwrap_or(0), + y: md.bbox.y + i32::try_from(h_minus_1 / 2).unwrap_or(0), + }) +} + +/// Resolve a parsed expression's bbox against a predicate result. +/// +/// # Errors +/// +/// Returns `ExpressionUnresolved` when there is no `match_data`. +pub fn resolve_bbox(r: &PredicateResult) -> Result { + let md = r + .match_data + .as_ref() + .ok_or_else(|| RuntimeError::ExpressionUnresolved("predicate has no match_data".into()))?; + Ok(md.bbox) +} + +#[cfg(test)] +mod tests { + use super::*; + use vcli_core::predicate::{Confidence, MatchData}; + + #[test] + fn parse_match_center() { + let p = parse("$skip.match.center").unwrap(); + assert_eq!(p.predicate, "skip"); + assert_eq!(p.accessor, Accessor::MatchCenter); + } + + #[test] + fn parse_match_bbox() { + let p = parse("$x.match.bbox").unwrap(); + assert_eq!(p.accessor, Accessor::MatchBbox); + } + + #[test] + fn parse_rejects_missing_dollar() { + assert!(matches!( + parse("skip.match.center"), + Err(RuntimeError::ExpressionUnresolved(_)) + )); + } + + #[test] + fn parse_rejects_unknown_accessor() { + assert!(matches!( + parse("$skip.match.topleft"), + Err(RuntimeError::ExpressionUnresolved(_)) + )); + } + + #[test] + fn resolve_center_averages_bbox() { + let r = PredicateResult { + truthy: true, + at: 0, + match_data: Some(MatchData { + bbox: Rect { + x: 10, + y: 20, + w: 40, + h: 20, + }, + confidence: Confidence(0.9), + }), + }; + assert_eq!(resolve_center(&r).unwrap(), Point { x: 29, y: 29 }); + } + + #[test] + fn resolve_without_match_data_errors() { + let r = PredicateResult { + truthy: true, + at: 0, + match_data: None, + }; + assert!(resolve_center(&r).is_err()); + } +} diff --git a/crates/vcli-runtime/src/lib.rs b/crates/vcli-runtime/src/lib.rs new file mode 100644 index 0000000..88a9ec5 --- /dev/null +++ b/crates/vcli-runtime/src/lib.rs @@ -0,0 +1,28 @@ +//! vcli-runtime — deterministic 10 fps scheduler. +//! +//! See the v0 design spec at `docs/superpowers/specs/2026-04-16-vcli-design.md` +//! — §Runtime & scheduler (§311) is the authoritative source for everything +//! in this crate. + +#![forbid(unsafe_code)] +#![deny(rust_2018_idioms)] +#![warn(missing_docs)] +#![warn(clippy::pedantic)] +#![allow(clippy::module_name_repetitions)] + +mod arbiter; +mod body; +pub mod clock; +mod command; +mod error; +mod event; +mod expr; +mod program; +mod scheduler; +mod triggers; +mod watches; + +pub use clock::{ManualClock, RuntimeClock, SystemRuntimeClock}; +pub use command::SchedulerCommand; +pub use error::{ErrorCode, RuntimeError}; +pub use scheduler::{Scheduler, SchedulerConfig}; diff --git a/crates/vcli-runtime/src/program.rs b/crates/vcli-runtime/src/program.rs new file mode 100644 index 0000000..0b8bfdc --- /dev/null +++ b/crates/vcli-runtime/src/program.rs @@ -0,0 +1,112 @@ +//! `RunningProgram` — the scheduler's in-memory view of a submitted program. + +use std::collections::HashMap; + +use vcli_core::state::ProgramState; +use vcli_core::{Program, UnixMs}; + +/// Scheduler-owned per-program state. The scheduler keys programs by +/// `ProgramId` in a `HashMap`, so the id is not duplicated on the value. +pub struct RunningProgram { + /// Parsed program. + pub program: Program, + /// Current lifecycle state. + pub state: ProgramState, + /// Wall-clock ms when the program entered `Running` (for `timeout_ms` + `Lifetime::TimeoutMs`). + pub running_since_ms: Option, + /// Active body step index. `Some(n)` while advancing; `None` when body exhausted. + pub body_cursor: Option, + /// Per-watch bookkeeping, keyed by watch index. Populated on entry to + /// `Running`; drained as watches retire. + pub watch_state: HashMap, + /// If set, the next successful transition emits `program.resumed{from_step}`. + pub resumed_from: Option, + /// Per-program body executor state (`SleepMs` / `WaitFor` accumulators). + /// Persists across ticks so in-flight waits advance between frames. + pub body_state: crate::body::BodyState, +} + +/// Per-watch runtime state. +#[derive(Debug, Default, Clone)] +pub struct WatchRuntime { + /// Last fire timestamp (for `throttle_ms`). None = has never fired. + pub last_fired_ms: Option, + /// Last tick's truthiness result (for false→true edge detection). + pub last_truthy: bool, + /// Whether the watch has already been retired (`OneShot` or `UntilPredicate` tripped). + pub retired: bool, +} + +impl RunningProgram { + /// Construct at `Pending` with default bookkeeping. + #[must_use] + pub fn pending(program: Program) -> Self { + Self { + program, + state: ProgramState::Pending, + running_since_ms: None, + body_cursor: None, + watch_state: HashMap::new(), + resumed_from: None, + body_state: crate::body::BodyState::default(), + } + } + + /// Whether `body_cursor` points past the last body step. + #[must_use] + pub fn body_complete(&self) -> bool { + match self.body_cursor { + Some(n) => usize::try_from(n).unwrap_or(usize::MAX) >= self.program.body.len(), + None => self.program.body.is_empty(), + } + } + + /// Count of watches that are not yet retired. + #[must_use] + pub fn active_watch_count(&self) -> usize { + self.watch_state.values().filter(|w| !w.retired).count() + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::collections::BTreeMap; + use vcli_core::{program::DslVersion, trigger::Trigger}; + + fn sample_program() -> Program { + Program { + version: DslVersion(DslVersion::V0_1.to_string()), + name: "t".into(), + id: None, + trigger: Trigger::OnSubmit, + predicates: BTreeMap::new(), + watches: vec![], + body: vec![], + on_complete: None, + on_fail: None, + timeout_ms: None, + labels: BTreeMap::new(), + priority: vcli_core::Priority::default(), + } + } + + #[test] + fn pending_is_initial_state() { + let rp = RunningProgram::pending(sample_program()); + assert_eq!(rp.state, ProgramState::Pending); + assert!(rp.body_complete(), "empty body is trivially complete"); + assert_eq!(rp.active_watch_count(), 0); + } + + #[test] + fn body_complete_detects_exhausted_cursor() { + let mut p = sample_program(); + p.body = vec![vcli_core::Step::SleepMs { ms: 10 }]; + let mut rp = RunningProgram::pending(p); + rp.body_cursor = Some(0); + assert!(!rp.body_complete()); + rp.body_cursor = Some(1); + assert!(rp.body_complete()); + } +} diff --git a/crates/vcli-runtime/src/scheduler.rs b/crates/vcli-runtime/src/scheduler.rs new file mode 100644 index 0000000..574138a --- /dev/null +++ b/crates/vcli-runtime/src/scheduler.rs @@ -0,0 +1,645 @@ +//! Scheduler entrypoint. Owns a `HashMap` and +//! advances one tick per `tick_interval_ms`. + +use std::collections::HashMap; +use std::sync::Arc; + +use crossbeam_channel::{Receiver, Sender}; +use vcli_capture::Capture; +use vcli_core::state::ProgramState; +use vcli_core::{Event, EventData, ProgramId}; +use vcli_input::InputSink; +use vcli_perception::Perception; + +use crate::clock::RuntimeClock; +use crate::command::SchedulerCommand; +use crate::event::EventEmitter; +use crate::program::RunningProgram; + +/// Per-program evaluation slice passed through `tick_once`: +/// `(program_id, truthies-by-watch-idx, retires-by-watch-idx)`. +type PerProgEval = (ProgramId, Vec<(u32, bool)>, Vec); + +/// Arbiter payload for a firing watch: `(watch_idx, steps, is_one_shot)`. +type WatchFirePayload = (u32, Vec, bool); + +/// Tunable knobs. +#[derive(Debug, Clone, Copy)] +pub struct SchedulerConfig { + /// Tick cadence. + pub tick_interval_ms: u32, + /// Soft budget per tick. + pub tick_budget_ms: u32, + /// Concurrent-program cap. + pub max_inflight: usize, +} + +impl Default for SchedulerConfig { + fn default() -> Self { + Self { + tick_interval_ms: 100, + tick_budget_ms: 90, + max_inflight: 256, + } + } +} + +/// The scheduler. +/// +/// Single clock storage: `Arc` covers both wall-clock reads +/// (via `unix_ms`) and tick pacing (via `sleep_ms`). `RuntimeClock` does not +/// extend `vcli_core::Clock` — that would require trait-object upcasting, +/// which is only stable from Rust 1.86 and our MSRV is 1.75. +pub struct Scheduler { + config: SchedulerConfig, + capture: Box, + input: Arc, + perception: Perception, + clock: Arc, + cmd_rx: Receiver, + event: EventEmitter, + programs: HashMap, +} + +impl Scheduler { + /// Construct. The daemon will call `run_until_shutdown()` from a dedicated + /// OS thread immediately after construction. + #[must_use] + pub fn new( + config: SchedulerConfig, + capture: Box, + input: Arc, + perception: Perception, + clock: Arc, + cmd_rx: Receiver, + event_tx: Sender, + ) -> Self { + let event = EventEmitter::new(event_tx, Arc::clone(&clock)); + Self { + config, + capture, + input, + perception, + clock, + cmd_rx, + event, + programs: HashMap::new(), + } + } + + /// Drive a single tick. Public so scenario tests can step deterministically + /// without a live wall-clock loop. + pub fn tick_once_pub(&mut self) { + self.drain_commands(); + self.tick_once(); + } + + /// Main loop. Ticks on each `tick_interval_ms` deadline; returns when a + /// `Shutdown` command is observed. Does NOT emit `daemon.stopped` — that + /// is the daemon's responsibility. + pub fn run_until_shutdown(mut self) { + loop { + if self.drain_commands() { + break; + } + self.tick_once(); + self.clock.sleep_ms(self.config.tick_interval_ms); + } + } + + /// Drain pending commands. Returns `true` if a `Shutdown` was observed. + fn drain_commands(&mut self) -> bool { + while let Ok(cmd) = self.cmd_rx.try_recv() { + match cmd { + SchedulerCommand::Shutdown => return true, + SchedulerCommand::SubmitValidated { + program_id, + program, + } => { + let mut rp = RunningProgram::pending(program); + rp.state = ProgramState::Waiting; + self.event.emit(EventData::ProgramSubmitted { + program_id, + name: rp.program.name.clone(), + }); + self.event.emit(EventData::ProgramStateChanged { + program_id, + from: ProgramState::Pending, + to: ProgramState::Waiting, + reason: "submitted".into(), + }); + self.programs.insert(program_id, rp); + } + SchedulerCommand::Cancel { program_id, reason } => { + if let Some(rp) = self.programs.get_mut(&program_id) { + let from = rp.state; + rp.state = ProgramState::Cancelled; + self.event.emit(EventData::ProgramStateChanged { + program_id, + from, + to: ProgramState::Cancelled, + reason, + }); + } + } + SchedulerCommand::Start { program_id } => { + if let Some(rp) = self.programs.get_mut(&program_id) { + let from = rp.state; + rp.state = ProgramState::Running; + rp.running_since_ms = Some(self.clock.unix_ms()); + self.event.emit(EventData::ProgramStateChanged { + program_id, + from, + to: ProgramState::Running, + reason: "start".into(), + }); + } + } + SchedulerCommand::ResumeRunning { + program_id, + from_step, + program, + } => { + let mut rp = RunningProgram::pending(program); + rp.state = ProgramState::Running; + rp.running_since_ms = Some(self.clock.unix_ms()); + rp.body_cursor = Some(from_step); + rp.resumed_from = Some(from_step); + self.event.emit(EventData::ProgramStateChanged { + program_id, + from: ProgramState::Failed, + to: ProgramState::Running, + reason: "resume".into(), + }); + self.event.emit(EventData::ProgramResumed { + program_id, + from_step, + }); + self.programs.insert(program_id, rp); + } + } + } + false + } + + #[allow(clippy::too_many_lines)] + fn tick_once(&mut self) { + let frame = match self.capture.grab_screen() { + Ok(f) => std::sync::Arc::new(f), + Err(e) => { + tracing::warn!("capture failed: {e}"); + self.event.emit(EventData::TickFrameSkipped { + reason: "capture_failed".into(), + }); + return; + } + }; + self.perception.clear(); + let now_ms = self.clock.unix_ms(); + let assets: std::collections::BTreeMap> = std::collections::BTreeMap::new(); + + let waiting: Vec = self + .programs + .iter() + .filter(|(_, p)| p.state == ProgramState::Waiting) + .map(|(id, _)| *id) + .collect(); + for id in waiting { + let fire = { + let p = &self.programs[&id]; + crate::triggers::trigger_fires( + &p.program.trigger, + &p.program.predicates, + &frame, + now_ms, + &assets, + &self.perception, + id, + ) + }; + match fire { + Ok(true) => { + let rp = self.programs.get_mut(&id).unwrap(); + rp.state = ProgramState::Running; + rp.running_since_ms = Some(now_ms); + rp.body_cursor = Some(0); + for (i, _) in rp.program.watches.iter().enumerate() { + rp.watch_state.insert( + u32::try_from(i).unwrap_or(u32::MAX), + crate::program::WatchRuntime::default(), + ); + } + self.event.emit(EventData::ProgramStateChanged { + program_id: id, + from: ProgramState::Waiting, + to: ProgramState::Running, + reason: "trigger_fired".into(), + }); + } + Ok(false) => {} + Err(e) => { + self.programs.get_mut(&id).unwrap().state = ProgramState::Failed; + self.event.emit(EventData::ProgramFailed { + program_id: id, + reason: e.code().as_str().into(), + step: None, + emit: None, + }); + } + } + } + + let running_ids: Vec = self + .programs + .iter() + .filter(|(_, p)| p.state == ProgramState::Running) + .map(|(id, _)| *id) + .collect(); + + // `truthies` covers every non-retired watch that produced a result. + // `retires` holds watches whose Lifetime::TimeoutMs window elapsed or + // whose UntilPredicate terminator tripped this tick. + let mut evals: Vec = Vec::new(); + let mut candidates: Vec> = Vec::new(); + let mut direct_fires: Vec<(ProgramId, u32, bool)> = Vec::new(); + + for id in running_ids.iter().copied() { + let rp = self.programs.get(&id).unwrap(); + let running_since = rp.running_since_ms.unwrap_or(now_ms); + let mut truthies: Vec<(u32, bool)> = Vec::new(); + let mut fires: Vec<(u32, Vec, bool)> = Vec::new(); + let mut retires: Vec = Vec::new(); + for (idx, w) in rp.program.watches.iter().enumerate() { + let idx_u32 = u32::try_from(idx).unwrap_or(u32::MAX); + let state = rp.watch_state.get(&idx_u32).cloned().unwrap_or_default(); + if state.retired { + continue; + } + // UntilPredicate: terminator truthy → retire without firing. + if let vcli_core::watch::Lifetime::UntilPredicate { name } = &w.lifetime { + match self.perception.evaluate_named( + name, + &rp.program.predicates, + &frame, + now_ms, + &assets, + Some(id), + ) { + Ok(r) if r.truthy => { + retires.push(idx_u32); + continue; + } + Ok(_) => {} + Err(e) => tracing::warn!("perception (until terminator): {e}"), + } + } + let truthy = match &w.when { + vcli_core::watch::WatchWhen::ByName(n) => match self.perception.evaluate_named( + n, + &rp.program.predicates, + &frame, + now_ms, + &assets, + Some(id), + ) { + Ok(r) => r.truthy, + Err(e) => { + tracing::warn!("perception: {e}"); + continue; + } + }, + vcli_core::watch::WatchWhen::Inline(p) => { + let mut tmp = rp.program.predicates.clone(); + let key = format!("__inline_{idx}"); + tmp.insert(key.clone(), *p.clone()); + match self.perception.evaluate_named( + &key, + &tmp, + &frame, + now_ms, + &assets, + Some(id), + ) { + Ok(r) => r.truthy, + Err(e) => { + tracing::warn!("perception: {e}"); + continue; + } + } + } + }; + truthies.push((idx_u32, truthy)); + match crate::watches::decide(w, &state, truthy, now_ms, running_since) { + crate::watches::WatchDecision::Fire => fires.push(( + idx_u32, + w.steps.clone(), + matches!(w.lifetime, vcli_core::watch::Lifetime::OneShot), + )), + crate::watches::WatchDecision::Skip => {} + crate::watches::WatchDecision::Retire => retires.push(idx_u32), + } + } + if let Some((idx, steps, one_shot)) = fires.into_iter().next() { + if steps.is_empty() { + direct_fires.push((id, idx, one_shot)); + } else { + candidates.push(crate::arbiter::Candidate { + program_id: id, + priority: rp.program.priority, + payload: (idx, steps, one_shot), + }); + } + } + evals.push((id, truthies, retires)); + } + + let decisions = crate::arbiter::resolve(candidates); + // Track which (program, watch_idx) actually dispatched so we can apply + // `after_fire` updates below. Also track programs whose dispatch errored + // so we skip their body step this tick. + let mut dispatched_fires: Vec<(ProgramId, u32, bool)> = Vec::new(); + let mut dispatch_failed: std::collections::HashSet = + std::collections::HashSet::new(); + + // Empty-step watches have nothing to arbitrate — emit watch.fired + // directly and carry them through the post-fire bookkeeping. + for (prog_id, watch_idx, one_shot) in &direct_fires { + self.event.emit(EventData::WatchFired { + program_id: *prog_id, + watch_index: *watch_idx, + predicate: "watch".into(), + }); + dispatched_fires.push((*prog_id, *watch_idx, *one_shot)); + } + for d in decisions { + let prog_id = d.program_id; + let (watch_idx, steps, one_shot) = d.payload; + if d.dispatch { + let preds = self.programs[&prog_id].program.predicates.clone(); + let mut err: Option = None; + for s in &steps { + let value = serde_json::to_value(s).unwrap_or(serde_json::Value::Null); + self.event.emit(EventData::ActionDispatched { + program_id: prog_id, + step: value, + target: None, + }); + if let Err(e) = crate::body::dispatch_action( + s, + &preds, + &frame, + now_ms, + &assets, + &self.perception, + prog_id, + &self.input, + ) { + err = Some(e); + break; + } + } + if let Some(e) = err { + dispatch_failed.insert(prog_id); + let rp = self.programs.get_mut(&prog_id).unwrap(); + let emit = rp.program.on_fail.as_ref().map(|f| f.emit.clone()); + rp.state = ProgramState::Failed; + self.event.emit(EventData::ProgramFailed { + program_id: prog_id, + reason: e.code().as_str().into(), + step: Some(format!("watch[{watch_idx}]")), + emit, + }); + continue; + } + self.event.emit(EventData::WatchFired { + program_id: prog_id, + watch_index: watch_idx, + predicate: "watch".into(), + }); + dispatched_fires.push((prog_id, watch_idx, one_shot)); + } else if let Some(w) = d.loser_of { + let value = serde_json::Value::Array( + steps + .iter() + .map(|s| serde_json::to_value(s).unwrap_or(serde_json::Value::Null)) + .collect(), + ); + self.event.emit(EventData::ActionDeferred { + program_id: prog_id, + step: value, + reason: serde_json::json!({ "conflict_with": w.to_string() }), + }); + } + } + + // Apply WatchRuntime updates: truthy edges, timeout retirements, and + // post-fire bookkeeping (last_fired_ms + OneShot retirement). + for (id, truthies, retires) in &evals { + let rp = self.programs.get_mut(id).unwrap(); + for (idx, t) in truthies { + let wr = rp.watch_state.entry(*idx).or_default(); + wr.last_truthy = *t; + } + for idx in retires { + let wr = rp.watch_state.entry(*idx).or_default(); + wr.retired = true; + } + } + for (id, idx, _one_shot) in &dispatched_fires { + let rp = self.programs.get_mut(id).unwrap(); + let watch = &rp.program.watches[*idx as usize]; + let wr = rp.watch_state.entry(*idx).or_default(); + crate::watches::after_fire(watch, wr, now_ms); + } + + for id in running_ids.iter().copied() { + if dispatch_failed.contains(&id) { + continue; + } + let rp = self.programs.get_mut(&id).unwrap(); + if rp.state != ProgramState::Running { + continue; + } + let cursor = rp.body_cursor.unwrap_or(0); + let body = rp.program.body.clone(); + let outcome = crate::body::step_once( + id, + &body, + cursor, + &mut rp.body_state, + &rp.program.predicates, + &frame, + now_ms, + &assets, + &self.perception, + &self.input, + ); + match outcome { + crate::body::StepOutcome::Advanced => { + rp.body_cursor = Some(cursor + 1); + if rp.body_complete() && rp.active_watch_count() == 0 { + let emit = rp.program.on_complete.as_ref().map(|c| c.emit.clone()); + rp.state = ProgramState::Completed; + self.event.emit(EventData::ProgramStateChanged { + program_id: id, + from: ProgramState::Running, + to: ProgramState::Completed, + reason: "body_complete".into(), + }); + self.event.emit(EventData::ProgramCompleted { + program_id: id, + emit, + }); + } + } + crate::body::StepOutcome::Stalled => {} + crate::body::StepOutcome::BodyComplete => { + if rp.active_watch_count() == 0 { + let emit = rp.program.on_complete.as_ref().map(|c| c.emit.clone()); + rp.state = ProgramState::Completed; + self.event.emit(EventData::ProgramStateChanged { + program_id: id, + from: ProgramState::Running, + to: ProgramState::Completed, + reason: "body_complete".into(), + }); + self.event.emit(EventData::ProgramCompleted { + program_id: id, + emit, + }); + } + } + crate::body::StepOutcome::Failed(err) => { + let emit = rp.program.on_fail.as_ref().map(|f| f.emit.clone()); + rp.state = ProgramState::Failed; + self.event.emit(EventData::ProgramFailed { + program_id: id, + reason: err.code().as_str().into(), + step: Some(format!("body[{cursor}]")), + emit, + }); + } + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::clock::ManualClock; + use crossbeam_channel::unbounded; + use std::collections::BTreeMap; + use vcli_capture::capture::{Capture, WindowDescriptor}; + use vcli_capture::error::CaptureError; + use vcli_core::action::{Button, Modifier}; + use vcli_core::frame::{Frame, FrameFormat}; + use vcli_core::geom::{Point, Rect}; + use vcli_core::{program::DslVersion, trigger::Trigger, Program}; + use vcli_input::error::InputError; + use vcli_input::sink::DragSegment; + + struct StaticCapture; + impl Capture for StaticCapture { + fn supported_formats(&self) -> &[FrameFormat] { + &[FrameFormat::Rgba8] + } + fn enumerate_windows(&self) -> Result, CaptureError> { + Ok(vec![]) + } + fn grab_screen(&mut self) -> Result { + Ok(Frame::new( + FrameFormat::Rgba8, + Rect { + x: 0, + y: 0, + w: 1, + h: 1, + }, + 4, + std::sync::Arc::from(vec![0u8, 0, 0, 0]), + 0, + )) + } + fn grab_window(&mut self, _: &WindowDescriptor) -> Result { + self.grab_screen() + } + } + + struct NopInput; + impl InputSink for NopInput { + fn mouse_move(&self, _: Point) -> Result<(), InputError> { + Ok(()) + } + fn click(&self, _: Point, _: Button, _: &[Modifier], _: u32) -> Result<(), InputError> { + Ok(()) + } + fn double_click(&self, _: Point, _: Button) -> Result<(), InputError> { + Ok(()) + } + fn drag(&self, _: Point, _: &[DragSegment], _: Button) -> Result<(), InputError> { + Ok(()) + } + fn type_text(&self, _: &str) -> Result<(), InputError> { + Ok(()) + } + fn key_combo(&self, _: &[Modifier], _: &str) -> Result<(), InputError> { + Ok(()) + } + } + + fn empty_program() -> Program { + Program { + version: DslVersion(DslVersion::V0_1.to_string()), + name: "x".into(), + id: None, + trigger: Trigger::OnSubmit, + predicates: BTreeMap::new(), + watches: vec![], + body: vec![], + on_complete: None, + on_fail: None, + timeout_ms: None, + labels: BTreeMap::new(), + priority: vcli_core::Priority::default(), + } + } + + #[test] + fn shutdown_exits_cleanly() { + let (cmd_tx, cmd_rx) = unbounded::(); + let (ev_tx, ev_rx) = unbounded::(); + let clock = std::sync::Arc::new(ManualClock::new(0)); + let perc = Perception::new(); + let sched = Scheduler::new( + SchedulerConfig::default(), + Box::new(StaticCapture), + std::sync::Arc::new(NopInput), + perc, + clock, + cmd_rx, + ev_tx, + ); + let id: ProgramId = "12345678-1234-4567-8910-111213141516".parse().unwrap(); + cmd_tx + .send(SchedulerCommand::SubmitValidated { + program_id: id, + program: empty_program(), + }) + .unwrap(); + cmd_tx.send(SchedulerCommand::Shutdown).unwrap(); + let handle = std::thread::spawn(move || sched.run_until_shutdown()); + handle.join().unwrap(); + let mut kinds: Vec = Vec::new(); + while let Ok(ev) = ev_rx.try_recv() { + kinds.push( + serde_json::to_value(&ev).unwrap()["type"] + .as_str() + .unwrap() + .to_string(), + ); + } + assert!(kinds.iter().any(|k| k == "program.submitted")); + assert!(kinds.iter().any(|k| k == "program.state_changed")); + } +} diff --git a/crates/vcli-runtime/src/triggers.rs b/crates/vcli-runtime/src/triggers.rs new file mode 100644 index 0000000..322704f --- /dev/null +++ b/crates/vcli-runtime/src/triggers.rs @@ -0,0 +1,148 @@ +//! Trigger evaluation. A program in `Waiting` advances to `Running` as soon +//! as its trigger fires. Spec §425. + +use std::collections::BTreeMap; + +use vcli_core::{Frame, Predicate, ProgramId, Trigger, UnixMs}; +use vcli_perception::Perception; + +use crate::error::RuntimeError; + +/// Return `true` if the trigger says the program may start this tick. +/// +/// `Manual` triggers never auto-fire; they require a `SchedulerCommand::Start`. +/// +/// # Errors +/// +/// Propagates [`RuntimeError::Perception`] when evaluating `OnPredicate`. +pub fn trigger_fires( + trig: &Trigger, + predicates: &BTreeMap, + frame: &Frame, + now_ms: UnixMs, + assets: &BTreeMap>, + perception: &Perception, + program_id: ProgramId, +) -> Result { + match trig { + Trigger::OnSubmit => Ok(true), + Trigger::Manual => Ok(false), + Trigger::OnPredicate { name } => { + let r = perception + .evaluate_named(name, predicates, frame, now_ms, assets, Some(program_id)) + .map_err(|e| RuntimeError::Perception(e.to_string()))?; + Ok(r.truthy) + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::Arc; + use vcli_core::frame::FrameFormat; + use vcli_core::geom::{Point, Rect}; + use vcli_core::predicate::{PredicateKind, Rgb}; + + fn red_frame() -> Frame { + Frame::new( + FrameFormat::Rgba8, + Rect { + x: 0, + y: 0, + w: 1, + h: 1, + }, + 4, + Arc::from(vec![255u8, 0, 0, 255]), + 0, + ) + } + + fn some_id() -> ProgramId { + "12345678-1234-4567-8910-111213141516".parse().unwrap() + } + + #[test] + fn on_submit_always_fires() { + let preds = BTreeMap::new(); + let assets = BTreeMap::new(); + let p = Perception::new(); + let out = trigger_fires( + &Trigger::OnSubmit, + &preds, + &red_frame(), + 0, + &assets, + &p, + some_id(), + ) + .unwrap(); + assert!(out); + } + + #[test] + fn manual_never_auto_fires() { + let preds = BTreeMap::new(); + let assets = BTreeMap::new(); + let p = Perception::new(); + let out = trigger_fires( + &Trigger::Manual, + &preds, + &red_frame(), + 0, + &assets, + &p, + some_id(), + ) + .unwrap(); + assert!(!out); + } + + #[test] + fn on_predicate_defers_to_perception() { + let mut preds = BTreeMap::new(); + preds.insert( + "is_red".into(), + PredicateKind::ColorAt { + point: Point { x: 0, y: 0 }, + rgb: Rgb([255, 0, 0]), + tolerance: 0, + }, + ); + let assets = BTreeMap::new(); + let p = Perception::new(); + let out = trigger_fires( + &Trigger::OnPredicate { + name: "is_red".into(), + }, + &preds, + &red_frame(), + 100, + &assets, + &p, + some_id(), + ) + .unwrap(); + assert!(out); + } + + #[test] + fn on_predicate_unknown_name_errors() { + let preds = BTreeMap::new(); + let assets = BTreeMap::new(); + let p = Perception::new(); + let out = trigger_fires( + &Trigger::OnPredicate { + name: "nope".into(), + }, + &preds, + &red_frame(), + 0, + &assets, + &p, + some_id(), + ); + assert!(matches!(out, Err(RuntimeError::Perception(_)))); + } +} diff --git a/crates/vcli-runtime/src/watches.rs b/crates/vcli-runtime/src/watches.rs new file mode 100644 index 0000000..1d6352d --- /dev/null +++ b/crates/vcli-runtime/src/watches.rs @@ -0,0 +1,135 @@ +//! Watch lifetime bookkeeping. Spec §416. + +use vcli_core::watch::Lifetime; +use vcli_core::{UnixMs, Watch}; + +use crate::program::WatchRuntime; + +/// Decision for this tick. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum WatchDecision { + /// Do not fire and do not retire. + Skip, + /// Fire the watch (run its `do` steps). + Fire, + /// Retire without firing (e.g. `UntilPredicate` went truthy mid-tick). + Retire, +} + +/// Decide for a single watch given its current `state` and the *new* truthiness +/// result from this tick. Does NOT mutate; the scheduler applies mutation +/// (`state.last_truthy` / `state.last_fired_ms` / `state.retired`) only after +/// the watch's `do` steps actually dispatch (so a kill-switch or arbiter drop +/// does not consume the firing budget). +#[must_use] +pub fn decide( + watch: &Watch, + state: &WatchRuntime, + truthy_now: bool, + now_ms: UnixMs, + running_since_ms: UnixMs, +) -> WatchDecision { + if state.retired { + return WatchDecision::Skip; + } + if let Lifetime::TimeoutMs { ms } = watch.lifetime { + if now_ms.saturating_sub(running_since_ms) >= UnixMs::from(ms) { + return WatchDecision::Retire; + } + } + if state.last_truthy || !truthy_now { + return WatchDecision::Skip; + } + if let Some(last) = state.last_fired_ms { + if now_ms.saturating_sub(last) < UnixMs::from(watch.throttle_ms) { + return WatchDecision::Skip; + } + } + WatchDecision::Fire +} + +/// Apply post-firing retirement rules for `OneShot`. +pub fn after_fire(watch: &Watch, state: &mut WatchRuntime, now_ms: UnixMs) { + state.last_fired_ms = Some(now_ms); + if matches!(watch.lifetime, Lifetime::OneShot) { + state.retired = true; + } +} + +#[cfg(test)] +mod tests { + use super::*; + use vcli_core::watch::{Lifetime, WatchWhen}; + + fn mk_watch(lifetime: Lifetime, throttle_ms: u32) -> Watch { + Watch { + when: WatchWhen::ByName("p".into()), + steps: vec![], + throttle_ms, + lifetime, + } + } + + #[test] + fn fires_on_false_to_true_edge() { + let w = mk_watch(Lifetime::Persistent, 0); + let st = WatchRuntime::default(); + assert_eq!(decide(&w, &st, true, 100, 0), WatchDecision::Fire); + } + + #[test] + fn skips_on_level_high() { + let w = mk_watch(Lifetime::Persistent, 0); + let st = WatchRuntime { + last_truthy: true, + ..Default::default() + }; + assert_eq!(decide(&w, &st, true, 100, 0), WatchDecision::Skip); + } + + #[test] + fn throttle_suppresses_rapid_refires() { + let w = mk_watch(Lifetime::Persistent, 500); + let st = WatchRuntime { + last_truthy: false, + last_fired_ms: Some(100), + retired: false, + }; + assert_eq!(decide(&w, &st, true, 400, 0), WatchDecision::Skip); + assert_eq!(decide(&w, &st, true, 700, 0), WatchDecision::Fire); + } + + #[test] + fn one_shot_retires_after_fire() { + let w = mk_watch(Lifetime::OneShot, 0); + let mut st = WatchRuntime::default(); + after_fire(&w, &mut st, 50); + assert!(st.retired); + } + + #[test] + fn persistent_does_not_retire_after_fire() { + let w = mk_watch(Lifetime::Persistent, 0); + let mut st = WatchRuntime::default(); + after_fire(&w, &mut st, 50); + assert!(!st.retired); + assert_eq!(st.last_fired_ms, Some(50)); + } + + #[test] + fn timeout_ms_retires_when_window_elapsed() { + let w = mk_watch(Lifetime::TimeoutMs { ms: 1_000 }, 0); + let st = WatchRuntime::default(); + assert_eq!(decide(&w, &st, true, 2_001, 1_000), WatchDecision::Retire); + } + + #[test] + fn retired_watch_is_inert() { + let w = mk_watch(Lifetime::Persistent, 0); + let st = WatchRuntime { + retired: true, + ..Default::default() + }; + assert_eq!(decide(&w, &st, true, 100, 0), WatchDecision::Skip); + } +} diff --git a/crates/vcli-runtime/tests/common/mock_capture.rs b/crates/vcli-runtime/tests/common/mock_capture.rs new file mode 100644 index 0000000..176a7c1 --- /dev/null +++ b/crates/vcli-runtime/tests/common/mock_capture.rs @@ -0,0 +1,71 @@ +//! Scripted Capture: returns a queue of pre-built Frames, in order, then +//! repeats the last one. `enumerate_windows` returns an empty list. + +use std::sync::{Arc, Mutex}; + +use vcli_capture::capture::{Capture, WindowDescriptor}; +use vcli_capture::error::CaptureError; +use vcli_core::frame::{Frame, FrameFormat}; +use vcli_core::geom::Rect; + +pub struct ScriptedCapture { + frames: Mutex>, + cursor: Mutex, +} + +impl ScriptedCapture { + #[must_use] + pub fn new(frames: Vec) -> Self { + Self { + frames: Mutex::new(frames), + cursor: Mutex::new(0), + } + } + + #[must_use] + pub fn solid(width: u32, height: u32, rgba: [u8; 4]) -> Frame { + let n = (width * height) as usize; + let mut buf = Vec::with_capacity(n * 4); + for _ in 0..n { + buf.extend_from_slice(&rgba); + } + Frame::new( + FrameFormat::Rgba8, + Rect { + x: 0, + y: 0, + w: i32::try_from(width).unwrap(), + h: i32::try_from(height).unwrap(), + }, + 4, + Arc::from(buf), + 0, + ) + } +} + +impl Capture for ScriptedCapture { + fn supported_formats(&self) -> &[FrameFormat] { + &[FrameFormat::Rgba8] + } + + fn enumerate_windows(&self) -> Result, CaptureError> { + Ok(vec![]) + } + + fn grab_screen(&mut self) -> Result { + let frames = self.frames.lock().unwrap(); + let mut cursor = self.cursor.lock().unwrap(); + if frames.is_empty() { + return Ok(Self::solid(1, 1, [0, 0, 0, 0])); + } + let i = (*cursor).min(frames.len() - 1); + let out = frames[i].clone(); + *cursor = (i + 1).min(frames.len() - 1); + Ok(out) + } + + fn grab_window(&mut self, _: &WindowDescriptor) -> Result { + self.grab_screen() + } +} diff --git a/crates/vcli-runtime/tests/common/mock_input.rs b/crates/vcli-runtime/tests/common/mock_input.rs new file mode 100644 index 0000000..8d73161 --- /dev/null +++ b/crates/vcli-runtime/tests/common/mock_input.rs @@ -0,0 +1,76 @@ +//! Recording InputSink: captures every method call into a `Vec` for assertion. + +use std::sync::Mutex; + +use vcli_core::action::{Button, Modifier}; +use vcli_core::geom::Point; +use vcli_input::error::InputError; +use vcli_input::sink::{DragSegment, InputSink}; + +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum Call { + Move(Point), + Click(Point, Button), + DoubleClick(Point, Button), + Drag(Point, Vec, Button), + Type(String), + Key(Vec, String), +} + +#[derive(Default)] +pub struct RecordingInputSink { + pub calls: Mutex>, +} + +impl RecordingInputSink { + #[must_use] + pub fn new() -> Self { + Self::default() + } + + #[must_use] + pub fn calls(&self) -> Vec { + self.calls.lock().unwrap().clone() + } +} + +impl InputSink for RecordingInputSink { + fn mouse_move(&self, to: Point) -> Result<(), InputError> { + self.calls.lock().unwrap().push(Call::Move(to)); + Ok(()) + } + + fn click(&self, at: Point, button: Button, _: &[Modifier], _: u32) -> Result<(), InputError> { + self.calls.lock().unwrap().push(Call::Click(at, button)); + Ok(()) + } + + fn double_click(&self, at: Point, button: Button) -> Result<(), InputError> { + self.calls + .lock() + .unwrap() + .push(Call::DoubleClick(at, button)); + Ok(()) + } + + fn drag(&self, from: Point, segs: &[DragSegment], button: Button) -> Result<(), InputError> { + self.calls + .lock() + .unwrap() + .push(Call::Drag(from, segs.to_vec(), button)); + Ok(()) + } + + fn type_text(&self, s: &str) -> Result<(), InputError> { + self.calls.lock().unwrap().push(Call::Type(s.to_string())); + Ok(()) + } + + fn key_combo(&self, mods: &[Modifier], k: &str) -> Result<(), InputError> { + self.calls + .lock() + .unwrap() + .push(Call::Key(mods.to_vec(), k.to_string())); + Ok(()) + } +} diff --git a/crates/vcli-runtime/tests/common/mod.rs b/crates/vcli-runtime/tests/common/mod.rs new file mode 100644 index 0000000..f7a4060 --- /dev/null +++ b/crates/vcli-runtime/tests/common/mod.rs @@ -0,0 +1,40 @@ +//! Shared test helpers for vcli-runtime scenario tests. + +#![allow(dead_code)] // each scenario uses a different subset + +use std::collections::BTreeMap; + +pub use crossbeam_channel::{unbounded, Receiver}; +pub use vcli_core::{Event, ProgramId}; +pub use vcli_runtime::{Scheduler, SchedulerCommand, SchedulerConfig}; + +pub mod mock_capture; +pub mod mock_input; + +pub use mock_capture::ScriptedCapture; +pub use mock_input::RecordingInputSink; + +/// Extract the `type` tag from a serialized event, for pattern-matching in assertions. +#[must_use] +pub fn event_type(e: &Event) -> String { + serde_json::to_value(e).unwrap()["type"] + .as_str() + .unwrap_or("") + .to_string() +} + +/// Drain a channel into a Vec (non-blocking). +#[must_use] +pub fn drain_events(rx: &Receiver) -> Vec { + let mut out = Vec::new(); + while let Ok(e) = rx.try_recv() { + out.push(e); + } + out +} + +/// Build an empty predicate map (convenience for scenarios). +#[must_use] +pub fn empty_predicates() -> BTreeMap { + BTreeMap::new() +} diff --git a/crates/vcli-runtime/tests/scenarios/action_conflict.rs b/crates/vcli-runtime/tests/scenarios/action_conflict.rs new file mode 100644 index 0000000..2afc3d2 --- /dev/null +++ b/crates/vcli-runtime/tests/scenarios/action_conflict.rs @@ -0,0 +1,103 @@ +//! Scenario: Two programs firing on the same tick arbitrate to one winner. + +#[path = "../common/mod.rs"] +mod common; + +use std::collections::BTreeMap; +use std::sync::Arc; +use std::time::Duration; + +use common::*; +use vcli_core::action::Button; +use vcli_core::geom::Point; +use vcli_core::predicate::{PredicateKind, Rgb}; +use vcli_core::step::{Step, Target}; +use vcli_core::watch::{Lifetime, Watch, WatchWhen}; +use vcli_core::{program::DslVersion, trigger::Trigger, Program}; +use vcli_perception::Perception; +use vcli_runtime::clock::ManualClock; + +fn build_program(name: &str) -> Program { + let mut preds = BTreeMap::new(); + preds.insert( + "red".into(), + PredicateKind::ColorAt { + point: Point { x: 0, y: 0 }, + rgb: Rgb([255, 0, 0]), + tolerance: 0, + }, + ); + Program { + version: DslVersion(DslVersion::V0_1.to_string()), + name: name.into(), + id: None, + trigger: Trigger::OnSubmit, + predicates: preds, + watches: vec![Watch { + when: WatchWhen::ByName("red".into()), + steps: vec![Step::Click { + at: Target::Absolute(Point { x: 0, y: 0 }), + button: Button::Left, + }], + throttle_ms: 0, + lifetime: Lifetime::OneShot, + }], + body: vec![], + on_complete: None, + on_fail: None, + timeout_ms: None, + labels: BTreeMap::new(), + priority: Default::default(), + } +} + +#[test] +fn two_programs_same_frame_arbitrate_to_one_winner() { + let red = ScriptedCapture::solid(1, 1, [255, 0, 0, 255]); + let capture = Box::new(ScriptedCapture::new(vec![red])); + let (cmd_tx, cmd_rx) = unbounded::(); + let (ev_tx, ev_rx) = unbounded::(); + let input = Arc::new(RecordingInputSink::new()); + let clock = Arc::new(ManualClock::new(0)); + + let id1: ProgramId = "11111111-1111-4567-8910-111213141516".parse().unwrap(); + let id2: ProgramId = "22222222-2222-4567-8910-111213141516".parse().unwrap(); + + let sched = Scheduler::new( + SchedulerConfig::default(), + capture, + input.clone(), + Perception::new(), + clock, + cmd_rx, + ev_tx, + ); + cmd_tx + .send(SchedulerCommand::SubmitValidated { + program_id: id1, + program: build_program("p1"), + }) + .unwrap(); + cmd_tx + .send(SchedulerCommand::SubmitValidated { + program_id: id2, + program: build_program("p2"), + }) + .unwrap(); + let h = std::thread::spawn(move || sched.run_until_shutdown()); + std::thread::sleep(Duration::from_millis(300)); + cmd_tx.send(SchedulerCommand::Shutdown).unwrap(); + h.join().unwrap(); + + let events = drain_events(&ev_rx); + let dispatched = events + .iter() + .filter(|e| event_type(e) == "action.dispatched") + .count(); + let deferred = events + .iter() + .filter(|e| event_type(e) == "action.deferred") + .count(); + assert_eq!(dispatched, 1, "exactly one winner"); + assert_eq!(deferred, 1, "exactly one loser"); +} diff --git a/crates/vcli-runtime/tests/scenarios/assert_failure.rs b/crates/vcli-runtime/tests/scenarios/assert_failure.rs new file mode 100644 index 0000000..efd2d05 --- /dev/null +++ b/crates/vcli-runtime/tests/scenarios/assert_failure.rs @@ -0,0 +1,83 @@ +//! Scenario: body `assert` with `on_fail: Fail` fails the program. + +#[path = "../common/mod.rs"] +mod common; + +use std::collections::BTreeMap; +use std::sync::Arc; +use std::time::Duration; + +use common::*; +use vcli_core::geom::Point; +use vcli_core::predicate::{PredicateKind, Rgb}; +use vcli_core::step::{OnFail, Step}; +use vcli_core::{program::DslVersion, trigger::Trigger, Program}; +use vcli_perception::Perception; +use vcli_runtime::clock::ManualClock; + +#[test] +fn assert_fail_on_fail_fails_program() { + let blue = ScriptedCapture::solid(1, 1, [0, 0, 255, 255]); + let capture = Box::new(ScriptedCapture::new(vec![blue])); + let (cmd_tx, cmd_rx) = unbounded::(); + let (ev_tx, ev_rx) = unbounded::(); + let input = Arc::new(RecordingInputSink::new()); + let clock = Arc::new(ManualClock::new(0)); + + let id: ProgramId = "12345678-1234-4567-8910-111213141516".parse().unwrap(); + let mut preds = BTreeMap::new(); + preds.insert( + "red".into(), + PredicateKind::ColorAt { + point: Point { x: 0, y: 0 }, + rgb: Rgb([255, 0, 0]), + tolerance: 0, + }, + ); + let program = Program { + version: DslVersion(DslVersion::V0_1.to_string()), + name: "assert".into(), + id: None, + trigger: Trigger::OnSubmit, + predicates: preds, + watches: vec![], + body: vec![Step::Assert { + predicate: "red".into(), + on_fail: OnFail::Fail, + }], + on_complete: None, + on_fail: None, + timeout_ms: None, + labels: BTreeMap::new(), + priority: Default::default(), + }; + + let sched = Scheduler::new( + SchedulerConfig::default(), + capture, + input, + Perception::new(), + clock, + cmd_rx, + ev_tx, + ); + cmd_tx + .send(SchedulerCommand::SubmitValidated { + program_id: id, + program, + }) + .unwrap(); + let h = std::thread::spawn(move || sched.run_until_shutdown()); + std::thread::sleep(Duration::from_millis(300)); + cmd_tx.send(SchedulerCommand::Shutdown).unwrap(); + h.join().unwrap(); + + let events = drain_events(&ev_rx); + assert_eq!( + events + .iter() + .filter(|e| event_type(e) == "program.failed") + .count(), + 1 + ); +} diff --git a/crates/vcli-runtime/tests/scenarios/daemon_restart_marker.rs b/crates/vcli-runtime/tests/scenarios/daemon_restart_marker.rs new file mode 100644 index 0000000..aef35f4 --- /dev/null +++ b/crates/vcli-runtime/tests/scenarios/daemon_restart_marker.rs @@ -0,0 +1,77 @@ +//! Scenario: `ResumeRunning` starts the body at `from_step` and emits a +//! `program.resumed` event. + +#[path = "../common/mod.rs"] +mod common; + +use std::collections::BTreeMap; +use std::sync::Arc; +use std::time::Duration; + +use common::*; +use vcli_core::step::Step; +use vcli_core::{program::DslVersion, trigger::Trigger, Program}; +use vcli_perception::Perception; +use vcli_runtime::clock::ManualClock; + +#[test] +fn resume_running_starts_body_at_cursor_and_emits_resumed() { + let blank = ScriptedCapture::solid(1, 1, [0, 0, 0, 0]); + let capture = Box::new(ScriptedCapture::new(vec![blank])); + let (cmd_tx, cmd_rx) = unbounded::(); + let (ev_tx, ev_rx) = unbounded::(); + let input = Arc::new(RecordingInputSink::new()); + let clock = Arc::new(ManualClock::new(0)); + + let id: ProgramId = "12345678-1234-4567-8910-111213141516".parse().unwrap(); + let program = Program { + version: DslVersion(DslVersion::V0_1.to_string()), + name: "resumable".into(), + id: None, + trigger: Trigger::OnSubmit, + predicates: BTreeMap::new(), + watches: vec![], + body: vec![ + Step::SleepMs { ms: 1 }, + Step::SleepMs { ms: 1 }, + Step::SleepMs { ms: 1 }, + ], + on_complete: None, + on_fail: None, + timeout_ms: None, + labels: BTreeMap::new(), + priority: Default::default(), + }; + + let sched = Scheduler::new( + SchedulerConfig::default(), + capture, + input, + Perception::new(), + clock, + cmd_rx, + ev_tx, + ); + cmd_tx + .send(SchedulerCommand::ResumeRunning { + program_id: id, + from_step: 2, + program, + }) + .unwrap(); + let h = std::thread::spawn(move || sched.run_until_shutdown()); + std::thread::sleep(Duration::from_millis(300)); + cmd_tx.send(SchedulerCommand::Shutdown).unwrap(); + h.join().unwrap(); + + let events = drain_events(&ev_rx); + let resumed = events.iter().find(|e| event_type(e) == "program.resumed"); + assert!(resumed.is_some(), "must emit program.resumed"); + assert_eq!( + events + .iter() + .filter(|e| event_type(e) == "program.completed") + .count(), + 1 + ); +} diff --git a/crates/vcli-runtime/tests/scenarios/elapsed_ms_since_true.rs b/crates/vcli-runtime/tests/scenarios/elapsed_ms_since_true.rs new file mode 100644 index 0000000..686a4e4 --- /dev/null +++ b/crates/vcli-runtime/tests/scenarios/elapsed_ms_since_true.rs @@ -0,0 +1,98 @@ +//! Scenario: `ElapsedMsSinceTrue` fires only after the referenced predicate +//! has been truthy for the required duration. + +#[path = "../common/mod.rs"] +mod common; + +use std::collections::BTreeMap; +use std::sync::Arc; +use std::time::Duration; + +use common::*; +use vcli_core::action::Button; +use vcli_core::geom::Point; +use vcli_core::predicate::{PredicateKind, Rgb}; +use vcli_core::step::{Step, Target}; +use vcli_core::watch::{Lifetime, Watch, WatchWhen}; +use vcli_core::{program::DslVersion, trigger::Trigger, Program}; +use vcli_perception::Perception; +use vcli_runtime::clock::ManualClock; + +#[test] +fn elapsed_ms_since_true_fires_after_delay() { + let red = ScriptedCapture::solid(1, 1, [255, 0, 0, 255]); + let capture = Box::new(ScriptedCapture::new(vec![red])); + let (cmd_tx, cmd_rx) = unbounded::(); + let (ev_tx, ev_rx) = unbounded::(); + let input = Arc::new(RecordingInputSink::new()); + let clock = Arc::new(ManualClock::new(0)); + + let id: ProgramId = "12345678-1234-4567-8910-111213141516".parse().unwrap(); + let mut preds = BTreeMap::new(); + preds.insert( + "red".into(), + PredicateKind::ColorAt { + point: Point { x: 0, y: 0 }, + rgb: Rgb([255, 0, 0]), + tolerance: 0, + }, + ); + preds.insert( + "delayed_red".into(), + PredicateKind::ElapsedMsSinceTrue { + predicate: "red".into(), + ms: 200, + }, + ); + let program = Program { + version: DslVersion(DslVersion::V0_1.to_string()), + name: "delay".into(), + id: None, + trigger: Trigger::OnSubmit, + predicates: preds, + watches: vec![Watch { + when: WatchWhen::ByName("delayed_red".into()), + steps: vec![Step::Click { + at: Target::Absolute(Point { x: 0, y: 0 }), + button: Button::Left, + }], + throttle_ms: 0, + lifetime: Lifetime::OneShot, + }], + body: vec![], + on_complete: None, + on_fail: None, + timeout_ms: None, + labels: BTreeMap::new(), + priority: Default::default(), + }; + + let sched = Scheduler::new( + SchedulerConfig::default(), + capture, + input.clone(), + Perception::new(), + clock, + cmd_rx, + ev_tx, + ); + cmd_tx + .send(SchedulerCommand::SubmitValidated { + program_id: id, + program, + }) + .unwrap(); + let h = std::thread::spawn(move || sched.run_until_shutdown()); + std::thread::sleep(Duration::from_millis(500)); + cmd_tx.send(SchedulerCommand::Shutdown).unwrap(); + h.join().unwrap(); + + let events = drain_events(&ev_rx); + assert_eq!( + events + .iter() + .filter(|e| event_type(e) == "watch.fired") + .count(), + 1 + ); +} diff --git a/crates/vcli-runtime/tests/scenarios/input_postcondition.rs b/crates/vcli-runtime/tests/scenarios/input_postcondition.rs new file mode 100644 index 0000000..22009b8 --- /dev/null +++ b/crates/vcli-runtime/tests/scenarios/input_postcondition.rs @@ -0,0 +1,101 @@ +//! Scenario: Click → WaitFor(postcondition) succeeds when the postcondition +//! flips truthy. + +#[path = "../common/mod.rs"] +mod common; + +use std::collections::BTreeMap; +use std::sync::Arc; +use std::time::Duration; + +use common::*; +use vcli_core::action::Button; +use vcli_core::geom::Point; +use vcli_core::predicate::{PredicateKind, Rgb}; +use vcli_core::step::{OnTimeout, Step, Target}; +use vcli_core::{program::DslVersion, trigger::Trigger, Program}; +use vcli_perception::Perception; +use vcli_runtime::clock::ManualClock; + +#[test] +fn click_then_wait_for_new_state_succeeds() { + let red = ScriptedCapture::solid(1, 1, [255, 0, 0, 255]); + let green = ScriptedCapture::solid(1, 1, [0, 255, 0, 255]); + let capture = Box::new(ScriptedCapture::new(vec![red, green])); + let (cmd_tx, cmd_rx) = unbounded::(); + let (ev_tx, ev_rx) = unbounded::(); + let input = Arc::new(RecordingInputSink::new()); + let clock = Arc::new(ManualClock::new(0)); + + let id: ProgramId = "12345678-1234-4567-8910-111213141516".parse().unwrap(); + let mut preds = BTreeMap::new(); + preds.insert( + "green".into(), + PredicateKind::ColorAt { + point: Point { x: 0, y: 0 }, + rgb: Rgb([0, 255, 0]), + tolerance: 0, + }, + ); + let program = Program { + version: DslVersion(DslVersion::V0_1.to_string()), + name: "postcond".into(), + id: None, + trigger: Trigger::OnSubmit, + predicates: preds, + watches: vec![], + body: vec![ + Step::Click { + at: Target::Absolute(Point { x: 0, y: 0 }), + button: Button::Left, + }, + Step::WaitFor { + predicate: "green".into(), + timeout_ms: 300, + on_timeout: OnTimeout::Fail, + }, + ], + on_complete: None, + on_fail: None, + timeout_ms: None, + labels: BTreeMap::new(), + priority: Default::default(), + }; + + let sched = Scheduler::new( + SchedulerConfig::default(), + capture, + input.clone(), + Perception::new(), + clock, + cmd_rx, + ev_tx, + ); + cmd_tx + .send(SchedulerCommand::SubmitValidated { + program_id: id, + program, + }) + .unwrap(); + let h = std::thread::spawn(move || sched.run_until_shutdown()); + std::thread::sleep(Duration::from_millis(400)); + cmd_tx.send(SchedulerCommand::Shutdown).unwrap(); + h.join().unwrap(); + + let events = drain_events(&ev_rx); + assert_eq!( + events + .iter() + .filter(|e| event_type(e) == "program.completed") + .count(), + 1 + ); + assert_eq!( + input + .calls() + .iter() + .filter(|c| matches!(c, common::mock_input::Call::Click(_, _))) + .count(), + 1 + ); +} diff --git a/crates/vcli-runtime/tests/scenarios/novelty_timeout.rs b/crates/vcli-runtime/tests/scenarios/novelty_timeout.rs new file mode 100644 index 0000000..34db906 --- /dev/null +++ b/crates/vcli-runtime/tests/scenarios/novelty_timeout.rs @@ -0,0 +1,92 @@ +//! Scenario: Click → WaitFor(postcondition) fails when the postcondition +//! never flips before `timeout_ms`. + +#[path = "../common/mod.rs"] +mod common; + +use std::collections::BTreeMap; +use std::sync::Arc; +use std::time::Duration; + +use common::*; +use vcli_core::action::Button; +use vcli_core::geom::Point; +use vcli_core::predicate::{PredicateKind, Rgb}; +use vcli_core::step::{OnTimeout, Step, Target}; +use vcli_core::{program::DslVersion, trigger::Trigger, Program}; +use vcli_perception::Perception; +use vcli_runtime::clock::ManualClock; + +#[test] +fn wait_for_postcondition_fails_when_never_truthy() { + let red = ScriptedCapture::solid(1, 1, [255, 0, 0, 255]); + let capture = Box::new(ScriptedCapture::new(vec![red.clone(), red.clone(), red])); + let (cmd_tx, cmd_rx) = unbounded::(); + let (ev_tx, ev_rx) = unbounded::(); + let input = Arc::new(RecordingInputSink::new()); + let clock = Arc::new(ManualClock::new(0)); + + let id: ProgramId = "12345678-1234-4567-8910-111213141516".parse().unwrap(); + let mut preds = BTreeMap::new(); + preds.insert( + "green".into(), + PredicateKind::ColorAt { + point: Point { x: 0, y: 0 }, + rgb: Rgb([0, 255, 0]), + tolerance: 0, + }, + ); + let program = Program { + version: DslVersion(DslVersion::V0_1.to_string()), + name: "novelty".into(), + id: None, + trigger: Trigger::OnSubmit, + predicates: preds, + watches: vec![], + body: vec![ + Step::Click { + at: Target::Absolute(Point { x: 0, y: 0 }), + button: Button::Left, + }, + Step::WaitFor { + predicate: "green".into(), + timeout_ms: 200, + on_timeout: OnTimeout::Fail, + }, + ], + on_complete: None, + on_fail: None, + timeout_ms: None, + labels: BTreeMap::new(), + priority: Default::default(), + }; + + let sched = Scheduler::new( + SchedulerConfig::default(), + capture, + input, + Perception::new(), + clock, + cmd_rx, + ev_tx, + ); + cmd_tx + .send(SchedulerCommand::SubmitValidated { + program_id: id, + program, + }) + .unwrap(); + let h = std::thread::spawn(move || sched.run_until_shutdown()); + std::thread::sleep(Duration::from_millis(500)); + cmd_tx.send(SchedulerCommand::Shutdown).unwrap(); + h.join().unwrap(); + + let events = drain_events(&ev_rx); + assert_eq!( + events + .iter() + .filter(|e| event_type(e) == "program.failed") + .count(), + 1 + ); +} diff --git a/crates/vcli-runtime/tests/scenarios/one_shot_watch.rs b/crates/vcli-runtime/tests/scenarios/one_shot_watch.rs new file mode 100644 index 0000000..0a89d29 --- /dev/null +++ b/crates/vcli-runtime/tests/scenarios/one_shot_watch.rs @@ -0,0 +1,108 @@ +//! Scenario: `Lifetime::OneShot` fires once and the program completes. + +#[path = "../common/mod.rs"] +mod common; + +use std::collections::BTreeMap; +use std::sync::Arc; +use std::time::Duration; + +use common::*; +use vcli_core::action::Button; +use vcli_core::geom::Point; +use vcli_core::predicate::{PredicateKind, Rgb}; +use vcli_core::step::{Step, Target}; +use vcli_core::watch::{Lifetime, Watch, WatchWhen}; +use vcli_core::{program::DslVersion, trigger::Trigger, Program}; +use vcli_perception::Perception; +use vcli_runtime::clock::ManualClock; + +#[test] +fn one_shot_watch_fires_once_and_completes() { + let (cmd_tx, cmd_rx) = unbounded::(); + let (ev_tx, ev_rx) = unbounded::(); + + let red_frame = ScriptedCapture::solid(1, 1, [255, 0, 0, 255]); + let capture = Box::new(ScriptedCapture::new(vec![red_frame])); + let input = Arc::new(RecordingInputSink::new()); + let clock = Arc::new(ManualClock::new(1_000)); + + let id: ProgramId = "12345678-1234-4567-8910-111213141516".parse().unwrap(); + let mut preds = BTreeMap::new(); + preds.insert( + "red".into(), + PredicateKind::ColorAt { + point: Point { x: 0, y: 0 }, + rgb: Rgb([255, 0, 0]), + tolerance: 0, + }, + ); + let program = Program { + version: DslVersion(DslVersion::V0_1.to_string()), + name: "one_shot".into(), + id: None, + trigger: Trigger::OnSubmit, + predicates: preds, + watches: vec![Watch { + when: WatchWhen::ByName("red".into()), + steps: vec![Step::Click { + at: Target::Absolute(Point { x: 10, y: 20 }), + button: Button::Left, + }], + throttle_ms: 0, + lifetime: Lifetime::OneShot, + }], + body: vec![], + on_complete: None, + on_fail: None, + timeout_ms: None, + labels: BTreeMap::new(), + priority: Default::default(), + }; + + let sched = Scheduler::new( + SchedulerConfig::default(), + capture, + input.clone(), + Perception::new(), + clock.clone(), + cmd_rx, + ev_tx, + ); + + cmd_tx + .send(SchedulerCommand::SubmitValidated { + program_id: id, + program, + }) + .unwrap(); + + let handle = std::thread::spawn(move || sched.run_until_shutdown()); + std::thread::sleep(Duration::from_millis(300)); + cmd_tx.send(SchedulerCommand::Shutdown).unwrap(); + handle.join().unwrap(); + + let events = drain_events(&ev_rx); + let types: Vec<_> = events.iter().map(event_type).collect(); + + assert!( + types.iter().any(|t| t == "program.submitted"), + "types: {types:?}" + ); + assert!(types.iter().any(|t| t == "watch.fired"), "types: {types:?}"); + assert!( + types.iter().any(|t| t == "program.completed"), + "types: {types:?}" + ); + + let clicks: Vec<_> = input + .calls() + .into_iter() + .filter(|c| matches!(c, common::mock_input::Call::Click(_, _))) + .collect(); + assert_eq!( + clicks.len(), + 1, + "expected exactly one click, got {clicks:?}" + ); +} diff --git a/crates/vcli-runtime/tests/scenarios/persistent_watch.rs b/crates/vcli-runtime/tests/scenarios/persistent_watch.rs new file mode 100644 index 0000000..42bceac --- /dev/null +++ b/crates/vcli-runtime/tests/scenarios/persistent_watch.rs @@ -0,0 +1,100 @@ +//! Scenario: `Lifetime::Persistent` fires on every false→true edge. + +#[path = "../common/mod.rs"] +mod common; + +use std::collections::BTreeMap; +use std::sync::Arc; +use std::time::Duration; + +use common::*; +use vcli_core::action::Button; +use vcli_core::geom::Point; +use vcli_core::predicate::{PredicateKind, Rgb}; +use vcli_core::step::{Step, Target}; +use vcli_core::watch::{Lifetime, Watch, WatchWhen}; +use vcli_core::{program::DslVersion, trigger::Trigger, Program}; +use vcli_perception::Perception; +use vcli_runtime::clock::ManualClock; + +#[test] +fn persistent_watch_fires_on_each_edge() { + let (cmd_tx, cmd_rx) = unbounded::(); + let (ev_tx, ev_rx) = unbounded::(); + + let red = ScriptedCapture::solid(1, 1, [255, 0, 0, 255]); + let blue = ScriptedCapture::solid(1, 1, [0, 0, 255, 255]); + let capture = Box::new(ScriptedCapture::new(vec![ + red.clone(), + blue.clone(), + red.clone(), + blue, + red, + ])); + let input = Arc::new(RecordingInputSink::new()); + let clock = Arc::new(ManualClock::new(0)); + + let id: ProgramId = "12345678-1234-4567-8910-111213141516".parse().unwrap(); + let mut preds = BTreeMap::new(); + preds.insert( + "red".into(), + PredicateKind::ColorAt { + point: Point { x: 0, y: 0 }, + rgb: Rgb([255, 0, 0]), + tolerance: 0, + }, + ); + let program = Program { + version: DslVersion(DslVersion::V0_1.to_string()), + name: "persistent".into(), + id: None, + trigger: Trigger::OnSubmit, + predicates: preds, + watches: vec![Watch { + when: WatchWhen::ByName("red".into()), + steps: vec![Step::Click { + at: Target::Absolute(Point { x: 0, y: 0 }), + button: Button::Left, + }], + throttle_ms: 0, + lifetime: Lifetime::Persistent, + }], + body: vec![], + on_complete: None, + on_fail: None, + timeout_ms: None, + labels: BTreeMap::new(), + priority: Default::default(), + }; + + let sched = Scheduler::new( + SchedulerConfig::default(), + capture, + input.clone(), + Perception::new(), + clock, + cmd_rx, + ev_tx, + ); + cmd_tx + .send(SchedulerCommand::SubmitValidated { + program_id: id, + program, + }) + .unwrap(); + let handle = std::thread::spawn(move || sched.run_until_shutdown()); + std::thread::sleep(Duration::from_millis(500)); + cmd_tx.send(SchedulerCommand::Shutdown).unwrap(); + handle.join().unwrap(); + + let events = drain_events(&ev_rx); + let watch_fires = events + .iter() + .filter(|e| event_type(e) == "watch.fired") + .count(); + assert!( + watch_fires >= 2, + "expected >= 2 watch fires (two rising edges), got {watch_fires}; events={:?}", + events.iter().map(event_type).collect::>() + ); +} diff --git a/crates/vcli-runtime/tests/scenarios/predicate_dedup.rs b/crates/vcli-runtime/tests/scenarios/predicate_dedup.rs new file mode 100644 index 0000000..beb462d --- /dev/null +++ b/crates/vcli-runtime/tests/scenarios/predicate_dedup.rs @@ -0,0 +1,98 @@ +//! Scenario: Two programs sharing a predicate both fire when the predicate +//! goes truthy — the arbiter does not suppress empty-step watches. + +#[path = "../common/mod.rs"] +mod common; + +use std::collections::BTreeMap; +use std::sync::Arc; +use std::time::Duration; + +use common::*; +use vcli_core::geom::Point; +use vcli_core::predicate::{PredicateKind, Rgb}; +use vcli_core::program::Priority; +use vcli_core::watch::{Lifetime, Watch, WatchWhen}; +use vcli_core::{program::DslVersion, trigger::Trigger, Program}; +use vcli_perception::Perception; +use vcli_runtime::clock::ManualClock; + +fn build_program(name: &str, pri: i32) -> Program { + let mut preds = BTreeMap::new(); + preds.insert( + "red".into(), + PredicateKind::ColorAt { + point: Point { x: 0, y: 0 }, + rgb: Rgb([255, 0, 0]), + tolerance: 0, + }, + ); + Program { + version: DslVersion(DslVersion::V0_1.to_string()), + name: name.into(), + id: None, + trigger: Trigger::OnSubmit, + predicates: preds, + watches: vec![Watch { + when: WatchWhen::ByName("red".into()), + steps: vec![], + throttle_ms: 0, + lifetime: Lifetime::OneShot, + }], + body: vec![], + on_complete: None, + on_fail: None, + timeout_ms: None, + labels: BTreeMap::new(), + priority: Priority(pri), + } +} + +#[test] +fn two_programs_same_predicate_both_fire() { + let red = ScriptedCapture::solid(1, 1, [255, 0, 0, 255]); + let capture = Box::new(ScriptedCapture::new(vec![red])); + let (cmd_tx, cmd_rx) = unbounded::(); + let (ev_tx, ev_rx) = unbounded::(); + let input = Arc::new(RecordingInputSink::new()); + let clock = Arc::new(ManualClock::new(0)); + + let id1: ProgramId = "11111111-1111-4567-8910-111213141516".parse().unwrap(); + let id2: ProgramId = "22222222-2222-4567-8910-111213141516".parse().unwrap(); + + let sched = Scheduler::new( + SchedulerConfig::default(), + capture, + input, + Perception::new(), + clock, + cmd_rx, + ev_tx, + ); + cmd_tx + .send(SchedulerCommand::SubmitValidated { + program_id: id1, + program: build_program("p1", 1), + }) + .unwrap(); + cmd_tx + .send(SchedulerCommand::SubmitValidated { + program_id: id2, + program: build_program("p2", 2), + }) + .unwrap(); + let h = std::thread::spawn(move || sched.run_until_shutdown()); + std::thread::sleep(Duration::from_millis(300)); + cmd_tx.send(SchedulerCommand::Shutdown).unwrap(); + h.join().unwrap(); + + let events = drain_events(&ev_rx); + assert_eq!( + events + .iter() + .filter(|e| event_type(e) == "watch.fired") + .count(), + 2, + "both programs must fire" + ); +} diff --git a/crates/vcli-runtime/tests/scenarios/until_predicate.rs b/crates/vcli-runtime/tests/scenarios/until_predicate.rs new file mode 100644 index 0000000..f591f8f --- /dev/null +++ b/crates/vcli-runtime/tests/scenarios/until_predicate.rs @@ -0,0 +1,109 @@ +//! Scenario: `UntilPredicate` retires the watch when its terminator fires. + +#[path = "../common/mod.rs"] +mod common; + +use std::collections::BTreeMap; +use std::sync::Arc; +use std::time::Duration; + +use common::*; +use vcli_core::action::Button; +use vcli_core::geom::Point; +use vcli_core::predicate::{PredicateKind, Rgb}; +use vcli_core::step::{Step, Target}; +use vcli_core::watch::{Lifetime, Watch, WatchWhen}; +use vcli_core::{program::DslVersion, trigger::Trigger, Program}; +use vcli_perception::Perception; +use vcli_runtime::clock::ManualClock; + +#[test] +fn until_predicate_retires_watch() { + let (cmd_tx, cmd_rx) = unbounded::(); + let (ev_tx, ev_rx) = unbounded::(); + + let red = ScriptedCapture::solid(1, 1, [255, 0, 0, 255]); + let green = ScriptedCapture::solid(1, 1, [0, 255, 0, 255]); + let capture = Box::new(ScriptedCapture::new(vec![red, green.clone(), green])); + let input = Arc::new(RecordingInputSink::new()); + let clock = Arc::new(ManualClock::new(0)); + + let id: ProgramId = "12345678-1234-4567-8910-111213141516".parse().unwrap(); + let mut preds = BTreeMap::new(); + preds.insert( + "is_red".into(), + PredicateKind::ColorAt { + point: Point { x: 0, y: 0 }, + rgb: Rgb([255, 0, 0]), + tolerance: 0, + }, + ); + preds.insert( + "done".into(), + PredicateKind::ColorAt { + point: Point { x: 0, y: 0 }, + rgb: Rgb([0, 255, 0]), + tolerance: 0, + }, + ); + + let program = Program { + version: DslVersion(DslVersion::V0_1.to_string()), + name: "until".into(), + id: None, + trigger: Trigger::OnSubmit, + predicates: preds, + watches: vec![Watch { + when: WatchWhen::ByName("is_red".into()), + steps: vec![Step::Click { + at: Target::Absolute(Point { x: 0, y: 0 }), + button: Button::Left, + }], + throttle_ms: 0, + lifetime: Lifetime::UntilPredicate { + name: "done".into(), + }, + }], + body: vec![], + on_complete: None, + on_fail: None, + timeout_ms: None, + labels: BTreeMap::new(), + priority: Default::default(), + }; + + let sched = Scheduler::new( + SchedulerConfig::default(), + capture, + input.clone(), + Perception::new(), + clock, + cmd_rx, + ev_tx, + ); + cmd_tx + .send(SchedulerCommand::SubmitValidated { + program_id: id, + program, + }) + .unwrap(); + let h = std::thread::spawn(move || sched.run_until_shutdown()); + std::thread::sleep(Duration::from_millis(400)); + cmd_tx.send(SchedulerCommand::Shutdown).unwrap(); + h.join().unwrap(); + + let events = drain_events(&ev_rx); + let fires = events + .iter() + .filter(|e| event_type(e) == "watch.fired") + .count(); + let completes = events + .iter() + .filter(|e| event_type(e) == "program.completed") + .count(); + assert!( + (1..=2).contains(&fires), + "expected 1..=2 fires, got {fires}" + ); + assert_eq!(completes, 1, "expected exactly one completion"); +} diff --git a/crates/vcli-runtime/tests/scenarios/wait_for_timeout.rs b/crates/vcli-runtime/tests/scenarios/wait_for_timeout.rs new file mode 100644 index 0000000..0b6dc7a --- /dev/null +++ b/crates/vcli-runtime/tests/scenarios/wait_for_timeout.rs @@ -0,0 +1,83 @@ +//! Scenario: body `wait_for` fails when predicate never becomes truthy +//! and `on_timeout: Fail`. + +#[path = "../common/mod.rs"] +mod common; + +use std::collections::BTreeMap; +use std::sync::Arc; +use std::time::Duration; + +use common::*; +use vcli_core::geom::Point; +use vcli_core::predicate::{PredicateKind, Rgb}; +use vcli_core::step::{OnTimeout, Step}; +use vcli_core::{program::DslVersion, trigger::Trigger, Program}; +use vcli_perception::Perception; +use vcli_runtime::clock::ManualClock; + +#[test] +fn wait_for_fails_on_timeout_when_on_timeout_fail() { + let blue = ScriptedCapture::solid(1, 1, [0, 0, 255, 255]); + let capture = Box::new(ScriptedCapture::new(vec![blue])); + let (cmd_tx, cmd_rx) = unbounded::(); + let (ev_tx, ev_rx) = unbounded::(); + let input = Arc::new(RecordingInputSink::new()); + let clock = Arc::new(ManualClock::new(0)); + + let id: ProgramId = "12345678-1234-4567-8910-111213141516".parse().unwrap(); + let mut preds = BTreeMap::new(); + preds.insert( + "red".into(), + PredicateKind::ColorAt { + point: Point { x: 0, y: 0 }, + rgb: Rgb([255, 0, 0]), + tolerance: 0, + }, + ); + let program = Program { + version: DslVersion(DslVersion::V0_1.to_string()), + name: "wait".into(), + id: None, + trigger: Trigger::OnSubmit, + predicates: preds, + watches: vec![], + body: vec![Step::WaitFor { + predicate: "red".into(), + timeout_ms: 150, + on_timeout: OnTimeout::Fail, + }], + on_complete: None, + on_fail: None, + timeout_ms: None, + labels: BTreeMap::new(), + priority: Default::default(), + }; + + let sched = Scheduler::new( + SchedulerConfig::default(), + capture, + input, + Perception::new(), + clock, + cmd_rx, + ev_tx, + ); + cmd_tx + .send(SchedulerCommand::SubmitValidated { + program_id: id, + program, + }) + .unwrap(); + let h = std::thread::spawn(move || sched.run_until_shutdown()); + std::thread::sleep(Duration::from_millis(500)); + cmd_tx.send(SchedulerCommand::Shutdown).unwrap(); + h.join().unwrap(); + + let events = drain_events(&ev_rx); + let failed = events + .iter() + .filter(|e| event_type(e) == "program.failed") + .count(); + assert_eq!(failed, 1, "expected exactly one program.failed"); +} diff --git a/crates/vcli-runtime/tests/scenarios/watch_timeout.rs b/crates/vcli-runtime/tests/scenarios/watch_timeout.rs new file mode 100644 index 0000000..6fa79ae --- /dev/null +++ b/crates/vcli-runtime/tests/scenarios/watch_timeout.rs @@ -0,0 +1,93 @@ +//! Scenario: `Lifetime::TimeoutMs` retires the watch once the window elapses +//! even without a rising edge. + +#[path = "../common/mod.rs"] +mod common; + +use std::collections::BTreeMap; +use std::sync::Arc; +use std::time::Duration; + +use common::*; +use vcli_core::geom::Point; +use vcli_core::predicate::{PredicateKind, Rgb}; +use vcli_core::watch::{Lifetime, Watch, WatchWhen}; +use vcli_core::{program::DslVersion, trigger::Trigger, Program}; +use vcli_perception::Perception; +use vcli_runtime::clock::ManualClock; + +#[test] +fn watch_timeout_retires_after_budget() { + let (cmd_tx, cmd_rx) = unbounded::(); + let (ev_tx, ev_rx) = unbounded::(); + let blue = ScriptedCapture::solid(1, 1, [0, 0, 255, 255]); + let capture = Box::new(ScriptedCapture::new(vec![blue])); + let input = Arc::new(RecordingInputSink::new()); + let clock = Arc::new(ManualClock::new(0)); + + let id: ProgramId = "12345678-1234-4567-8910-111213141516".parse().unwrap(); + let mut preds = BTreeMap::new(); + preds.insert( + "red".into(), + PredicateKind::ColorAt { + point: Point { x: 0, y: 0 }, + rgb: Rgb([255, 0, 0]), + tolerance: 0, + }, + ); + let program = Program { + version: DslVersion(DslVersion::V0_1.to_string()), + name: "timeout".into(), + id: None, + trigger: Trigger::OnSubmit, + predicates: preds, + watches: vec![Watch { + when: WatchWhen::ByName("red".into()), + steps: vec![], + throttle_ms: 0, + lifetime: Lifetime::TimeoutMs { ms: 200 }, + }], + body: vec![], + on_complete: None, + on_fail: None, + timeout_ms: None, + labels: BTreeMap::new(), + priority: Default::default(), + }; + + let sched = Scheduler::new( + SchedulerConfig::default(), + capture, + input, + Perception::new(), + clock, + cmd_rx, + ev_tx, + ); + cmd_tx + .send(SchedulerCommand::SubmitValidated { + program_id: id, + program, + }) + .unwrap(); + let h = std::thread::spawn(move || sched.run_until_shutdown()); + std::thread::sleep(Duration::from_millis(500)); + cmd_tx.send(SchedulerCommand::Shutdown).unwrap(); + h.join().unwrap(); + + let events = drain_events(&ev_rx); + assert_eq!( + events + .iter() + .filter(|e| event_type(e) == "watch.fired") + .count(), + 0 + ); + assert_eq!( + events + .iter() + .filter(|e| event_type(e) == "program.completed") + .count(), + 1 + ); +} diff --git a/crates/vcli-runtime/tests/scenarios/while_true_retry.rs b/crates/vcli-runtime/tests/scenarios/while_true_retry.rs new file mode 100644 index 0000000..cedc582 --- /dev/null +++ b/crates/vcli-runtime/tests/scenarios/while_true_retry.rs @@ -0,0 +1,102 @@ +//! Scenario: `throttle_ms` caps a persistent watch at one fire per window. + +#[path = "../common/mod.rs"] +mod common; + +use std::collections::BTreeMap; +use std::sync::Arc; +use std::time::Duration; + +use common::*; +use vcli_core::action::Button; +use vcli_core::geom::Point; +use vcli_core::predicate::{PredicateKind, Rgb}; +use vcli_core::step::{Step, Target}; +use vcli_core::watch::{Lifetime, Watch, WatchWhen}; +use vcli_core::{program::DslVersion, trigger::Trigger, Program}; +use vcli_perception::Perception; +use vcli_runtime::clock::ManualClock; + +#[test] +fn throttled_persistent_fires_at_most_once_per_window() { + let red = ScriptedCapture::solid(1, 1, [255, 0, 0, 255]); + let blue = ScriptedCapture::solid(1, 1, [0, 0, 255, 255]); + let capture = Box::new(ScriptedCapture::new(vec![ + red.clone(), + blue.clone(), + red.clone(), + blue, + red, + ])); + let (cmd_tx, cmd_rx) = unbounded::(); + let (ev_tx, ev_rx) = unbounded::(); + let input = Arc::new(RecordingInputSink::new()); + let clock = Arc::new(ManualClock::new(0)); + + let id: ProgramId = "12345678-1234-4567-8910-111213141516".parse().unwrap(); + let mut preds = BTreeMap::new(); + preds.insert( + "red".into(), + PredicateKind::ColorAt { + point: Point { x: 0, y: 0 }, + rgb: Rgb([255, 0, 0]), + tolerance: 0, + }, + ); + let program = Program { + version: DslVersion(DslVersion::V0_1.to_string()), + name: "throttle".into(), + id: None, + trigger: Trigger::OnSubmit, + predicates: preds, + watches: vec![Watch { + when: WatchWhen::ByName("red".into()), + steps: vec![Step::Click { + at: Target::Absolute(Point { x: 0, y: 0 }), + button: Button::Left, + }], + throttle_ms: 10_000, + lifetime: Lifetime::Persistent, + }], + body: vec![], + on_complete: None, + on_fail: None, + timeout_ms: None, + labels: BTreeMap::new(), + priority: Default::default(), + }; + + let sched = Scheduler::new( + SchedulerConfig::default(), + capture, + input.clone(), + Perception::new(), + clock, + cmd_rx, + ev_tx, + ); + cmd_tx + .send(SchedulerCommand::SubmitValidated { + program_id: id, + program, + }) + .unwrap(); + let h = std::thread::spawn(move || sched.run_until_shutdown()); + std::thread::sleep(Duration::from_millis(500)); + cmd_tx.send(SchedulerCommand::Shutdown).unwrap(); + h.join().unwrap(); + + // Drop unused event stream. + let _ = ev_rx; + + let clicks: Vec<_> = input + .calls() + .into_iter() + .filter(|c| matches!(c, common::mock_input::Call::Click(_, _))) + .collect(); + assert_eq!( + clicks.len(), + 1, + "throttle must cap fires at one per window; got {clicks:?}" + ); +}