Fix notification state after automation condition edits

docs/automations.md

@@ -138,6 +138,16 @@ Notify behavior:
   recipient. Restarting while that state is unchanged does not resend it. If
   the current state changed while Sensorius was stopped, the corresponding
   ACTIVATED or CLEARED message remains eligible until delivery succeeds.
+- Editing a Notify rule's conditions (including its sensor, metric, threshold,
+  or hysteresis) starts a new notification state. A currently true condition
+  can send a fresh ACTIVATED email; a currently false condition starts inactive
+  without a CLEARED email for the old conditions. Renaming or saving unchanged
+  conditions preserves the existing state. The manager stores an internal
+  `_notification_revision` in `script_json` so this behavior survives restarts;
+  existing rules without it retain their state until their conditions are edited.
+  Queued messages from superseded conditions are discarded when the evaluator
+  observes the edit. An SMTP send already in progress may still complete, but
+  cannot mark the edited conditions as delivered.
 
 Revert behavior:
 

sensorius/__init__.py

@@ -4,4 +4,4 @@
 release notes, and supporting tooling can report a consistent build identity.
 """
 
-__version__ = "v0.26.263.3"
+__version__ = "v0.26.269.1"

sensorius/saiAutomationManager.py

@@ -40,6 +40,7 @@
 import json
 import re
 import shutil
+import uuid
 import threading
 from pathlib import Path
 from typing import Any, Dict, Callable, TypeVar
@@ -655,6 +656,30 @@ def _mutate(data: Dict[str, Any]) -> None:
                 except Exception:
                     script_json = s  # store verbatim; runtime should validate before executing
 
+            # Notification state belongs to the conditions that produced it.
+            # Keep this metadata server-owned: the editor need not round-trip it.
+            previous_rule = adv.get(rule_id)
+            if previous_rule is None and rule_id not in self._ignored_legacy_rule_ids(data):
+                previous_rule = (self._load_legacy_data(hostname).get(SECTION_ADV) or {}).get(rule_id)
+            try:
+                incoming = json.loads(script_json)
+                previous = json.loads((previous_rule or {}).get("script_json", "{}"))
+            except (ValueError, TypeError):
+                incoming = previous = None
+            if isinstance(incoming, dict) and isinstance(previous, dict):
+                revision = previous.get("_notification_revision")
+                has_notify = any(
+                    str(action.get("type", "")).lower() == "notify"
+                    for action in (incoming.get("actions") or [])
+                    if isinstance(action, dict)
+                )
+                if previous_rule is not None and has_notify and incoming.get("conditions") != previous.get("conditions"):
+                    revision = uuid.uuid4().hex
+                incoming.pop("_notification_revision", None)
+                if revision:
+                    incoming["_notification_revision"] = revision
+                script_json = json.dumps(incoming, separators=(",", ":"), ensure_ascii=False)
+
             adv[rule_id] = {"enabled": bool(enabled), "script_json": script_json}
             data[SECTION_ADV] = adv
             meta = data.get(SECTION_META, {}) or {}

sensorius/saiEmailNotifications.py

@@ -283,6 +283,7 @@ def __init__(
         self._delivery_guards = self._load_delivery_guards()
         self._pending_by_rule: set[str] = set()
         self._pending_automation_targets: dict[str, set[bool]] = {}
+        self._automation_revisions: dict[str, str] = {}
         self._sent_epochs_fallback: deque[float] = deque()
         self._last_error = ""
 
@@ -451,13 +452,27 @@ def _enqueue_transition(
             })
 
     @staticmethod
-    def automation_delivery_id(rule_id: str, recipient: str) -> str:
+    def automation_delivery_id(rule_id: str, recipient: str, revision: str = "") -> str:
         """Return the stable persisted identity for one automation recipient."""
-        return f"automation:{str(rule_id or '').strip()}:{str(recipient or '').strip().lower()}"
+        base = f"automation:{str(rule_id or '').strip()}:{str(recipient or '').strip().lower()}"
+        return f"{base}:revision:{revision}" if revision else base
 
-    def persisted_automation_state(self, rule_id: str, recipient: str) -> bool:
-        """Return the last successfully delivered state for an automation actor."""
-        delivery_id = self.automation_delivery_id(rule_id, recipient)
+    def persisted_automation_state(self, rule_id: str, recipient: str, revision: str = "") -> bool:
+        """Reconcile obsolete queued edges and return this revision's sent state."""
+        base = self.automation_delivery_id(rule_id, recipient)
+        delivery_id = self.automation_delivery_id(rule_id, recipient, revision)
+        with self._queue_lock:
+            if self._automation_revisions.get(base) == revision:
+                return bool(self._active_by_rule.get(delivery_id, False))
+            self._automation_revisions[base] = revision
+            retained = deque()
+            for item in self._queue:
+                if item.get("automation_base_id") == base and item.get("revision", "") != revision:
+                    self._pending_by_rule.discard(item["delivery_id"])
+                    self._clear_automation_pending(item)
+                else:
+                    retained.append(item)
+            self._queue = retained
         return bool(self._active_by_rule.get(delivery_id, False))
 
     def enqueue_automation_transition(
@@ -468,13 +483,14 @@ def enqueue_automation_transition(
         recipient: str,
         subject: str,
         body: str,
+        revision: str = "",
     ) -> bool:
         """Queue one automation edge for guarded, persisted SMTP delivery."""
         rid = str(rule_id or "").strip()
         to_address = str(recipient or "").strip()
         if not rid or not to_address or not EmailConfig.from_environment().enabled:
             return False
-        delivery_id = self.automation_delivery_id(rid, to_address)
+        delivery_id = self.automation_delivery_id(rid, to_address, revision)
         guard = self._delivery_guards.get(delivery_id, {})
         if float(guard.get("failure_retry_after_epoch", 0.0) or 0.0) > time.time():
             return False
@@ -490,6 +506,8 @@ def enqueue_automation_transition(
                 "rule_id": delivery_id,
                 "delivery_id": pending_id,
                 "kind": "automation",
+                "automation_base_id": self.automation_delivery_id(rid, to_address),
+                "revision": revision,
                 "active": target_state,
                 "value": None,
                 "timestamp": "",
@@ -563,6 +581,11 @@ async def _deliver_item(self, item: dict[str, Any]) -> None:
         """Attempt one queued delivery and requeue it within the retry limit."""
         rule_id = str(item["rule_id"])
         delivery_id = str(item.get("delivery_id") or rule_id)
+        base = item.get("automation_base_id")
+        if base in self._automation_revisions and item.get("revision", "") != self._automation_revisions[base]:
+            self._pending_by_rule.discard(delivery_id)
+            self._clear_automation_pending(item)
+            return
         try:
             send_kwargs = {}
             if item.get("to_addresses"):

sensorius/saiSwitch.py

@@ -2256,6 +2256,9 @@ def _eval_single_condition(
                         webui_states[notification_key] = rule_ok
                         continue
                     if action_type == "notify":
+                        revision_args = {}
+                        if script.get("_notification_revision"):
+                            revision_args["revision"] = str(script["_notification_revision"])
                         executor_sid = str(act.get("executor_switch_id", "") or "").strip()
                         own_sid = str(getattr(self, "switch_id", "") or "").strip()
                         if not executor_sid or executor_sid.lower() != own_sid.lower():
@@ -2268,6 +2271,7 @@ def _eval_single_condition(
                                 delivery_service.persisted_automation_state(
                                     str(_rule_id),
                                     recipient,
+                                    **revision_args,
                                 )
                             )
                         else:
@@ -2327,6 +2331,7 @@ def _eval_single_condition(
                                     recipient=recipient,
                                     subject=subject,
                                     body=body,
+                                    **revision_args,
                                 )
                         elif delivery_service is None:
                             notify_states[notify_key] = rule_ok

testApparatus/test_notification_rule_edits.py

@@ -0,0 +1,190 @@
+"""Verify notification state follows edited conditions across saves and restarts.
+
+Exercise the real manager, evaluator, delivery queue, and SQLite persistence
+with a fake sender so these regressions never send external email.
+"""
+
+import asyncio
+import copy
+import json
+import threading
+
+import pytest
+
+from sensorius.saiAutomationManager import AutomationManager
+from sensorius.saiDataLogger import saiDataLogger
+from sensorius.saiEmailNotifications import AutomationNotificationService, EmailNotificationService
+
+
+def notification_rule():
+    return {
+        "name": "Sunlight",
+        "enabled": True,
+        "conditions": [{"type": "sensor", "sensor": "removed", "metric": "Light Intensity",
+                        "op": ">", "value": 1100, "hyst": 100}],
+        "actions": [{"type": "notify", "to": "grower@example.com", "executor_switch_id": "__system__"}],
+    }
+
+
+@pytest.mark.parametrize("field,value", [
+    ("sensor", "lux"), ("metric", "Auto Light"), ("value", 1000),
+    ("hyst", 0), ("op", "<"),
+])
+def test_condition_edits_get_revision_but_renames_and_resaves_keep_it(tmp_path, field, value):
+    manager = AutomationManager(str(tmp_path / "automations"))
+    original = notification_rule()
+    manager.upsert_advanced_rule("__system__", "sun", enabled=True, script=original)
+    changed = copy.deepcopy(original)
+    changed["conditions"][0][field] = value
+    manager.upsert_advanced_rule("__system__", "sun", enabled=True, script=changed)
+    revision = manager.load_runtime_advanced("__system__")["sun"]["script_json"]["_notification_revision"]
+    changed["name"] = "Renamed"
+    manager.upsert_advanced_rule("__system__", "sun", enabled=True, script=changed)
+    assert manager.load_runtime_advanced("__system__")["sun"]["script_json"]["_notification_revision"] == revision
+    # Returning to old conditions is another edit, not a return to old sent state.
+    manager.upsert_advanced_rule("__system__", "sun", enabled=True, script=original)
+    assert manager.load_runtime_advanced("__system__")["sun"]["script_json"]["_notification_revision"] != revision
+
+
+@pytest.mark.parametrize("reading,expected_count", [(1809, 1), (0, 0)])
+def test_replaced_sensor_does_not_inherit_active_state(tmp_path, monkeypatch, reading, expected_count):
+    monkeypatch.setenv("SENSORIUS_EMAIL_ENABLED", "true")
+    manager = AutomationManager(str(tmp_path / "automations"))
+    rule = notification_rule()
+    manager.upsert_advanced_rule("__system__", "sun", enabled=True, script=rule)
+    logger = saiDataLogger(db_path=str(tmp_path / "notifications.db"))
+    sent = []
+
+    class Sender:
+        def send(self, subject, body, **kwargs):
+            sent.append(subject)
+
+    try:
+        legacy_id = EmailNotificationService.automation_delivery_id("sun", "grower@example.com")
+        logger.set_notification_rule_state(legacy_id, True, None, "")
+        service = EmailNotificationService(settings=None, data_logger=logger, sender=Sender(), evaluate_readings=False)
+        controller = AutomationNotificationService(data_logger=logger, email_delivery_service=service)._get_controller()
+        controller._load_triggers_dict = lambda: {"Advanced": manager.load_runtime_advanced("__system__")}
+        controller._get_values_for_sensor = lambda *args: {"Light Intensity": reading}
+        rule["conditions"][0]["sensor"] = "lux"
+        manager.upsert_advanced_rule("__system__", "sun", enabled=True, script=rule)
+        controller._evaluate_and_apply_advanced({})
+        assert len(service._queue) == expected_count
+        if expected_count:
+            asyncio.run(service._deliver_item(service._pop()))
+            assert sent[0].startswith("Sensorius ACTIVATED:")
+        # A new service reloads SQLite state, and a new manager reloads the TOML revision.
+        restarted = EmailNotificationService(settings=None, data_logger=logger, sender=Sender(), evaluate_readings=False)
+        controller.email_delivery_service = restarted
+        manager = AutomationManager(str(tmp_path / "automations"))
+        controller._evaluate_and_apply_advanced({})
+        assert restarted._pop() is None
+        assert len(sent) == expected_count
+    finally:
+        logger.close()
+
+
+def test_edit_discards_queued_and_popped_old_revision(tmp_path, monkeypatch):
+    monkeypatch.setenv("SENSORIUS_EMAIL_ENABLED", "true")
+    logger = saiDataLogger(db_path=str(tmp_path / "notifications.db"))
+    sent = []
+
+    class Sender:
+        def send(self, *args, **kwargs):
+            sent.append(args)
+
+    try:
+        service = EmailNotificationService(settings=None, data_logger=logger, sender=Sender())
+        args = dict(rule_id="sun", recipient="grower@example.com", subject="old", body="old")
+        service.enqueue_automation_transition(**args, triggered=True)
+        popped = service._pop()
+        service.enqueue_automation_transition(**args, triggered=False)
+        assert not service.persisted_automation_state("sun", "grower@example.com", revision="new")
+        assert service._pop() is None
+        asyncio.run(service._deliver_item(popped))
+        assert sent == []
+        assert not service._pending_by_rule
+        assert not service._pending_automation_targets
+    finally:
+        logger.close()
+
+
+def test_editing_legacy_rule_assigns_revision_without_changing_legacy_file(tmp_path):
+    legacy_root = tmp_path / "switch_settings"
+    legacy = AutomationManager(str(legacy_root / "automations"))
+    rule = notification_rule()
+    legacy.upsert_advanced_rule("__system__", "sun", enabled=True, script=rule)
+    old_bytes = legacy.get_storage_path().read_bytes()
+    manager = AutomationManager(str(tmp_path / "automations"), legacy_base_dir=str(legacy_root))
+    rule["conditions"][0]["sensor"] = "lux"
+    manager.upsert_advanced_rule("__system__", "sun", enabled=True, script=rule)
+    assert manager.load_runtime_advanced("__system__")["sun"]["script_json"]["_notification_revision"]
+    assert legacy.get_storage_path().read_bytes() == old_bytes
+
+
+def test_inflight_old_email_cannot_activate_edited_conditions(tmp_path, monkeypatch):
+    monkeypatch.setenv("SENSORIUS_EMAIL_ENABLED", "true")
+    logger = saiDataLogger(db_path=str(tmp_path / "notifications.db"))
+
+    async def exercise():
+        started = asyncio.Event()
+        release = threading.Event()
+        loop = asyncio.get_running_loop()
+
+        class Sender:
+            def send(self, *args, **kwargs):
+                loop.call_soon_threadsafe(started.set)
+                assert release.wait(5)
+
+        service = EmailNotificationService(settings=None, data_logger=logger, sender=Sender())
+        service.persisted_automation_state("sun", "grower@example.com")
+        service.enqueue_automation_transition(rule_id="sun", recipient="grower@example.com",
+                                              triggered=True, subject="old", body="old")
+        task = asyncio.create_task(service._deliver_item(service._pop()))
+        try:
+            await asyncio.wait_for(started.wait(), 5)
+            assert not service.persisted_automation_state("sun", "grower@example.com", revision="new")
+        finally:
+            release.set()
+            await task
+        assert not service.persisted_automation_state("sun", "grower@example.com", revision="new")
+        restarted = EmailNotificationService(settings=None, data_logger=logger)
+        assert not restarted.persisted_automation_state("sun", "grower@example.com", revision="new")
+
+    try:
+        asyncio.run(exercise())
+    finally:
+        logger.close()
+
+
+@pytest.mark.asyncio
+async def test_editor_save_preserves_server_owned_revision(tmp_path, monkeypatch):
+    from httpx import ASGITransport, AsyncClient
+    from sensorius import saiAutomationManager
+    from testApparatus.test_nodus_settings_schema_writes import _build_app
+
+    app, *_ = await _build_app(tmp_path, monkeypatch)
+    monkeypatch.setenv("SENSORIUS_EMAIL_ENABLED", "true")
+
+    class TestManager(AutomationManager):
+        def __init__(self, _base_dir="automation_settings"):
+            super().__init__(str(tmp_path / "automations"))
+
+    monkeypatch.setattr(saiAutomationManager, "AutomationManager", TestManager)
+    rule = notification_rule()
+    payload = {"switch_id": "__system__", "rule_id": "sun", "enabled": "true"}
+    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
+        for sensor in ["removed", "lux", "lux"]:
+            rule["conditions"][0]["sensor"] = sensor
+            response = await client.post("/submit-advanced-trigger", json={
+                **payload, "script_json": json.dumps(rule),
+            })
+            assert response.status_code == 200, response.text
+            saved = TestManager().load_runtime_advanced("__system__")["sun"]["script_json"]
+            if sensor == "removed":
+                assert "_notification_revision" not in saved
+                revision = None
+            elif revision is None:
+                revision = saved["_notification_revision"]
+            else:
+                assert saved["_notification_revision"] == revision