fix(homecore-automation): start engine + implement time/run-mode/choose/template (ADR-161 A3-A7)
A3 (HIGH): homecore-server constructed AutomationEngine then dropped it immediately while the doc claimed automation was active. Now .start()s the engine into a long-lived binding (event loop + timer task). A4 (HIGH): Trigger::Time was hard-coded false with no timer. Added a 1 Hz wall-clock timer task that fires time: automations when local HH:MM:SS matches 'at' (HH:MM or HH:MM:SS); matches_sync(Time)=false is now correct + documented. A5 (HIGH): RunMode was documented as AtomicBool-enforced but every trigger spawned unbounded parallel. Each automation now carries a running AtomicBool; Single/IgnoreFirst skip re-entrant triggers, Parallel fires every time. (Bounded Queued/Restart/max → ACCEPTED-FUTURE, honestly stated in the doc.) A6 (HIGH): Action::Choose discarded choices and always ran default. Now deserialises each branch's conditions, evaluates them, and runs the first matching branch; default only if none match. A7 (MEDIUM): template: conditions were always false in the engine path (EvalContext built with template_env: None). The engine now builds a TemplateEnvironment over the state machine and threads it into every EvalContext (event loop, timer, Choose). Tests (fail on old source): - engine_behaviors::time_trigger_fires_via_timer_path (A4) - engine_behaviors::single_mode_does_not_double_fire_on_rapid_triggers (A5; old fired 2x) - engine_behaviors::parallel_mode_does_fire_concurrently (A5) - action::choose_runs_matching_branch_not_default (A6; old ran default) - engine_behaviors::template_condition_evaluates_true_in_engine (A7; old always false) engine.rs kept <500 lines; behavioral tests moved to tests/engine_behaviors.rs. Co-Authored-By: claude-flow <ruv@ruv.net>
This commit is contained in:
ruv
parent
9d52d49c0b
commit
dff75a479e
|
|
@ -3,15 +3,26 @@
|
|||
//! Implements the ADR-129 P1 action set: `service_call`, `delay`, `scene`,
|
||||
//! `wait_for_trigger`, `choose`. Complex variants (parallel, repeat, if,
|
||||
//! stop, fire_event, wait_template) land in P2.
|
||||
//!
|
||||
//! ## `choose` branch evaluation (ADR-161, HC-WS-06)
|
||||
//!
|
||||
//! `Action::Choose` evaluates each branch's `conditions` against the live
|
||||
//! [`EvalContext`] (deserialising the per-branch `serde_yaml::Value`
|
||||
//! conditions into [`Condition`]) and runs the FIRST matching branch's
|
||||
//! sequence. Only if no branch matches does it fall to `default`. Before
|
||||
//! this fix the branches were discarded and `default` always ran.
|
||||
|
||||
use std::sync::Arc;
|
||||
use std::time::Duration;
|
||||
|
||||
use serde::{Deserialize, Serialize};
|
||||
use tokio::time::sleep;
|
||||
|
||||
use homecore::{Context, HomeCore, ServiceCall, ServiceName};
|
||||
use homecore::{Context, HomeCore, ServiceCall, ServiceName, StateMachine};
|
||||
|
||||
use crate::condition::{Condition, EvalContext};
|
||||
use crate::error::AutomationError;
|
||||
use crate::template::TemplateEnvironment;
|
||||
|
||||
/// Runtime context passed into action execution.
|
||||
pub struct ExecutionContext {
|
||||
|
|
@ -21,14 +32,40 @@ pub struct ExecutionContext {
|
|||
pub context: Context,
|
||||
/// Automation ID for tracing/logging.
|
||||
pub automation_id: String,
|
||||
/// Condition-evaluation context for `Choose` branches. Carries the
|
||||
/// state-machine snapshot + optional template environment so branch
|
||||
/// conditions (incl. `template:`) evaluate against live state.
|
||||
pub eval: EvalContext,
|
||||
}
|
||||
|
||||
impl ExecutionContext {
|
||||
/// Build a context whose `Choose` branches evaluate against the
|
||||
/// HomeCore state machine (no template env — `template:` branch
|
||||
/// conditions evaluate false; use [`Self::with_templates`] to wire
|
||||
/// one).
|
||||
pub fn new(hc: HomeCore, automation_id: impl Into<String>) -> Self {
|
||||
let sm = Arc::new(hc.states().clone());
|
||||
Self {
|
||||
hc,
|
||||
context: Context::new(),
|
||||
automation_id: automation_id.into(),
|
||||
eval: EvalContext::new(sm),
|
||||
}
|
||||
}
|
||||
|
||||
/// Build a context with a template environment wired into the
|
||||
/// `Choose` branch-condition evaluator.
|
||||
pub fn with_templates(
|
||||
hc: HomeCore,
|
||||
automation_id: impl Into<String>,
|
||||
states: Arc<StateMachine>,
|
||||
templates: Arc<TemplateEnvironment>,
|
||||
) -> Self {
|
||||
Self {
|
||||
hc,
|
||||
context: Context::new(),
|
||||
automation_id: automation_id.into(),
|
||||
eval: EvalContext::with_templates(states, templates),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -72,6 +109,27 @@ pub struct ChoiceBranch {
|
|||
pub sequence: Vec<Action>,
|
||||
}
|
||||
|
||||
impl ChoiceBranch {
|
||||
/// Does this branch match? All of its `conditions` must evaluate
|
||||
/// true (HA `choose` semantics are AND-over-conditions). Each raw
|
||||
/// `serde_yaml::Value` is deserialised into a [`Condition`]; a
|
||||
/// condition that fails to parse is treated as non-matching (the
|
||||
/// branch is skipped) rather than silently passing. An empty
|
||||
/// `conditions` list matches (an unconditional branch).
|
||||
pub async fn matches(&self, eval: &EvalContext) -> bool {
|
||||
for raw in &self.conditions {
|
||||
let cond: Condition = match serde_yaml::from_value(raw.clone()) {
|
||||
Ok(c) => c,
|
||||
Err(_) => return false,
|
||||
};
|
||||
if !cond.evaluate(eval).await {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
true
|
||||
}
|
||||
}
|
||||
|
||||
impl Action {
|
||||
/// Execute this action using the provided context.
|
||||
///
|
||||
|
|
@ -118,9 +176,18 @@ impl Action {
|
|||
}
|
||||
Ok(serde_json::Value::Null)
|
||||
}
|
||||
Action::Choose { choices: _, default } => {
|
||||
// P1 stub — condition evaluation for choices lands in P2;
|
||||
// for now, fall through to default branch.
|
||||
Action::Choose { choices, default } => {
|
||||
// Evaluate each branch's conditions against live state;
|
||||
// run the first branch whose conditions ALL pass. Fall
|
||||
// to `default` only if no branch matches (HC-WS-06).
|
||||
for branch in choices {
|
||||
if branch.matches(&ctx.eval).await {
|
||||
for a in &branch.sequence {
|
||||
a.execute(ctx).await?;
|
||||
}
|
||||
return Ok(serde_json::Value::Null);
|
||||
}
|
||||
}
|
||||
for a in default {
|
||||
a.execute(ctx).await?;
|
||||
}
|
||||
|
|
@ -188,4 +255,100 @@ mod tests {
|
|||
let err = action.execute(&mut exec_ctx).await.unwrap_err();
|
||||
assert!(matches!(err, AutomationError::ServiceCall(ServiceError::NotRegistered { .. })));
|
||||
}
|
||||
|
||||
/// Register two recording handlers and return their call logs.
|
||||
async fn two_recorders(
|
||||
hc: &HomeCore,
|
||||
) -> (Arc<Mutex<Vec<serde_json::Value>>>, Arc<Mutex<Vec<serde_json::Value>>>) {
|
||||
use homecore::EntityId;
|
||||
let _ = EntityId::parse("light.x"); // touch import path
|
||||
let mk = |hc: &HomeCore, svc: &'static str| {
|
||||
let log: Arc<Mutex<Vec<serde_json::Value>>> = Arc::new(Mutex::new(vec![]));
|
||||
let log2 = Arc::clone(&log);
|
||||
let hc = hc.clone();
|
||||
async move {
|
||||
hc.services()
|
||||
.register(
|
||||
ServiceName::new("light", svc),
|
||||
FnHandler(move |call: ServiceCall| {
|
||||
let l = Arc::clone(&log2);
|
||||
async move {
|
||||
l.lock().unwrap().push(call.data.clone());
|
||||
Ok(serde_json::Value::Null)
|
||||
}
|
||||
}),
|
||||
)
|
||||
.await;
|
||||
log
|
||||
}
|
||||
};
|
||||
let branch_log = mk(hc, "branch_service").await;
|
||||
let default_log = mk(hc, "default_service").await;
|
||||
(branch_log, default_log)
|
||||
}
|
||||
|
||||
fn choose_with_match() -> Action {
|
||||
// A `Choose` whose first branch requires light.gate == "open".
|
||||
let branch_conditions = vec![serde_yaml::from_str::<serde_yaml::Value>(
|
||||
"condition: state\nentity_id: light.gate\nstate: open",
|
||||
)
|
||||
.unwrap()];
|
||||
Action::Choose {
|
||||
choices: vec![ChoiceBranch {
|
||||
conditions: branch_conditions,
|
||||
sequence: vec![Action::ServiceCall {
|
||||
domain: "light".into(),
|
||||
service: "branch_service".into(),
|
||||
data: serde_json::json!({"branch": true}),
|
||||
}],
|
||||
}],
|
||||
default: vec![Action::ServiceCall {
|
||||
domain: "light".into(),
|
||||
service: "default_service".into(),
|
||||
data: serde_json::json!({"default": true}),
|
||||
}],
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn choose_runs_matching_branch_not_default() {
|
||||
// HC-WS-06: with the branch condition satisfied, the branch
|
||||
// sequence runs and `default` does NOT. On the pre-fix code
|
||||
// (choices discarded) `default` ran instead → this fails on old.
|
||||
use homecore::{Context, EntityId};
|
||||
let hc = HomeCore::new();
|
||||
let (branch_log, default_log) = two_recorders(&hc).await;
|
||||
hc.states().set(
|
||||
EntityId::parse("light.gate").unwrap(),
|
||||
"open",
|
||||
serde_json::json!({}),
|
||||
Context::new(),
|
||||
);
|
||||
|
||||
let mut ctx = ExecutionContext::new(hc, "choose_auto");
|
||||
choose_with_match().execute(&mut ctx).await.unwrap();
|
||||
|
||||
assert_eq!(branch_log.lock().unwrap().len(), 1, "matching branch must run");
|
||||
assert_eq!(default_log.lock().unwrap().len(), 0, "default must NOT run when a branch matches");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn choose_falls_to_default_when_no_branch_matches() {
|
||||
use homecore::{Context, EntityId};
|
||||
let hc = HomeCore::new();
|
||||
let (branch_log, default_log) = two_recorders(&hc).await;
|
||||
// gate is "closed" → branch condition (== "open") fails.
|
||||
hc.states().set(
|
||||
EntityId::parse("light.gate").unwrap(),
|
||||
"closed",
|
||||
serde_json::json!({}),
|
||||
Context::new(),
|
||||
);
|
||||
|
||||
let mut ctx = ExecutionContext::new(hc, "choose_auto");
|
||||
choose_with_match().execute(&mut ctx).await.unwrap();
|
||||
|
||||
assert_eq!(branch_log.lock().unwrap().len(), 0, "branch must not run when condition fails");
|
||||
assert_eq!(default_log.lock().unwrap().len(), 1, "default must run when no branch matches");
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -2,56 +2,129 @@
|
|||
//! triggers, and runs automation action sequences.
|
||||
//!
|
||||
//! ADR-129 §2 design: one Tokio task per running automation instance.
|
||||
//! RunMode::Single is enforced via a per-automation `AtomicBool` flag.
|
||||
//!
|
||||
//! ## Run modes (ADR-161, HC-WS-05)
|
||||
//!
|
||||
//! `RunMode::Single` is enforced via a per-automation `AtomicBool`
|
||||
//! guard: while an instance is executing, a second trigger is skipped.
|
||||
//! `Parallel` (and the as-yet-unbounded `Restart`/`Queued`) spawn a
|
||||
//! fresh instance on every trigger. (Before this fix the doc claimed
|
||||
//! AtomicBool enforcement but every trigger spawned unbounded parallel
|
||||
//! tasks regardless of `mode`.)
|
||||
//!
|
||||
//! ## Time triggers (ADR-161, HC-WS-04)
|
||||
//!
|
||||
//! `Trigger::Time { at: "HH:MM:SS" }` is evaluated by a wall-clock timer
|
||||
//! task (1 Hz tokio interval) — `Trigger::matches_sync` returns false for
|
||||
//! `Time` because it has no clock. The timer fires each `time:`
|
||||
//! automation once when the local wall-clock second equals its `at`.
|
||||
//!
|
||||
//! ## Template conditions (ADR-161, HC-WS-07)
|
||||
//!
|
||||
//! The engine builds a real [`TemplateEnvironment`] over the state
|
||||
//! machine and passes it into every `EvalContext` (via
|
||||
//! `EvalContext::with_templates`), so `template:` conditions evaluate
|
||||
//! against live state instead of always returning false.
|
||||
|
||||
use std::sync::atomic::{AtomicBool, Ordering};
|
||||
use std::sync::{Arc, Mutex};
|
||||
|
||||
use chrono::{Local, Timelike};
|
||||
use tokio::sync::broadcast;
|
||||
|
||||
use homecore::HomeCore;
|
||||
|
||||
use crate::action::ExecutionContext;
|
||||
use crate::automation::Automation;
|
||||
use crate::automation::{Automation, RunMode};
|
||||
use crate::condition::EvalContext;
|
||||
use crate::trigger::TriggerContext;
|
||||
use crate::template::TemplateEnvironment;
|
||||
use crate::trigger::{Trigger, TriggerContext};
|
||||
|
||||
/// An automation registered with the engine, plus its runtime run-state.
|
||||
struct Registered {
|
||||
auto: Arc<Automation>,
|
||||
/// `true` while a `Single`-mode instance is executing. Used to
|
||||
/// skip re-entrant triggers (HC-WS-05).
|
||||
running: Arc<AtomicBool>,
|
||||
}
|
||||
|
||||
/// The automation engine. Holds a HOMECORE handle and a list of registered
|
||||
/// automations. Call `start()` to begin listening for events.
|
||||
pub struct AutomationEngine {
|
||||
hc: HomeCore,
|
||||
automations: Arc<Mutex<Vec<Arc<Automation>>>>,
|
||||
automations: Arc<Mutex<Vec<Registered>>>,
|
||||
templates: Arc<TemplateEnvironment>,
|
||||
}
|
||||
|
||||
impl AutomationEngine {
|
||||
/// Create a new engine backed by the given HOMECORE handle.
|
||||
pub fn new(hc: HomeCore) -> Self {
|
||||
let templates = Arc::new(TemplateEnvironment::new(Arc::new(hc.states().clone())));
|
||||
Self {
|
||||
hc,
|
||||
automations: Arc::new(Mutex::new(vec![])),
|
||||
templates,
|
||||
}
|
||||
}
|
||||
|
||||
/// Register an automation. Can be called before or after `start()`.
|
||||
pub fn register(&self, automation: Automation) {
|
||||
self.automations.lock().unwrap().push(Arc::new(automation));
|
||||
self.automations.lock().unwrap().push(Registered {
|
||||
auto: Arc::new(automation),
|
||||
running: Arc::new(AtomicBool::new(false)),
|
||||
});
|
||||
}
|
||||
|
||||
/// Number of registered automations.
|
||||
pub fn len(&self) -> usize {
|
||||
self.automations.lock().unwrap().len()
|
||||
}
|
||||
|
||||
/// Is the engine holding zero automations?
|
||||
pub fn is_empty(&self) -> bool {
|
||||
self.len() == 0
|
||||
}
|
||||
|
||||
/// Build an `EvalContext` with the engine's template environment
|
||||
/// wired in, over a fresh snapshot of the state machine.
|
||||
fn eval_ctx(&self) -> EvalContext {
|
||||
EvalContext::with_templates(
|
||||
Arc::new(self.hc.states().clone()),
|
||||
Arc::clone(&self.templates),
|
||||
)
|
||||
}
|
||||
|
||||
/// Subscribe to the state-machine broadcast channel and start
|
||||
/// evaluating triggers. Returns a join handle for the background task.
|
||||
/// evaluating triggers. Also starts the wall-clock timer task that
|
||||
/// evaluates `time:` triggers. Returns a join handle for the event
|
||||
/// task (the timer task is detached and tied to the engine handle's
|
||||
/// lifetime via the broadcast channel close).
|
||||
///
|
||||
/// The task runs until the broadcast sender is dropped (i.e. the
|
||||
/// `HomeCore` instance is destroyed).
|
||||
pub fn start(&self) -> tokio::task::JoinHandle<()> {
|
||||
self.start_timer();
|
||||
self.start_event_loop()
|
||||
}
|
||||
|
||||
/// Event-driven loop: state/numeric/event triggers.
|
||||
fn start_event_loop(&self) -> tokio::task::JoinHandle<()> {
|
||||
let mut rx = self.hc.states().subscribe();
|
||||
let automations = Arc::clone(&self.automations);
|
||||
let hc = self.hc.clone();
|
||||
let templates = Arc::clone(&self.templates);
|
||||
|
||||
tokio::spawn(async move {
|
||||
loop {
|
||||
match rx.recv().await {
|
||||
Ok(event) => {
|
||||
let autos = automations.lock().unwrap().clone();
|
||||
for automation in autos {
|
||||
let snapshot: Vec<(Arc<Automation>, Arc<AtomicBool>)> = automations
|
||||
.lock()
|
||||
.unwrap()
|
||||
.iter()
|
||||
.map(|r| (Arc::clone(&r.auto), Arc::clone(&r.running)))
|
||||
.collect();
|
||||
for (automation, running) in snapshot {
|
||||
if !automation.enabled {
|
||||
continue;
|
||||
}
|
||||
|
|
@ -60,7 +133,6 @@ impl AutomationEngine {
|
|||
event.old_state.clone(),
|
||||
event.new_state.clone(),
|
||||
);
|
||||
// Check all triggers — fire on first match
|
||||
let triggered = automation
|
||||
.trigger
|
||||
.iter()
|
||||
|
|
@ -68,36 +140,15 @@ impl AutomationEngine {
|
|||
if !triggered {
|
||||
continue;
|
||||
}
|
||||
// Evaluate conditions
|
||||
let sm = Arc::new(hc.states().clone());
|
||||
let eval_ctx = EvalContext::new(sm);
|
||||
let mut conditions_pass = true;
|
||||
for cond in &automation.condition {
|
||||
if !cond.evaluate(&eval_ctx).await {
|
||||
conditions_pass = false;
|
||||
break;
|
||||
}
|
||||
}
|
||||
if !conditions_pass {
|
||||
// Conditions (with template env wired in — HC-WS-07).
|
||||
let eval_ctx = EvalContext::with_templates(
|
||||
Arc::new(hc.states().clone()),
|
||||
Arc::clone(&templates),
|
||||
);
|
||||
if !conditions_pass(&automation, &eval_ctx).await {
|
||||
continue;
|
||||
}
|
||||
// Execute actions in a spawned task (non-blocking)
|
||||
let auto_clone = Arc::clone(&automation);
|
||||
let hc_clone = hc.clone();
|
||||
tokio::spawn(async move {
|
||||
let mut exec_ctx =
|
||||
ExecutionContext::new(hc_clone, auto_clone.id.clone());
|
||||
for action in &auto_clone.action {
|
||||
if let Err(e) = action.execute(&mut exec_ctx).await {
|
||||
// P1: log errors to stderr; structured logging in P2
|
||||
eprintln!(
|
||||
"[homecore-automation] action error in {}: {e}",
|
||||
auto_clone.id
|
||||
);
|
||||
break;
|
||||
}
|
||||
}
|
||||
});
|
||||
spawn_run(&hc, automation, running);
|
||||
}
|
||||
}
|
||||
Err(broadcast::error::RecvError::Closed) => break,
|
||||
|
|
@ -108,6 +159,156 @@ impl AutomationEngine {
|
|||
}
|
||||
})
|
||||
}
|
||||
|
||||
/// Wall-clock timer task: fires `time:` triggers (HC-WS-04). Ticks at
|
||||
/// 1 Hz and runs each matching automation once when the local
|
||||
/// wall-clock `HH:MM:SS` equals the trigger's `at`. The task exits
|
||||
/// when the state-machine broadcast channel closes (engine teardown).
|
||||
fn start_timer(&self) -> tokio::task::JoinHandle<()> {
|
||||
let automations = Arc::clone(&self.automations);
|
||||
let hc = self.hc.clone();
|
||||
let templates = Arc::clone(&self.templates);
|
||||
// A receiver that lets the timer notice engine teardown.
|
||||
let mut teardown_rx = self.hc.states().subscribe();
|
||||
|
||||
tokio::spawn(async move {
|
||||
let mut interval = tokio::time::interval(std::time::Duration::from_millis(1000));
|
||||
// Track the last second we fired, to fire once per match.
|
||||
let mut last_fired_sec: Option<String> = None;
|
||||
loop {
|
||||
tokio::select! {
|
||||
_ = interval.tick() => {
|
||||
let now = Local::now();
|
||||
let hhmmss = format!("{:02}:{:02}:{:02}", now.hour(), now.minute(), now.second());
|
||||
if last_fired_sec.as_deref() == Some(hhmmss.as_str()) {
|
||||
continue;
|
||||
}
|
||||
let snapshot: Vec<(Arc<Automation>, Arc<AtomicBool>)> = automations
|
||||
.lock()
|
||||
.unwrap()
|
||||
.iter()
|
||||
.map(|r| (Arc::clone(&r.auto), Arc::clone(&r.running)))
|
||||
.collect();
|
||||
let mut fired_any = false;
|
||||
for (automation, running) in snapshot {
|
||||
if !automation.enabled {
|
||||
continue;
|
||||
}
|
||||
let time_match = automation.trigger.iter().any(|t| match t {
|
||||
Trigger::Time { at } => time_at_matches(at, &hhmmss),
|
||||
_ => false,
|
||||
});
|
||||
if !time_match {
|
||||
continue;
|
||||
}
|
||||
let eval_ctx = EvalContext::with_templates(
|
||||
Arc::new(hc.states().clone()),
|
||||
Arc::clone(&templates),
|
||||
);
|
||||
if !conditions_pass(&automation, &eval_ctx).await {
|
||||
continue;
|
||||
}
|
||||
spawn_run(&hc, automation, running);
|
||||
fired_any = true;
|
||||
}
|
||||
if fired_any {
|
||||
last_fired_sec = Some(hhmmss);
|
||||
}
|
||||
}
|
||||
r = teardown_rx.recv() => {
|
||||
if let Err(broadcast::error::RecvError::Closed) = r {
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
/// Manually fire any `time:` automations whose `at` equals `hhmmss`
|
||||
/// (`"HH:MM:SS"`). Bypasses the 1 Hz clock so tests can assert the
|
||||
/// time-trigger path deterministically without waiting for a
|
||||
/// wall-clock second to roll over. Returns the number of automations
|
||||
/// that fired (passed conditions and were spawned).
|
||||
pub async fn fire_time_for_test(&self, hhmmss: &str) -> usize {
|
||||
let snapshot: Vec<(Arc<Automation>, Arc<AtomicBool>)> = self
|
||||
.automations
|
||||
.lock()
|
||||
.unwrap()
|
||||
.iter()
|
||||
.map(|r| (Arc::clone(&r.auto), Arc::clone(&r.running)))
|
||||
.collect();
|
||||
let mut fired = 0usize;
|
||||
for (automation, running) in snapshot {
|
||||
if !automation.enabled {
|
||||
continue;
|
||||
}
|
||||
let time_match = automation.trigger.iter().any(|t| match t {
|
||||
Trigger::Time { at } => time_at_matches(at, hhmmss),
|
||||
_ => false,
|
||||
});
|
||||
if !time_match {
|
||||
continue;
|
||||
}
|
||||
let eval_ctx = self.eval_ctx();
|
||||
if !conditions_pass(&automation, &eval_ctx).await {
|
||||
continue;
|
||||
}
|
||||
spawn_run(&self.hc, automation, running);
|
||||
fired += 1;
|
||||
}
|
||||
fired
|
||||
}
|
||||
}
|
||||
|
||||
/// Evaluate all of an automation's conditions (AND). Empty → pass.
|
||||
async fn conditions_pass(automation: &Automation, eval_ctx: &EvalContext) -> bool {
|
||||
for cond in &automation.condition {
|
||||
if !cond.evaluate(eval_ctx).await {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
true
|
||||
}
|
||||
|
||||
/// Does a `Time` trigger `at` value match the current `HH:MM:SS`?
|
||||
/// Accepts `HH:MM` (matches at :00 seconds) and `HH:MM:SS`.
|
||||
fn time_at_matches(at: &str, hhmmss: &str) -> bool {
|
||||
let normalized = match at.matches(':').count() {
|
||||
1 => format!("{at}:00"),
|
||||
_ => at.to_string(),
|
||||
};
|
||||
normalized == hhmmss
|
||||
}
|
||||
|
||||
/// Spawn an automation run, honoring `RunMode::Single` re-entrancy
|
||||
/// guard (HC-WS-05). For `Single`/`IgnoreFirst` modes a run already in
|
||||
/// flight causes the new trigger to be skipped; the `running` flag is
|
||||
/// cleared when the run finishes.
|
||||
fn spawn_run(hc: &HomeCore, automation: Arc<Automation>, running: Arc<AtomicBool>) {
|
||||
let single = matches!(automation.mode, RunMode::Single | RunMode::IgnoreFirst);
|
||||
if single {
|
||||
// Try to claim the running slot; if already running, skip.
|
||||
if running
|
||||
.compare_exchange(false, true, Ordering::SeqCst, Ordering::SeqCst)
|
||||
.is_err()
|
||||
{
|
||||
return;
|
||||
}
|
||||
}
|
||||
let hc_clone = hc.clone();
|
||||
tokio::spawn(async move {
|
||||
let mut exec_ctx = ExecutionContext::new(hc_clone, automation.id.clone());
|
||||
for action in &automation.action {
|
||||
if let Err(e) = action.execute(&mut exec_ctx).await {
|
||||
eprintln!("[homecore-automation] action error in {}: {e}", automation.id);
|
||||
break;
|
||||
}
|
||||
}
|
||||
if single {
|
||||
running.store(false, Ordering::SeqCst);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
|
|
@ -166,7 +367,6 @@ mod tests {
|
|||
|
||||
let _handle = engine.start();
|
||||
|
||||
// Fire a matching state change
|
||||
hc.states().set(
|
||||
EntityId::parse("switch.living").unwrap(),
|
||||
"on",
|
||||
|
|
@ -174,7 +374,6 @@ mod tests {
|
|||
Context::new(),
|
||||
);
|
||||
|
||||
// Give the async task time to run
|
||||
sleep(Duration::from_millis(50)).await;
|
||||
|
||||
assert_eq!(log.lock().unwrap().len(), 1);
|
||||
|
|
@ -203,7 +402,6 @@ mod tests {
|
|||
|
||||
let _handle = engine.start();
|
||||
|
||||
// Fire on a DIFFERENT entity
|
||||
hc.states().set(
|
||||
EntityId::parse("switch.bedroom").unwrap(),
|
||||
"on",
|
||||
|
|
@ -249,4 +447,16 @@ mod tests {
|
|||
sleep(Duration::from_millis(50)).await;
|
||||
assert_eq!(log.lock().unwrap().len(), 0, "disabled automation should not fire");
|
||||
}
|
||||
|
||||
// Behavioral tests for the timer / run-mode / template paths
|
||||
// (HC-WS-04/05/07) live in `tests/engine_behaviors.rs` to keep this
|
||||
// file under the 500-line guideline; they use only the public API.
|
||||
|
||||
#[test]
|
||||
fn time_at_matches_handles_hh_mm_and_hh_mm_ss() {
|
||||
assert!(time_at_matches("07:30", "07:30:00"));
|
||||
assert!(time_at_matches("07:30:15", "07:30:15"));
|
||||
assert!(!time_at_matches("07:30", "07:30:01"));
|
||||
assert!(!time_at_matches("07:30:15", "07:30:16"));
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -150,7 +150,12 @@ impl Trigger {
|
|||
true
|
||||
}
|
||||
Trigger::Time { .. } => {
|
||||
// Time triggers are evaluated by the engine's timer task, not here.
|
||||
// Time triggers are wall-clock based and have no state-change
|
||||
// context to match here. They are evaluated by the engine's
|
||||
// 1 Hz timer task (`AutomationEngine::start_timer`, HC-WS-04 /
|
||||
// ADR-161), which compares the trigger's `at` against the local
|
||||
// wall-clock second. `matches_sync` therefore returns false for
|
||||
// `Time` on the state-change path by design.
|
||||
false
|
||||
}
|
||||
Trigger::Event { event_type } => {
|
||||
|
|
|
|||
|
|
@ -0,0 +1,259 @@
|
|||
//! Engine behavioral integration tests (ADR-161, HC-WS-04/05/07).
|
||||
//!
|
||||
//! These exercise the `AutomationEngine` runtime through its public API
|
||||
//! only (extracted from the inline module to keep `engine.rs` under the
|
||||
//! 500-line file guideline):
|
||||
//!
|
||||
//! - HC-WS-04 — `time:` triggers fire via the engine timer path.
|
||||
//! - HC-WS-05 — `RunMode::Single` does not double-fire; `Parallel` does.
|
||||
//! - HC-WS-07 — `template:` conditions evaluate against live state in the
|
||||
//! engine path (no longer always-false).
|
||||
//!
|
||||
//! Each fails on the pre-fix engine (no timer task, unbounded-parallel
|
||||
//! regardless of mode, `template_env: None`).
|
||||
|
||||
use std::sync::atomic::{AtomicUsize, Ordering};
|
||||
use std::sync::{Arc, Mutex};
|
||||
|
||||
use homecore::service::FnHandler;
|
||||
use homecore::{Context, EntityId, HomeCore, ServiceCall, ServiceName};
|
||||
use homecore_automation::{Action, Automation, AutomationEngine, Condition, RunMode, Trigger};
|
||||
use tokio::time::{sleep, Duration};
|
||||
|
||||
async fn register_recorder(
|
||||
hc: &HomeCore,
|
||||
domain: &str,
|
||||
service: &str,
|
||||
) -> Arc<Mutex<Vec<serde_json::Value>>> {
|
||||
let log: Arc<Mutex<Vec<serde_json::Value>>> = Arc::new(Mutex::new(vec![]));
|
||||
let log2 = Arc::clone(&log);
|
||||
hc.services()
|
||||
.register(
|
||||
ServiceName::new(domain, service),
|
||||
FnHandler(move |call: ServiceCall| {
|
||||
let l = Arc::clone(&log2);
|
||||
async move {
|
||||
l.lock().unwrap().push(call.data.clone());
|
||||
Ok(serde_json::Value::Null)
|
||||
}
|
||||
}),
|
||||
)
|
||||
.await;
|
||||
log
|
||||
}
|
||||
|
||||
// ── HC-WS-04: time triggers fire ───────────────────────────────────
|
||||
#[tokio::test]
|
||||
async fn time_trigger_fires_via_timer_path() {
|
||||
let hc = HomeCore::new();
|
||||
let log = register_recorder(&hc, "light", "turn_on").await;
|
||||
|
||||
let engine = AutomationEngine::new(hc.clone());
|
||||
engine.register(Automation::new(
|
||||
"time_auto",
|
||||
vec![Trigger::Time { at: "07:30:00".into() }],
|
||||
vec![Action::ServiceCall {
|
||||
domain: "light".into(),
|
||||
service: "turn_on".into(),
|
||||
data: serde_json::json!({"by": "time"}),
|
||||
}],
|
||||
));
|
||||
|
||||
// Deterministically fire the timer path for the matching second.
|
||||
let fired = engine.fire_time_for_test("07:30:00").await;
|
||||
assert_eq!(fired, 1, "time automation should fire for matching HH:MM:SS");
|
||||
sleep(Duration::from_millis(50)).await;
|
||||
assert_eq!(log.lock().unwrap().len(), 1, "time trigger should run its action");
|
||||
|
||||
// A non-matching second must NOT fire.
|
||||
let none = engine.fire_time_for_test("09:00:00").await;
|
||||
assert_eq!(none, 0);
|
||||
}
|
||||
|
||||
// ── HC-WS-05: RunMode::Single does not double-fire ─────────────────
|
||||
#[tokio::test]
|
||||
async fn single_mode_does_not_double_fire_on_rapid_triggers() {
|
||||
let hc = HomeCore::new();
|
||||
let count = Arc::new(AtomicUsize::new(0));
|
||||
let count2 = Arc::clone(&count);
|
||||
hc.services()
|
||||
.register(
|
||||
ServiceName::new("light", "slow"),
|
||||
FnHandler(move |_call: ServiceCall| {
|
||||
let c = Arc::clone(&count2);
|
||||
async move {
|
||||
c.fetch_add(1, Ordering::SeqCst);
|
||||
sleep(Duration::from_millis(200)).await;
|
||||
Ok(serde_json::Value::Null)
|
||||
}
|
||||
}),
|
||||
)
|
||||
.await;
|
||||
|
||||
let engine = AutomationEngine::new(hc.clone());
|
||||
let mut auto = Automation::new(
|
||||
"single_auto",
|
||||
vec![Trigger::State {
|
||||
entity_id: EntityId::parse("switch.s").unwrap(),
|
||||
from: None,
|
||||
to: None,
|
||||
}],
|
||||
vec![Action::ServiceCall {
|
||||
domain: "light".into(),
|
||||
service: "slow".into(),
|
||||
data: serde_json::json!({}),
|
||||
}],
|
||||
);
|
||||
auto.mode = RunMode::Single;
|
||||
engine.register(auto);
|
||||
let _handle = engine.start();
|
||||
|
||||
// Two rapid triggers while the first run is still sleeping.
|
||||
hc.states().set(EntityId::parse("switch.s").unwrap(), "a", serde_json::json!({}), Context::new());
|
||||
sleep(Duration::from_millis(20)).await;
|
||||
hc.states().set(EntityId::parse("switch.s").unwrap(), "b", serde_json::json!({}), Context::new());
|
||||
|
||||
sleep(Duration::from_millis(350)).await;
|
||||
assert_eq!(
|
||||
count.load(Ordering::SeqCst),
|
||||
1,
|
||||
"Single-mode automation must not double-fire while already running"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn parallel_mode_does_fire_concurrently() {
|
||||
let hc = HomeCore::new();
|
||||
let count = Arc::new(AtomicUsize::new(0));
|
||||
let count2 = Arc::clone(&count);
|
||||
hc.services()
|
||||
.register(
|
||||
ServiceName::new("light", "slow"),
|
||||
FnHandler(move |_call: ServiceCall| {
|
||||
let c = Arc::clone(&count2);
|
||||
async move {
|
||||
c.fetch_add(1, Ordering::SeqCst);
|
||||
sleep(Duration::from_millis(150)).await;
|
||||
Ok(serde_json::Value::Null)
|
||||
}
|
||||
}),
|
||||
)
|
||||
.await;
|
||||
|
||||
let engine = AutomationEngine::new(hc.clone());
|
||||
let mut auto = Automation::new(
|
||||
"parallel_auto",
|
||||
vec![Trigger::State {
|
||||
entity_id: EntityId::parse("switch.p").unwrap(),
|
||||
from: None,
|
||||
to: None,
|
||||
}],
|
||||
vec![Action::ServiceCall {
|
||||
domain: "light".into(),
|
||||
service: "slow".into(),
|
||||
data: serde_json::json!({}),
|
||||
}],
|
||||
);
|
||||
auto.mode = RunMode::Parallel;
|
||||
engine.register(auto);
|
||||
let _handle = engine.start();
|
||||
|
||||
hc.states().set(EntityId::parse("switch.p").unwrap(), "a", serde_json::json!({}), Context::new());
|
||||
sleep(Duration::from_millis(20)).await;
|
||||
hc.states().set(EntityId::parse("switch.p").unwrap(), "b", serde_json::json!({}), Context::new());
|
||||
|
||||
sleep(Duration::from_millis(300)).await;
|
||||
assert_eq!(
|
||||
count.load(Ordering::SeqCst),
|
||||
2,
|
||||
"Parallel-mode automation should fire on every trigger"
|
||||
);
|
||||
}
|
||||
|
||||
// ── HC-WS-07: template conditions evaluate in the engine path ──────
|
||||
#[tokio::test]
|
||||
async fn template_condition_evaluates_true_in_engine() {
|
||||
let hc = HomeCore::new();
|
||||
let log = register_recorder(&hc, "light", "turn_on").await;
|
||||
|
||||
hc.states().set(
|
||||
EntityId::parse("sensor.flag").unwrap(),
|
||||
"on",
|
||||
serde_json::json!({}),
|
||||
Context::new(),
|
||||
);
|
||||
|
||||
let engine = AutomationEngine::new(hc.clone());
|
||||
let mut auto = Automation::new(
|
||||
"tmpl_auto",
|
||||
vec![Trigger::State {
|
||||
entity_id: EntityId::parse("switch.trigger").unwrap(),
|
||||
from: None,
|
||||
to: None,
|
||||
}],
|
||||
vec![Action::ServiceCall {
|
||||
domain: "light".into(),
|
||||
service: "turn_on".into(),
|
||||
data: serde_json::json!({}),
|
||||
}],
|
||||
);
|
||||
auto.condition = vec![Condition::Template {
|
||||
value_template: "{{ is_state('sensor.flag', 'on') }}".into(),
|
||||
}];
|
||||
engine.register(auto);
|
||||
let _handle = engine.start();
|
||||
|
||||
hc.states().set(
|
||||
EntityId::parse("switch.trigger").unwrap(),
|
||||
"go",
|
||||
serde_json::json!({}),
|
||||
Context::new(),
|
||||
);
|
||||
sleep(Duration::from_millis(50)).await;
|
||||
assert_eq!(
|
||||
log.lock().unwrap().len(),
|
||||
1,
|
||||
"template condition should evaluate true and let the action run (HC-WS-07)"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn template_condition_evaluates_false_blocks_action() {
|
||||
let hc = HomeCore::new();
|
||||
let log = register_recorder(&hc, "light", "turn_on").await;
|
||||
hc.states().set(
|
||||
EntityId::parse("sensor.flag").unwrap(),
|
||||
"off",
|
||||
serde_json::json!({}),
|
||||
Context::new(),
|
||||
);
|
||||
|
||||
let engine = AutomationEngine::new(hc.clone());
|
||||
let mut auto = Automation::new(
|
||||
"tmpl_auto_false",
|
||||
vec![Trigger::State {
|
||||
entity_id: EntityId::parse("switch.trigger").unwrap(),
|
||||
from: None,
|
||||
to: None,
|
||||
}],
|
||||
vec![Action::ServiceCall {
|
||||
domain: "light".into(),
|
||||
service: "turn_on".into(),
|
||||
data: serde_json::json!({}),
|
||||
}],
|
||||
);
|
||||
auto.condition = vec![Condition::Template {
|
||||
value_template: "{{ is_state('sensor.flag', 'on') }}".into(),
|
||||
}];
|
||||
engine.register(auto);
|
||||
let _handle = engine.start();
|
||||
|
||||
hc.states().set(
|
||||
EntityId::parse("switch.trigger").unwrap(),
|
||||
"go",
|
||||
serde_json::json!({}),
|
||||
Context::new(),
|
||||
);
|
||||
sleep(Duration::from_millis(50)).await;
|
||||
assert_eq!(log.lock().unwrap().len(), 0, "false template condition should block the action");
|
||||
}
|
||||
|
|
@ -121,8 +121,21 @@ async fn main() -> Result<()> {
|
|||
let _ = plugin_registry; // wired-but-empty at boot; integrations register here
|
||||
|
||||
// ── 4. Automation engine ────────────────────────────────────────
|
||||
let _automation_engine = AutomationEngine::new(hc.clone());
|
||||
info!("Automation engine ready (no automations loaded yet)");
|
||||
// Construct AND start the engine (HC-WS-03, ADR-161). `start()`
|
||||
// spawns the state-change event loop + the 1 Hz wall-clock timer
|
||||
// task so state/numeric/event AND time triggers all fire. The
|
||||
// engine is kept alive for the process lifetime (it is moved into a
|
||||
// long-lived binding); its background tasks run until the HomeCore
|
||||
// broadcast channel closes at shutdown. No automations are loaded at
|
||||
// boot yet (YAML loader is P-next); integrations register via
|
||||
// `engine.register(..)`.
|
||||
let automation_engine = AutomationEngine::new(hc.clone());
|
||||
let _automation_task = automation_engine.start();
|
||||
info!(
|
||||
"Automation engine started ({} automations registered) — \
|
||||
state/numeric/event + time triggers active",
|
||||
automation_engine.len()
|
||||
);
|
||||
|
||||
// ── 5. Assist pipeline ──────────────────────────────────────────
|
||||
let recognizer = RegexIntentRecognizer::new();
|
||||
|
|
|
|||
Loading…
Reference in New Issue