From 49daa73106579f119b1bbc6dfef5618caf6eee79 Mon Sep 17 00:00:00 2001
From: RaresKeY <158580472+RaresKeY@users.noreply.github.com>
Date: Mon, 20 Jul 2026 20:05:40 +0000
Subject: [PATCH] fix(email): preserve urgency transaction lifecycle

---
 src/builtin_actions.py                 | 316 +++++++++++++---
 tests/test_email_urgency_checkpoint.py | 475 ++++++++++++++++++++++++-
 2 files changed, 720 insertions(+), 71 deletions(-)

diff --git a/src/builtin_actions.py b/src/builtin_actions.py
index b560385db..936c6db60 100644
--- a/src/builtin_actions.py
+++ b/src/builtin_actions.py
@@ -20,41 +20,87 @@ from src.interactive_gate import wait_for_interactive_quiet
 logger = logging.getLogger(__name__)
 
 
-def _run_email_urgency_state_transaction(state_path, lock_db_path, operation):
-    """Serialize one owner urgency checkpoint across worker processes.
-
-    ``operation`` receives the latest JSON state and returns
-    ``(result, next_state)``.  The scheduled-email database is used only as a
-    durable cross-process write lock; the UI-facing state remains the existing
-    JSON file.  Running the operation in a worker thread keeps the async event
-    loop free while reminder delivery is in flight.
-    """
+def _acquire_email_urgency_state_lock(
+    state_path,
+    lock_db_path,
+    cancel_event,
+    timeout_seconds=120,
+):
+    """Acquire the cross-process urgency lock without blocking the app loop."""
     import sqlite3
-    import uuid
+    import time
     from pathlib import Path
 
     state_path = Path(state_path)
     state_path.parent.mkdir(parents=True, exist_ok=True)
-    conn = sqlite3.connect(str(lock_db_path), timeout=120)
-    temp_path = state_path.with_name(
-        f".{state_path.name}.{uuid.uuid4().hex}.tmp"
-    )
-    try:
-        conn.execute("BEGIN IMMEDIATE")
+    deadline = time.monotonic() + timeout_seconds
+
+    while not cancel_event.is_set():
+        remaining = deadline - time.monotonic()
+        if remaining <= 0:
+            raise sqlite3.OperationalError("timed out waiting for urgency state lock")
+        conn = sqlite3.connect(
+            str(lock_db_path),
+            timeout=min(0.25, max(0.01, remaining)),
+            check_same_thread=False,
+        )
         try:
-            prior = json.loads(state_path.read_text(encoding="utf-8")) \
-                if state_path.exists() else {}
+            conn.execute("BEGIN IMMEDIATE")
+        except sqlite3.OperationalError as exc:
+            conn.close()
+            if "locked" not in str(exc).lower():
+                raise
+            cancel_event.wait(min(0.05, max(0.0, remaining)))
+            continue
+        except BaseException:
+            conn.close()
+            raise
+
+        if cancel_event.is_set():
+            conn.rollback()
+            conn.close()
+            return None, None
+        try:
+            prior = (
+                json.loads(state_path.read_text(encoding="utf-8"))
+                if state_path.exists()
+                else {}
+            )
             if not isinstance(prior, dict):
                 prior = {}
         except Exception:
             prior = {}
+        return conn, prior
 
-        result, next_state = operation(prior)
+    return None, None
+
+
+def _close_email_urgency_state_lock(conn):
+    if conn is None:
+        return
+    try:
+        try:
+            conn.rollback()
+        except Exception:
+            pass
+    finally:
+        conn.close()
+
+
+def _commit_email_urgency_state(conn, state_path, next_state):
+    """Atomically publish JSON before releasing the SQLite write lock."""
+    import uuid
+    from pathlib import Path
+
+    state_path = Path(state_path)
+    temp_path = state_path.with_name(
+        f".{state_path.name}.{uuid.uuid4().hex}.tmp"
+    )
+    try:
         temp_path.write_text(json.dumps(next_state), encoding="utf-8")
         temp_path.replace(state_path)
         conn.commit()
-        return result
-    except Exception:
+    except BaseException:
         conn.rollback()
         raise
     finally:
@@ -62,6 +108,130 @@ def _run_email_urgency_state_transaction(state_path, lock_db_path, operation):
         conn.close()
 
 
+async def _run_email_urgency_state_transaction(
+    state_path,
+    lock_db_path,
+    operation,
+):
+    """Serialize one urgency decision while keeping async work on this loop.
+
+    Only lock acquisition waits in a worker thread. ``operation`` is awaited
+    on the caller's long-lived event loop, where shared async clients, locks,
+    and the browser-notification queue belong. Cancellation rolls back the
+    SQLite transaction and never publishes a checkpoint.
+    """
+    import asyncio
+    import threading
+
+    loop = asyncio.get_running_loop()
+    cancel_event = threading.Event()
+    acquire_future = loop.run_in_executor(
+        None,
+        _acquire_email_urgency_state_lock,
+        state_path,
+        lock_db_path,
+        cancel_event,
+    )
+    try:
+        conn, prior = await asyncio.shield(acquire_future)
+    except asyncio.CancelledError as cancelled:
+        cancel_event.set()
+        # The acquisition worker owns any connection until it returns. Wait
+        # for its short busy-poll to observe cancellation, then close a lock it
+        # may have won concurrently with the cancellation request.
+        while True:
+            try:
+                conn, _prior = await asyncio.shield(acquire_future)
+                break
+            except asyncio.CancelledError:
+                continue
+            except Exception:
+                conn = None
+                break
+        _close_email_urgency_state_lock(conn)
+        raise cancelled
+
+    if conn is None:
+        raise asyncio.CancelledError
+
+    try:
+        result, next_state = await operation(prior)
+        # Keep this small atomic publish synchronous. There is no await between
+        # the successful operation and commit, so cancellation cannot be
+        # observed and then followed by a checkpoint.
+        try:
+            _commit_email_urgency_state(conn, state_path, next_state)
+        finally:
+            conn = None
+        return result
+    except BaseException:
+        _close_email_urgency_state_lock(conn)
+        raise
+
+
+def _email_urgency_account_key(message_key):
+    return str(message_key).split(":", 1)[0]
+
+
+def _merge_email_urgency_state(
+    prior,
+    *,
+    owner,
+    per_uid_scores,
+    notified_uids,
+    all_unread_keys,
+    fully_scanned_account_ids,
+    timestamp,
+):
+    """Merge current scan knowledge without deleting unscanned accounts."""
+    prior_per_uid = prior.get("per_uid", {})
+    if not isinstance(prior_per_uid, dict):
+        prior_per_uid = {}
+    complete = {str(account_id) for account_id in fully_scanned_account_ids}
+
+    merged_per_uid = dict(prior_per_uid)
+    for key in list(merged_per_uid):
+        if _email_urgency_account_key(key) in complete:
+            merged_per_uid.pop(key, None)
+    # Partial scans may add or refresh facts, but absence from a partial scan
+    # is not evidence that another checkpoint or UI row is stale.
+    merged_per_uid.update(per_uid_scores)
+
+    merged_notified = set(notified_uids)
+    for key in list(merged_notified):
+        if (
+            _email_urgency_account_key(key) in complete
+            and key not in all_unread_keys
+        ):
+            merged_notified.discard(key)
+
+    total_unread = 0
+    total_urgent = 0
+    max_score = 0
+    for value in merged_per_uid.values():
+        if not isinstance(value, dict):
+            continue
+        try:
+            score = max(0, min(3, int(value.get("score", 0))))
+        except (TypeError, ValueError):
+            score = 0
+        max_score = max(max_score, score)
+        if value.get("unread"):
+            total_unread += 1
+            if score >= 2:
+                total_urgent += 1
+
+    return {
+        "ts": timestamp,
+        "owner": owner or "",
+        "total_unread": total_unread,
+        "total_urgent": total_urgent,
+        "max_score": max_score,
+        "per_uid": merged_per_uid,
+        "notified_uids": sorted(merged_notified),
+    }
+
+
 class TaskNoop(BaseException):
     """Raised by an action when it determined there's nothing to do.
 
@@ -1972,6 +2142,7 @@ async def action_check_email_urgency(owner: str, **kwargs) -> Tuple[str, bool]:
         failed_classifications = []
         tag_write_details = []
         scanned = 0
+        fully_scanned_account_ids = set()
 
         def _heuristic_email_verdict(item: dict) -> dict:
             blob = (
@@ -2067,16 +2238,27 @@ async def action_check_email_urgency(owner: str, **kwargs) -> Tuple[str, bool]:
             def _scan_one(account=acc, cache_uids=cache.get("uids", {})):
                 """Sync IMAP work runs in a thread."""
                 results = []
+                scan_complete = True
                 conn = _imap_connect(account.id)
                 try:
-                    conn.select("INBOX", readonly=True)
+                    select_status, _select_data = conn.select("INBOX", readonly=True)
+                    if select_status != "OK":
+                        return results, False
                     # Tag recent inbox mail, not only unread mail. Urgency
                     # reminders below still only notify for unread messages.
                     since_str = AGE_CUTOFF.strftime("%d-%b-%Y")
                     status, data = conn.uid("SEARCH", None, f'(SINCE {since_str})')
-                    if status != "OK" or not data or not data[0]:
-                        return results
-                    uids = data[0].split()[-30:]
+                    if status != "OK":
+                        return results, False
+                    if not data or not data[0]:
+                        return results, True
+                    matching_uids = data[0].split()
+                    if len(matching_uids) > 30:
+                        # The scale guard deliberately processes only the most
+                        # recent 30. That is a partial account snapshot, so it
+                        # cannot justify pruning older checkpoint facts.
+                        scan_complete = False
+                    uids = matching_uids[-30:]
                     for uid_b in uids:
                         uid = uid_b.decode() if isinstance(uid_b, bytes) else str(uid_b)
                         key = f"{account.id}:{uid}"
@@ -2084,12 +2266,41 @@ async def action_check_email_urgency(owner: str, **kwargs) -> Tuple[str, bool]:
                         cached_ok = isinstance(cached, dict) and cached.get("triage_version") == TRIAGE_VERSION
                         results.append({"key": key, "uid": uid, "cached": cached if cached_ok else None})
                         if cached_ok:
-                            # Already classified — skip the fetch.
+                            # Cached verdicts still need a lightweight FLAGS
+                            # refresh. Without it a cached unread message looks
+                            # read and its successful notification checkpoint
+                            # is pruned on the next pass.
+                            try:
+                                st, flag_data = conn.uid("FETCH", uid_b, "(UID FLAGS)")
+                                if st != "OK" or not flag_data:
+                                    scan_complete = False
+                                    results.pop()
+                                    continue
+                                flag_parts = []
+                                for part in flag_data:
+                                    if isinstance(part, (bytes, bytearray)):
+                                        flag_parts.append(bytes(part))
+                                    elif (
+                                        isinstance(part, tuple)
+                                        and part
+                                        and isinstance(part[0], (bytes, bytearray))
+                                    ):
+                                        flag_parts.append(bytes(part[0]))
+                                flags_blob = b" ".join(flag_parts)
+                                results[-1]["unread"] = b"\\Seen" not in flags_blob
+                            except Exception as _fe:
+                                scan_complete = False
+                                results.pop()
+                                logger.debug(
+                                    f"urgency: flag fetch for uid {uid} failed: {_fe}"
+                                )
                             continue
                         # Pull headers + first ~800 chars of plaintext body.
                         try:
                             st, msg_data = conn.uid("FETCH", uid_b, "(UID FLAGS RFC822.HEADER BODY.PEEK[TEXT]<0.800>)")
                             if st != "OK" or not msg_data:
+                                scan_complete = False
+                                results.pop()
                                 continue
                             flags_blob = b" ".join(
                                 part[0] for part in msg_data
@@ -2103,6 +2314,8 @@ async def action_check_email_urgency(owner: str, **kwargs) -> Tuple[str, bool]:
                                 if isinstance(part, tuple) and part[1]:
                                     raw += part[1] + b"\n\n"
                             if not raw:
+                                scan_complete = False
+                                results.pop()
                                 continue
                             msg = _email_mod.message_from_bytes(raw)
                             # Skip Odysseus-generated reminders so the scanner
@@ -2158,17 +2371,21 @@ async def action_check_email_urgency(owner: str, **kwargs) -> Tuple[str, bool]:
                                 "unread": is_unread,
                             })
                         except Exception as _fe:
+                            scan_complete = False
+                            results.pop()
                             logger.debug(f"urgency: header fetch for uid {uid} failed: {_fe}")
                 finally:
                     try: conn.logout()
                     except Exception: pass
-                return results
+                return results, scan_complete
 
             try:
-                items = await _aio.to_thread(_scan_one)
+                items, scan_complete = await _aio.to_thread(_scan_one)
             except Exception as e:
                 logger.warning(f"urgency: IMAP scan failed for account {acc.id}: {e}")
                 continue
+            if scan_complete:
+                fully_scanned_account_ids.add(str(acc.id))
 
             for item in items:
                 scanned += 1
@@ -2305,13 +2522,13 @@ async def action_check_email_urgency(owner: str, **kwargs) -> Tuple[str, bool]:
                     logger.debug(f"urgency: LLM classify failed for {key}: {e}")
                     continue
 
-            # ── Prune cache entries for UIDs that are no longer in the recent
-            # scan window. Read messages remain cached because tags are useful
-            # on read mail too; unread state is refreshed per scan above.
-            seen_uids = {it["uid"] for it in items}
-            cache_uids = cache.get("uids", {})
-            for stale in [u for u in cache_uids if u not in seen_uids]:
-                cache_uids.pop(stale, None)
+            if scan_complete:
+                # Only a complete account scan proves a cached UID left the
+                # recent window. Partial/failing scans preserve prior facts.
+                seen_uids = {it["uid"] for it in items}
+                cache_uids = cache.get("uids", {})
+                for stale in [u for u in cache_uids if u not in seen_uids]:
+                    cache_uids.pop(stale, None)
 
             try:
                 cache_file.write_text(_json.dumps(cache), encoding="utf-8")
@@ -2415,7 +2632,6 @@ async def action_check_email_urgency(owner: str, **kwargs) -> Tuple[str, bool]:
 
         # ── 4. Aggregate state. urgent = score ≥ 2.
         urgent_keys = [k for k, v in per_uid_scores.items() if v.get("score", 0) >= 2 and v.get("unread")]
-        max_score = max((v.get("score", 0) for v in per_uid_scores.values()), default=0)
         total_urgent = len(urgent_keys)
 
         # ── 5. Fire a reminder only when a previously-unnotified UID scores
@@ -2464,12 +2680,12 @@ async def action_check_email_urgency(owner: str, **kwargs) -> Tuple[str, bool]:
                 owner=owner or "",
             )
 
-        def _dispatch_and_checkpoint(prior):
+        async def _dispatch_and_checkpoint(prior):
             notified_uids = set(prior.get("notified_uids", []))
             new_urgent = [k for k in urgent_keys if k not in notified_uids]
             if new_urgent:
                 try:
-                    dispatch_result = _aio.run(_dispatch_urgency_reminder())
+                    dispatch_result = await _dispatch_urgency_reminder()
                     channel = (settings.get("reminder_channel") or "browser").strip().lower()
                     delivered = bool(dispatch_result.get("browser_sent"))
                     if channel == "email":
@@ -2491,23 +2707,19 @@ async def action_check_email_urgency(owner: str, **kwargs) -> Tuple[str, bool]:
                     logger.warning(f"urgency: reminder dispatch failed: {e}")
                     notify_failed.update(new_urgent)
 
-            # Prune UIDs no longer unread, while retaining every successful
-            # checkpoint observed after this worker acquired the durable lock.
-            notified_uids.intersection_update(all_unread_keys)
-            next_state = {
-                "ts": _time.time(),
-                "owner": owner or "",
-                "total_unread": len(all_unread_keys),
-                "total_urgent": total_urgent,
-                "max_score": max_score,
-                "per_uid": per_uid_scores,
-                "notified_uids": sorted(notified_uids),
-            }
+            next_state = _merge_email_urgency_state(
+                prior,
+                owner=owner,
+                per_uid_scores=per_uid_scores,
+                notified_uids=notified_uids,
+                all_unread_keys=all_unread_keys,
+                fully_scanned_account_ids=fully_scanned_account_ids,
+                timestamp=_time.time(),
+            )
             return notified_uids, next_state
 
         try:
-            await _aio.to_thread(
-                _run_email_urgency_state_transaction,
+            await _run_email_urgency_state_transaction(
                 STATE_PATH,
                 STATE_LOCK_DB,
                 _dispatch_and_checkpoint,
diff --git a/tests/test_email_urgency_checkpoint.py b/tests/test_email_urgency_checkpoint.py
index 1790cc29d..ae7ef1dc3 100644
--- a/tests/test_email_urgency_checkpoint.py
+++ b/tests/test_email_urgency_checkpoint.py
@@ -1,24 +1,164 @@
+import asyncio
 import json
 import threading
-from concurrent.futures import ThreadPoolExecutor
+from types import SimpleNamespace
+
+import pytest
 
 
-def test_urgency_state_transaction_serializes_decision_and_checkpoint(tmp_path):
+class _Column:
+    def __eq__(self, _other):
+        return True
+
+    def __ne__(self, _other):
+        return True
+
+
+class _Query:
+    def __init__(self, rows):
+        self._rows = rows
+
+    def filter(self, *_args, **_kwargs):
+        return self
+
+    def all(self):
+        return list(self._rows)
+
+
+class _Db:
+    def __init__(self, rows):
+        self._rows = rows
+
+    def query(self, _model):
+        return _Query(self._rows())
+
+    def close(self):
+        return None
+
+
+class _EmailAccount:
+    enabled = _Column()
+    owner = _Column()
+    imap_user = _Column()
+    from_address = _Column()
+    id = _Column()
+
+
+class _FakeImap:
+    def __init__(self, account_id, failures, seen_accounts):
+        self.account_id = account_id
+        self.failures = failures
+        self.seen_accounts = seen_accounts
+
+    def select(self, *_args, **_kwargs):
+        if self.account_id in self.failures:
+            raise RuntimeError(f"{self.account_id} unavailable")
+        return "OK", []
+
+    def uid(self, command, *_args):
+        if self.account_id in self.failures:
+            raise RuntimeError(f"{self.account_id} unavailable")
+        if command == "SEARCH":
+            return "OK", [b"1"]
+        query = str(_args[-1]) if _args else ""
+        seen = "\\Seen" if self.account_id in self.seen_accounts else ""
+        flags = f"1 (UID 1 FLAGS ({seen}))".encode()
+        if query == "(UID FLAGS)":
+            return "OK", [flags]
+        raw = (
+            f"From: Sender {self.account_id} <sender-{self.account_id}@example.com>\r\n"
+            f"Subject: Urgent request for {self.account_id}\r\n"
+            f"Message-ID: <{self.account_id}-1@example.com>\r\n"
+            "\r\n"
+            "Please reply immediately."
+        ).encode()
+        return "OK", [(flags, raw)]
+
+    def logout(self):
+        return None
+
+
+def _account(account_id):
+    return SimpleNamespace(
+        id=account_id,
+        enabled=True,
+        owner="alice",
+        imap_user="alice",
+        from_address="alice",
+    )
+
+
+def _configure_action(monkeypatch, tmp_path, account_ids):
+    from core import database
+    from routes import email_helpers
+    from src import builtin_actions, llm_core, settings, task_endpoint
+
+    runtime = {
+        "accounts": list(account_ids),
+        "failures": set(),
+        "seen_accounts": set(),
+        "settings": {
+            "reminder_channel": "browser",
+            "reminder_llm_synthesis": False,
+            "app_public_url": "",
+        },
+    }
+
+    monkeypatch.setattr(builtin_actions, "DATA_DIR", str(tmp_path))
+    monkeypatch.setattr(
+        builtin_actions,
+        "EMAIL_URGENCY_CACHE_DIR",
+        str(tmp_path / "urgency-cache"),
+    )
+    monkeypatch.setattr(database, "EmailAccount", _EmailAccount)
+    monkeypatch.setattr(
+        database,
+        "SessionLocal",
+        lambda: _Db(lambda: [_account(value) for value in runtime["accounts"]]),
+    )
+    monkeypatch.setattr(
+        task_endpoint,
+        "resolve_task_candidates",
+        lambda *args, **kwargs: [("http://llm", "model", {})],
+    )
+
+    async def fake_fallback(*_args, **_kwargs):
+        return '{"score": 3, "reason": "urgent"}'
+
+    monkeypatch.setattr(llm_core, "llm_call_async_with_fallback", fake_fallback)
+    monkeypatch.setattr(settings, "load_settings", lambda: dict(runtime["settings"]))
+    monkeypatch.setattr(
+        email_helpers,
+        "SCHEDULED_DB",
+        tmp_path / "scheduled-emails.db",
+    )
+    monkeypatch.setattr(
+        email_helpers,
+        "_imap_connect",
+        lambda account_id=None, **_kwargs: _FakeImap(
+            str(account_id), runtime["failures"], runtime["seen_accounts"]
+        ),
+    )
+    return builtin_actions, runtime
+
+
+@pytest.mark.asyncio
+async def test_urgency_state_transaction_serializes_decision_and_checkpoint(tmp_path):
     """A later worker must observe the first worker's delivered UID."""
     from src.builtin_actions import _run_email_urgency_state_transaction
 
     state_path = tmp_path / "email_urgency_state_alice.json"
-    lock_db = tmp_path / "scheduled-emails.db"
-    first_entered = threading.Event()
-    allow_first_to_finish = threading.Event()
-    second_entered = threading.Event()
+    lock_db = tmp_path / "urgency.lock.sqlite3"
+    first_entered = asyncio.Event()
+    allow_first_to_finish = asyncio.Event()
+    second_entered = asyncio.Event()
     deliveries = []
 
-    def operation(name):
-        def update(prior):
+    async def operation(name):
+        async def update(prior):
             if name == "first":
                 first_entered.set()
-                assert allow_first_to_finish.wait(timeout=2)
+                await allow_first_to_finish.wait()
             else:
                 second_entered.set()
 
@@ -32,25 +172,322 @@ def test_urgency_state_transaction_serializes_decision_and_checkpoint(tmp_path):
                 "notified_uids": sorted(notified),
             }
 
-        return _run_email_urgency_state_transaction(
+        return await _run_email_urgency_state_transaction(
             state_path,
             lock_db,
             update,
         )
 
-    with ThreadPoolExecutor(max_workers=2) as executor:
-        first = executor.submit(operation, "first")
-        assert first_entered.wait(timeout=2)
-        second = executor.submit(operation, "second")
-        assert not second_entered.wait(timeout=0.2)
-        allow_first_to_finish.set()
-
-        assert first.result(timeout=2) is True
-        assert second.result(timeout=2) is False
+    first = asyncio.create_task(operation("first"))
+    await asyncio.wait_for(first_entered.wait(), timeout=2)
+    second = asyncio.create_task(operation("second"))
+    await asyncio.sleep(0.1)
+    assert not second_entered.is_set()
+    allow_first_to_finish.set()
 
+    assert await asyncio.wait_for(first, timeout=2) is True
+    assert await asyncio.wait_for(second, timeout=2) is False
     assert deliveries == ["first"]
     assert json.loads(state_path.read_text(encoding="utf-8")) == {
         "owner": "alice",
         "notified_uids": ["acct:42"],
     }
 
+
+@pytest.mark.asyncio
+async def test_waiting_transaction_cancellation_does_not_leak_lock(tmp_path):
+    from src.builtin_actions import _run_email_urgency_state_transaction
+
+    state_path = tmp_path / "email_urgency_state_alice.json"
+    lock_db = tmp_path / "urgency.lock.sqlite3"
+    first_entered = asyncio.Event()
+    release_first = asyncio.Event()
+
+    async def first_operation(_prior):
+        first_entered.set()
+        await release_first.wait()
+        return None, {"notified_uids": ["acct:1"]}
+
+    async def later_operation(prior):
+        return None, prior
+
+    first = asyncio.create_task(
+        _run_email_urgency_state_transaction(
+            state_path, lock_db, first_operation
+        )
+    )
+    await asyncio.wait_for(first_entered.wait(), timeout=2)
+    waiting = asyncio.create_task(
+        _run_email_urgency_state_transaction(
+            state_path, lock_db, later_operation
+        )
+    )
+    await asyncio.sleep(0.05)
+    waiting.cancel()
+    with pytest.raises(asyncio.CancelledError):
+        await asyncio.wait_for(waiting, timeout=2)
+
+    release_first.set()
+    await asyncio.wait_for(first, timeout=2)
+    await asyncio.wait_for(
+        _run_email_urgency_state_transaction(
+            state_path, lock_db, later_operation
+        ),
+        timeout=2,
+    )
+
+
+@pytest.mark.asyncio
+async def test_action_dispatch_stays_on_app_loop_and_queues_browser_notification(
+    monkeypatch,
+    tmp_path,
+):
+    import routes.note_routes as note_routes
+    from src import endpoint_resolver, llm_core
+    from src.task_scheduler import TaskScheduler
+
+    builtin_actions, runtime = _configure_action(
+        monkeypatch, tmp_path, ["acct-a"]
+    )
+    runtime["settings"]["reminder_llm_synthesis"] = True
+    monkeypatch.setattr(note_routes, "DATA_DIR", str(tmp_path))
+    monkeypatch.setattr(
+        endpoint_resolver,
+        "resolve_endpoint",
+        lambda *_args, **_kwargs: (
+            "https://api.openai.com/v1",
+            "utility-model",
+            {},
+        ),
+    )
+
+    expected_loop = asyncio.get_running_loop()
+    expected_thread = threading.get_ident()
+    synthesis_loops = []
+    shared_client = SimpleNamespace(is_closed=False)
+    monkeypatch.setattr(llm_core, "_http_client", shared_client)
+    monkeypatch.setattr(llm_core, "_response_cache", {})
+
+    class _Response:
+        is_success = True
+        status_code = 200
+        text = "ok"
+
+        @staticmethod
+        def json():
+            return {
+                "choices": [
+                    {"message": {"content": "Synthesized urgency reminder."}}
+                ]
+            }
+
+    async def fake_http_post(client, *_args, **_kwargs):
+        synthesis_loops.append(asyncio.get_running_loop())
+        assert client is shared_client
+        return _Response()
+
+    monkeypatch.setattr(
+        llm_core,
+        "httpx_post_kimi_aware_async",
+        fake_http_post,
+    )
+
+    scheduler = TaskScheduler(None)
+    notification_threads = []
+    original_add = scheduler.add_notification
+
+    def checked_add(*args, **kwargs):
+        notification_threads.append(threading.get_ident())
+        return original_add(*args, **kwargs)
+
+    monkeypatch.setattr(scheduler, "add_notification", checked_add)
+    monkeypatch.setattr(note_routes, "_scheduler_ref", scheduler)
+
+    message, ok = await builtin_actions.action_check_email_urgency("alice")
+
+    assert ok is True
+    assert "notified 1" in message
+    assert synthesis_loops == [expected_loop]
+    assert notification_threads == [expected_thread]
+    notifications = scheduler.pop_notifications(owner="alice")
+    assert len(notifications) == 1
+    assert notifications[0]["body"] == "Synthesized urgency reminder."
+
+
+@pytest.mark.asyncio
+async def test_action_cancellation_rolls_back_without_checkpoint(
+    monkeypatch,
+    tmp_path,
+):
+    import routes.note_routes as note_routes
+
+    builtin_actions, _runtime = _configure_action(
+        monkeypatch, tmp_path, ["acct-a"]
+    )
+    state_path = tmp_path / "email_urgency_state_alice.json"
+    state_path.write_text(
+        json.dumps({"owner": "alice", "per_uid": {}, "notified_uids": []}),
+        encoding="utf-8",
+    )
+    entered = threading.Event()
+    release = threading.Event()
+    dispatch_cancelled = threading.Event()
+
+    async def blocked_dispatch(**_kwargs):
+        entered.set()
+        try:
+            while not release.is_set():
+                await asyncio.sleep(0.01)
+        except asyncio.CancelledError:
+            dispatch_cancelled.set()
+            raise
+        return {
+            "browser_sent": True,
+            "email_sent": False,
+            "ntfy_sent": False,
+            "webhook_sent": False,
+        }
+
+    monkeypatch.setattr(note_routes, "dispatch_reminder", blocked_dispatch)
+    task = asyncio.create_task(
+        builtin_actions.action_check_email_urgency("alice")
+    )
+    for _ in range(200):
+        if entered.is_set():
+            break
+        await asyncio.sleep(0.01)
+    assert entered.is_set()
+
+    task.cancel()
+    with pytest.raises(asyncio.CancelledError):
+        await task
+    release.set()
+    await asyncio.sleep(0.1)
+
+    assert dispatch_cancelled.is_set()
+    assert json.loads(state_path.read_text(encoding="utf-8")) == {
+        "owner": "alice",
+        "per_uid": {},
+        "notified_uids": [],
+    }
+
+
+@pytest.mark.asyncio
+async def test_account_scoped_actions_merge_disjoint_checkpoints(
+    monkeypatch,
+    tmp_path,
+):
+    import routes.note_routes as note_routes
+
+    builtin_actions, runtime = _configure_action(
+        monkeypatch, tmp_path, ["acct-a"]
+    )
+    deliveries = []
+
+    async def delivered(**kwargs):
+        deliveries.append(kwargs["note_body"])
+        return {
+            "browser_sent": True,
+            "email_sent": False,
+            "ntfy_sent": False,
+            "webhook_sent": False,
+        }
+
+    monkeypatch.setattr(note_routes, "dispatch_reminder", delivered)
+
+    await builtin_actions.action_check_email_urgency(
+        "alice", prompt='{"account_id":"acct-a"}'
+    )
+    runtime["accounts"] = ["acct-b"]
+    await builtin_actions.action_check_email_urgency(
+        "alice", prompt='{"account_id":"acct-b"}'
+    )
+    runtime["accounts"] = ["acct-a"]
+    await builtin_actions.action_check_email_urgency(
+        "alice", prompt='{"account_id":"acct-a"}'
+    )
+
+    state = json.loads(
+        (tmp_path / "email_urgency_state_alice.json").read_text(encoding="utf-8")
+    )
+    assert len(deliveries) == 2
+    assert "Urgent request for acct-a" in deliveries[0]
+    assert "Urgent request for acct-b" in deliveries[1]
+    assert state["notified_uids"] == ["acct-a:1", "acct-b:1"]
+    assert set(state["per_uid"]) == {"acct-a:1", "acct-b:1"}
+    assert state["total_unread"] == 2
+    assert state["total_urgent"] == 2
+
+
+@pytest.mark.asyncio
+async def test_failed_account_scan_preserves_checkpoint_until_recovery(
+    monkeypatch,
+    tmp_path,
+):
+    import routes.note_routes as note_routes
+
+    builtin_actions, runtime = _configure_action(
+        monkeypatch, tmp_path, ["acct-a", "acct-b"]
+    )
+    deliveries = []
+
+    async def delivered(**kwargs):
+        deliveries.append(kwargs["note_body"])
+        return {
+            "browser_sent": True,
+            "email_sent": False,
+            "ntfy_sent": False,
+            "webhook_sent": False,
+        }
+
+    monkeypatch.setattr(note_routes, "dispatch_reminder", delivered)
+
+    await builtin_actions.action_check_email_urgency("alice")
+    runtime["failures"] = {"acct-b"}
+    await builtin_actions.action_check_email_urgency("alice")
+
+    failed_state = json.loads(
+        (tmp_path / "email_urgency_state_alice.json").read_text(encoding="utf-8")
+    )
+    assert failed_state["notified_uids"] == ["acct-a:1", "acct-b:1"]
+    assert set(failed_state["per_uid"]) == {"acct-a:1", "acct-b:1"}
+
+    runtime["failures"] = set()
+    runtime["accounts"] = ["acct-b"]
+    await builtin_actions.action_check_email_urgency(
+        "alice", prompt='{"account_id":"acct-b"}'
+    )
+
+    assert len(deliveries) == 1
+
+
+@pytest.mark.asyncio
+async def test_cached_flags_refresh_prunes_checkpoint_after_message_is_read(
+    monkeypatch,
+    tmp_path,
+):
+    import routes.note_routes as note_routes
+
+    builtin_actions, runtime = _configure_action(
+        monkeypatch, tmp_path, ["acct-a"]
+    )
+
+    async def delivered(**_kwargs):
+        return {
+            "browser_sent": True,
+            "email_sent": False,
+            "ntfy_sent": False,
+            "webhook_sent": False,
+        }
+
+    monkeypatch.setattr(note_routes, "dispatch_reminder", delivered)
+    await builtin_actions.action_check_email_urgency("alice")
+    runtime["seen_accounts"] = {"acct-a"}
+    await builtin_actions.action_check_email_urgency("alice")
+
+    state = json.loads(
+        (tmp_path / "email_urgency_state_alice.json").read_text(encoding="utf-8")
+    )
+    assert state["notified_uids"] == []
+    assert state["per_uid"]["acct-a:1"]["unread"] is False
+    assert state["total_unread"] == 0