From ffe94a7edfe93011e169c6f751a014aaaffe0e6a Mon Sep 17 00:00:00 2001 From: RaresKeY <158580472+RaresKeY@users.noreply.github.com> Date: Mon, 20 Jul 2026 21:00:58 +0000 Subject: [PATCH] fix(email): fence stale urgency scans --- src/builtin_actions.py | 134 ++++++++++++++++--- tests/test_email_urgency_checkpoint.py | 177 ++++++++++++++++++++++++- 2 files changed, 288 insertions(+), 23 deletions(-) diff --git a/src/builtin_actions.py b/src/builtin_actions.py index 936c6db60..b62524727 100644 --- a/src/builtin_actions.py +++ b/src/builtin_actions.py @@ -20,6 +20,64 @@ from src.interactive_gate import wait_for_interactive_quiet logger = logging.getLogger(__name__) +def _read_email_urgency_state(state_path): + """Read one atomic urgency checkpoint, tolerating the legacy shape.""" + from pathlib import Path + + state_path = Path(state_path) + try: + state = ( + json.loads(state_path.read_text(encoding="utf-8")) + if state_path.exists() + else {} + ) + except Exception: + return {} + return state if isinstance(state, dict) else {} + + +def _email_urgency_account_generations(state): + """Return normalized per-account checkpoint/complete generations. + + Checkpoint generations fence every accepted state mutation. Complete + generations advance only for a non-stale complete scan. Missing metadata + is the legacy generation zero. + """ + raw = state.get("account_generations", {}) if isinstance(state, dict) else {} + if not isinstance(raw, dict): + return {} + + generations = {} + for account_id, value in raw.items(): + if isinstance(value, dict): + checkpoint = value.get("checkpoint", 0) + complete = value.get("complete", 0) + else: + # Tolerate an intermediate scalar representation as one completed + # checkpoint generation instead of discarding its fence. + checkpoint = value + complete = value + try: + checkpoint = max(0, int(checkpoint)) + except (TypeError, ValueError): + checkpoint = 0 + try: + complete = max(0, int(complete)) + except (TypeError, ValueError): + complete = 0 + generations[str(account_id)] = { + "checkpoint": checkpoint, + "complete": complete, + } + return generations + + +def _email_urgency_string_set(value): + if not isinstance(value, (list, tuple, set, frozenset)): + return set() + return {str(item) for item in value if isinstance(item, (str, int))} + + def _acquire_email_urgency_state_lock( state_path, lock_db_path, @@ -60,17 +118,7 @@ def _acquire_email_urgency_state_lock( 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 + return conn, _read_email_urgency_state(state_path) return None, None @@ -181,29 +229,72 @@ def _merge_email_urgency_state( notified_uids, all_unread_keys, fully_scanned_account_ids, + base_account_generations, timestamp, ): - """Merge current scan knowledge without deleting unscanned accounts.""" + """Merge a scan without letting an older snapshot erase newer facts.""" 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} + prior_generations = _email_urgency_account_generations(prior) + base_generations = _email_urgency_account_generations( + {"account_generations": base_account_generations} + ) + observed_accounts = { + _email_urgency_account_key(key) for key in per_uid_scores + } | complete + stale_accounts = { + account_id + for account_id in observed_accounts + if prior_generations.get(account_id, {}).get("checkpoint", 0) + != base_generations.get(account_id, {}).get("checkpoint", 0) + } + fresh_complete = complete - stale_accounts + changed_accounts = set(fresh_complete) merged_per_uid = dict(prior_per_uid) for key in list(merged_per_uid): - if _email_urgency_account_key(key) in complete: + account_id = _email_urgency_account_key(key) + if account_id in fresh_complete: merged_per_uid.pop(key, None) + changed_accounts.add(account_id) # 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) + # is not evidence that another checkpoint or UI row is stale. When another + # worker committed after this scan captured its base generation, its value + # also wins collisions; the stale scan may still add facts it alone saw. + for key, value in per_uid_scores.items(): + account_id = _email_urgency_account_key(key) + if account_id in stale_accounts and key in merged_per_uid: + continue + if merged_per_uid.get(key) != value: + changed_accounts.add(account_id) + merged_per_uid[key] = value - merged_notified = set(notified_uids) + prior_notified = _email_urgency_string_set(prior.get("notified_uids", [])) + merged_notified = prior_notified | _email_urgency_string_set(notified_uids) + for key in merged_notified - prior_notified: + changed_accounts.add(_email_urgency_account_key(key)) for key in list(merged_notified): if ( - _email_urgency_account_key(key) in complete + _email_urgency_account_key(key) in fresh_complete and key not in all_unread_keys ): merged_notified.discard(key) + changed_accounts.add(_email_urgency_account_key(key)) + + next_generations = { + account_id: dict(value) + for account_id, value in prior_generations.items() + } + for account_id in changed_accounts: + generation = next_generations.setdefault( + account_id, + {"checkpoint": 0, "complete": 0}, + ) + generation["checkpoint"] += 1 + if account_id in fresh_complete: + generation["complete"] += 1 total_unread = 0 total_urgent = 0 @@ -229,6 +320,7 @@ def _merge_email_urgency_state( "max_score": max_score, "per_uid": merged_per_uid, "notified_uids": sorted(merged_notified), + "account_generations": next_generations, } @@ -2134,6 +2226,13 @@ async def action_check_email_urgency(owner: str, **kwargs) -> Tuple[str, bool]: if not accounts: raise TaskNoop("no email accounts configured") + # Capture each account's checkpoint generation before any IMAP work. + # The locked merge compares this basis with the latest committed state, + # so an older scan cannot destructively replace a newer worker's facts. + base_account_generations = _email_urgency_account_generations( + _read_email_urgency_state(STATE_PATH) + ) + urgency_prompt = settings.get("urgent_email_prompt", "") per_uid_scores = {} # key = ":" → {"score": 0-3, "reason": "..."} all_unread_keys = set() @@ -2714,6 +2813,7 @@ async def action_check_email_urgency(owner: str, **kwargs) -> Tuple[str, bool]: notified_uids=notified_uids, all_unread_keys=all_unread_keys, fully_scanned_account_ids=fully_scanned_account_ids, + base_account_generations=base_account_generations, timestamp=_time.time(), ) return notified_uids, next_state diff --git a/tests/test_email_urgency_checkpoint.py b/tests/test_email_urgency_checkpoint.py index ae7ef1dc3..b9ed330a2 100644 --- a/tests/test_email_urgency_checkpoint.py +++ b/tests/test_email_urgency_checkpoint.py @@ -45,10 +45,11 @@ class _EmailAccount: class _FakeImap: - def __init__(self, account_id, failures, seen_accounts): + def __init__(self, account_id, failures, seen_accounts, search_uids): self.account_id = account_id self.failures = failures self.seen_accounts = seen_accounts + self.search_uids = tuple(search_uids.get(account_id, ("1",))) def select(self, *_args, **_kwargs): if self.account_id in self.failures: @@ -59,16 +60,18 @@ class _FakeImap: if self.account_id in self.failures: raise RuntimeError(f"{self.account_id} unavailable") if command == "SEARCH": - return "OK", [b"1"] + return "OK", [" ".join(self.search_uids).encode()] + uid = _args[0] + uid = uid.decode() if isinstance(uid, bytes) else str(uid) 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() + flags = f"{uid} (UID {uid} FLAGS ({seen}))".encode() if query == "(UID FLAGS)": return "OK", [flags] raw = ( f"From: Sender {self.account_id} \r\n" - f"Subject: Urgent request for {self.account_id}\r\n" - f"Message-ID: <{self.account_id}-1@example.com>\r\n" + f"Subject: Urgent request for {self.account_id} uid {uid}\r\n" + f"Message-ID: <{self.account_id}-{uid}@example.com>\r\n" "\r\n" "Please reply immediately." ).encode() @@ -97,6 +100,7 @@ def _configure_action(monkeypatch, tmp_path, account_ids): "accounts": list(account_ids), "failures": set(), "seen_accounts": set(), + "search_uids": {}, "settings": { "reminder_channel": "browser", "reminder_llm_synthesis": False, @@ -136,7 +140,10 @@ def _configure_action(monkeypatch, tmp_path, account_ids): email_helpers, "_imap_connect", lambda account_id=None, **_kwargs: _FakeImap( - str(account_id), runtime["failures"], runtime["seen_accounts"] + str(account_id), + runtime["failures"], + runtime["seen_accounts"], + runtime["search_uids"], ), ) return builtin_actions, runtime @@ -194,6 +201,94 @@ async def test_urgency_state_transaction_serializes_decision_and_checkpoint(tmp_ } +def test_stale_complete_scan_preserves_newer_account_facts_and_checkpoints(): + from src.builtin_actions import _merge_email_urgency_state + + prior = { + "owner": "alice", + "per_uid": { + "acct:1": {"score": 3, "unread": True, "reason": "newer"}, + "acct:2": {"score": 2, "unread": True, "reason": "new UID"}, + }, + "notified_uids": ["acct:1", "acct:2"], + "account_generations": { + "acct": {"checkpoint": 1, "complete": 1}, + }, + } + + merged = _merge_email_urgency_state( + prior, + owner="alice", + per_uid_scores={ + "acct:1": {"score": 0, "unread": False, "reason": "older"}, + "acct:3": {"score": 3, "unread": True, "reason": "also observed"}, + }, + notified_uids={"acct:1", "acct:2", "acct:3"}, + all_unread_keys={"acct:3"}, + fully_scanned_account_ids={"acct"}, + base_account_generations={ + "acct": {"checkpoint": 0, "complete": 0}, + }, + timestamp=300.0, + ) + + assert merged["per_uid"]["acct:1"]["reason"] == "newer" + assert set(merged["per_uid"]) == {"acct:1", "acct:2", "acct:3"} + assert merged["notified_uids"] == ["acct:1", "acct:2", "acct:3"] + assert merged["account_generations"]["acct"] == { + "checkpoint": 2, + "complete": 1, + } + + +def test_newer_partial_checkpoint_fences_older_complete_snapshot(): + from src.builtin_actions import _merge_email_urgency_state + + legacy = { + "owner": "alice", + "per_uid": { + "acct:1": {"score": 3, "unread": True}, + }, + "notified_uids": ["acct:1"], + } + partial = _merge_email_urgency_state( + legacy, + owner="alice", + per_uid_scores={ + "acct:2": {"score": 3, "unread": True}, + }, + notified_uids={"acct:1", "acct:2"}, + all_unread_keys={"acct:2"}, + fully_scanned_account_ids=set(), + base_account_generations={}, + timestamp=200.0, + ) + assert partial["account_generations"]["acct"] == { + "checkpoint": 1, + "complete": 0, + } + + merged = _merge_email_urgency_state( + partial, + owner="alice", + per_uid_scores={ + "acct:1": {"score": 3, "unread": True}, + }, + notified_uids={"acct:1", "acct:2"}, + all_unread_keys={"acct:1"}, + fully_scanned_account_ids={"acct"}, + base_account_generations={}, + timestamp=300.0, + ) + + assert set(merged["per_uid"]) == {"acct:1", "acct:2"} + assert merged["notified_uids"] == ["acct:1", "acct:2"] + assert merged["account_generations"]["acct"] == { + "checkpoint": 1, + "complete": 0, + } + + @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 @@ -372,6 +467,76 @@ async def test_action_cancellation_rolls_back_without_checkpoint( } +@pytest.mark.asyncio +async def test_older_same_account_scan_cannot_erase_newer_committed_uid( + monkeypatch, + tmp_path, +): + import routes.note_routes as note_routes + + builtin_actions, runtime = _configure_action( + monkeypatch, tmp_path, ["acct-a"] + ) + runtime["search_uids"]["acct-a"] = ["1"] + 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) + original_transaction = builtin_actions._run_email_urgency_state_transaction + stale_scan_ready = asyncio.Event() + release_stale_scan = asyncio.Event() + transaction_count = 0 + + async def order_transactions(state_path, lock_db_path, operation): + nonlocal transaction_count + transaction_count += 1 + if transaction_count == 1: + stale_scan_ready.set() + await release_stale_scan.wait() + return await original_transaction(state_path, lock_db_path, operation) + + monkeypatch.setattr( + builtin_actions, + "_run_email_urgency_state_transaction", + order_transactions, + ) + + stale = asyncio.create_task( + builtin_actions.action_check_email_urgency("alice") + ) + await asyncio.wait_for(stale_scan_ready.wait(), timeout=2) + + runtime["search_uids"]["acct-a"] = ["1", "2"] + newer_result = await asyncio.wait_for( + builtin_actions.action_check_email_urgency("alice"), + timeout=2, + ) + release_stale_scan.set() + stale_result = await asyncio.wait_for(stale, timeout=2) + + state = json.loads( + (tmp_path / "email_urgency_state_alice.json").read_text(encoding="utf-8") + ) + assert newer_result[1] is True + assert stale_result[1] is True + assert len(deliveries) == 1 + assert "uid 2" in deliveries[0] + assert set(state["per_uid"]) == {"acct-a:1", "acct-a:2"} + assert state["notified_uids"] == ["acct-a:1", "acct-a:2"] + assert state["account_generations"]["acct-a"] == { + "checkpoint": 1, + "complete": 1, + } + + @pytest.mark.asyncio async def test_account_scoped_actions_merge_disjoint_checkpoints( monkeypatch,