diff --git a/src/builtin_actions.py b/src/builtin_actions.py index b62524727..d4c1d0c59 100644 --- a/src/builtin_actions.py +++ b/src/builtin_actions.py @@ -221,6 +221,23 @@ def _email_urgency_account_key(message_key): return str(message_key).split(":", 1)[0] +def _email_urgency_stale_accounts( + prior, + base_account_generations, + account_ids, +): + prior_generations = _email_urgency_account_generations(prior) + base_generations = _email_urgency_account_generations( + {"account_generations": base_account_generations} + ) + return { + str(account_id) + for account_id in account_ids + if prior_generations.get(str(account_id), {}).get("checkpoint", 0) + != base_generations.get(str(account_id), {}).get("checkpoint", 0) + } + + def _merge_email_urgency_state( prior, *, @@ -238,18 +255,14 @@ def _merge_email_urgency_state( 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) - } + stale_accounts = _email_urgency_stale_accounts( + prior, + base_account_generations, + observed_accounts, + ) fresh_complete = complete - stale_accounts changed_accounts = set(fresh_complete) @@ -261,20 +274,26 @@ def _merge_email_urgency_state( 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. 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. + # worker committed after this scan captured its base generation, discard + # this account's whole stale snapshot. A key absent from the newer state + # may have been removed/read, so even a stale-only key is not safely + # additive without another fresh scan. 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: + if account_id in stale_accounts: continue if merged_per_uid.get(key) != value: changed_accounts.add(account_id) merged_per_uid[key] = value 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)) + merged_notified = set(prior_notified) + for key in _email_urgency_string_set(notified_uids) - prior_notified: + account_id = _email_urgency_account_key(key) + if account_id in stale_accounts: + continue + merged_notified.add(key) + changed_accounts.add(account_id) for key in list(merged_notified): if ( _email_urgency_account_key(key) in fresh_complete @@ -2731,7 +2750,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")] - total_urgent = len(urgent_keys) # ── 5. Fire a reminder only when a previously-unnotified UID scores # urgent. The read, decision, delivery, and checkpoint are serialized @@ -2739,39 +2757,46 @@ async def action_check_email_urgency(owner: str, **kwargs) -> Tuple[str, bool]: # state or overwrite each other's successful checkpoint. newly_notified = set() notify_failed = set() - title = "Urgent email" if total_urgent == 1 else f"{total_urgent} urgent emails" - # Build a real listing — subject · sender · reason for each urgent one - # — so the reminder tells the user which messages to act on. - sorted_urgent = sorted( - ((k, per_uid_scores[k]) for k in urgent_keys), - key=lambda kv: kv[1].get("score", 0), reverse=True, - )[:10] - _pub = (settings.get("app_public_url") or "").strip().rstrip("/") - from urllib.parse import quote as _quote - lines = [f"{total_urgent} email" + ("" if total_urgent == 1 else "s") + " need an urgent reply:", ""] - for i, (k, v) in enumerate(sorted_urgent, 1): - subj = (v.get("subject") or "(no subject)")[:160] - frm = v.get("from") or "" - why = v.get("reason") or "" - uid_for_link = str(k).split(":", 1)[-1] - hash_link = f"#email={_quote('INBOX', safe='')}:{uid_for_link}" - open_link = f"{_pub}/{hash_link}" if _pub else hash_link - line = f"{i}. {subj}" - if frm: - line += f" — {frm}" - if why: - line += f" · {why}" - lines.append(line) - lines.append(f" Open email: {open_link}") - if total_urgent > len(sorted_urgent): - lines.append("") - lines.append(f"…and {total_urgent - len(sorted_urgent)} more.") - body = "\n".join(lines) - async def _dispatch_urgency_reminder(): + def _urgency_reminder_payload(reminder_keys): + total = len(reminder_keys) + title = "Urgent email" if total == 1 else f"{total} urgent emails" + sorted_urgent = sorted( + ((key, per_uid_scores[key]) for key in reminder_keys), + key=lambda item: item[1].get("score", 0), + reverse=True, + )[:10] + _pub = (settings.get("app_public_url") or "").strip().rstrip("/") + from urllib.parse import quote as _quote + lines = [ + f"{total} email" + ("" if total == 1 else "s") + + " need an urgent reply:", + "", + ] + for i, (key, value) in enumerate(sorted_urgent, 1): + subj = (value.get("subject") or "(no subject)")[:160] + frm = value.get("from") or "" + why = value.get("reason") or "" + uid_for_link = str(key).split(":", 1)[-1] + hash_link = f"#email={_quote('INBOX', safe='')}:{uid_for_link}" + open_link = f"{_pub}/{hash_link}" if _pub else hash_link + line = f"{i}. {subj}" + if frm: + line += f" — {frm}" + if why: + line += f" · {why}" + lines.append(line) + lines.append(f" Open email: {open_link}") + if total > len(sorted_urgent): + lines.append("") + lines.append(f"…and {total - len(sorted_urgent)} more.") + return title, "\n".join(lines) + + async def _dispatch_urgency_reminder(reminder_keys): # Call dispatch_reminder directly: a scheduler has no browser # session cookie with which to call the HTTP endpoint. from routes.note_routes import dispatch_reminder + title, body = _urgency_reminder_payload(reminder_keys) return await dispatch_reminder( title=title, note_body=body, @@ -2780,11 +2805,35 @@ async def action_check_email_urgency(owner: str, **kwargs) -> Tuple[str, bool]: ) 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] + notified_uids = _email_urgency_string_set( + prior.get("notified_uids", []) + ) + observed_accounts = { + _email_urgency_account_key(key) for key in per_uid_scores + } | fully_scanned_account_ids + stale_accounts = _email_urgency_stale_accounts( + prior, + base_account_generations, + observed_accounts, + ) + # Generation fencing must happen before delivery, not only during + # merge. A stale-only unread UID may have been removed, read, or + # downgraded by the newer completed scan. + deliverable_urgent = [ + key + for key in urgent_keys + if _email_urgency_account_key(key) not in stale_accounts + ] + new_urgent = [ + key + for key in deliverable_urgent + if key not in notified_uids + ] if new_urgent: try: - dispatch_result = await _dispatch_urgency_reminder() + dispatch_result = await _dispatch_urgency_reminder( + deliverable_urgent + ) channel = (settings.get("reminder_channel") or "browser").strip().lower() delivered = bool(dispatch_result.get("browser_sent")) if channel == "email": diff --git a/tests/test_email_urgency_checkpoint.py b/tests/test_email_urgency_checkpoint.py index b9ed330a2..5392776ae 100644 --- a/tests/test_email_urgency_checkpoint.py +++ b/tests/test_email_urgency_checkpoint.py @@ -201,16 +201,16 @@ async def test_urgency_state_transaction_serializes_decision_and_checkpoint(tmp_ } -def test_stale_complete_scan_preserves_newer_account_facts_and_checkpoints(): +def test_stale_complete_scan_preserves_newer_state_and_discards_stale_only_facts(): from src.builtin_actions import _merge_email_urgency_state prior = { "owner": "alice", "per_uid": { - "acct:1": {"score": 3, "unread": True, "reason": "newer"}, + "acct:1": {"score": 0, "unread": True, "reason": "newer"}, "acct:2": {"score": 2, "unread": True, "reason": "new UID"}, }, - "notified_uids": ["acct:1", "acct:2"], + "notified_uids": ["acct:2"], "account_generations": { "acct": {"checkpoint": 1, "complete": 1}, }, @@ -233,10 +233,10 @@ def test_stale_complete_scan_preserves_newer_account_facts_and_checkpoints(): ) 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 set(merged["per_uid"]) == {"acct:1", "acct:2"} + assert merged["notified_uids"] == ["acct:2"] assert merged["account_generations"]["acct"] == { - "checkpoint": 2, + "checkpoint": 1, "complete": 1, } @@ -467,17 +467,27 @@ async def test_action_cancellation_rolls_back_without_checkpoint( } +@pytest.mark.parametrize( + ("stale_uids", "newer_uids"), + [ + (["1"], ["1", "2"]), + (["1", "2"], ["1"]), + ], + ids=["preserve-newer-addition", "reject-stale-only-delivery"], +) @pytest.mark.asyncio -async def test_older_same_account_scan_cannot_erase_newer_committed_uid( +async def test_stale_same_account_scan_cannot_override_newer_commit( monkeypatch, tmp_path, + stale_uids, + newer_uids, ): import routes.note_routes as note_routes builtin_actions, runtime = _configure_action( monkeypatch, tmp_path, ["acct-a"] ) - runtime["search_uids"]["acct-a"] = ["1"] + runtime["search_uids"]["acct-a"] = stale_uids deliveries = [] async def delivered(**kwargs): @@ -514,7 +524,7 @@ async def test_older_same_account_scan_cannot_erase_newer_committed_uid( ) await asyncio.wait_for(stale_scan_ready.wait(), timeout=2) - runtime["search_uids"]["acct-a"] = ["1", "2"] + runtime["search_uids"]["acct-a"] = newer_uids newer_result = await asyncio.wait_for( builtin_actions.action_check_email_urgency("alice"), timeout=2, @@ -528,15 +538,85 @@ async def test_older_same_account_scan_cannot_erase_newer_committed_uid( 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"] + for uid in newer_uids: + assert f"uid {uid}" in deliveries[0] + for uid in set(stale_uids) - set(newer_uids): + assert f"uid {uid}" not in deliveries[0] + expected_keys = {f"acct-a:{uid}" for uid in newer_uids} + assert set(state["per_uid"]) == expected_keys + assert state["notified_uids"] == sorted(expected_keys) assert state["account_generations"]["acct-a"] == { "checkpoint": 1, "complete": 1, } +@pytest.mark.asyncio +async def test_mixed_stale_and_fresh_accounts_exclude_stale_rows_from_delivery( + 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) + 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["accounts"] = ["acct-a"] + await asyncio.wait_for( + builtin_actions.action_check_email_urgency("alice"), + timeout=2, + ) + release_stale_scan.set() + await asyncio.wait_for(stale, timeout=2) + + state = json.loads( + (tmp_path / "email_urgency_state_alice.json").read_text(encoding="utf-8") + ) + assert len(deliveries) == 2 + assert "acct-a" in deliveries[0] + assert "acct-b" not in deliveries[0] + assert "acct-b" in deliveries[1] + assert "acct-a" not in deliveries[1] + assert set(state["per_uid"]) == {"acct-a:1", "acct-b:1"} + assert state["notified_uids"] == ["acct-a:1", "acct-b:1"] + + @pytest.mark.asyncio async def test_account_scoped_actions_merge_disjoint_checkpoints( monkeypatch,