erabo/odysseus
1
0
Fork 0
mirror of https://github.com/pewdiepie-archdaemon/odysseus.git synced 2026-10-14 01:47:55 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 3

fix(email): fence stale urgency delivery

Browse source
This commit is contained in:
RaresKeY 2026-07-20 21:12:37 +00:00
parent ffe94a7edf
commit 60aabbd9e2
2 changed files with 189 additions and 60 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

145
src/builtin_actions.py
View file

@ -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":

104
tests/test_email_urgency_checkpoint.py
View file

@ -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,