diff --git a/backend/app/plugins/actions.py b/backend/app/plugins/actions.py index 54a1dfe..eaf4063 100644 --- a/backend/app/plugins/actions.py +++ b/backend/app/plugins/actions.py @@ -13,6 +13,7 @@ spam message into Junk without dragging its thread along. from __future__ import annotations +import datetime as dt import uuid from collections.abc import Iterable from email import message_from_bytes @@ -65,6 +66,42 @@ class PluginMessages: except FileNotFoundError: return None + async def sample( + self, + user_id: uuid.UUID, + *, + account_id: uuid.UUID, + special_use: str, + limit: int = 500, + older_than_days: int | None = None, + ) -> list[uuid.UUID]: + """Message ids from one account folder, newest first, capped at ``limit``. + + For building a training corpus out of mail the user already has — the + provider's own Junk folder is a labelled spam set, and aged Inbox mail + that was never moved is a reasonable ham set. ``older_than_days`` skips + recent arrivals, which are the ones a user has not triaged yet. + + Ids only: the caller fetches bodies through :meth:`raw` one at a time, + so a 500-message sample never materialises 500 message bodies at once. + """ + stmt = ( + select(Message.id) + .join(Mailbox, Mailbox.id == Message.mailbox_id) + .where( + Message.user_id == user_id, + Message.account_id == account_id, + Mailbox.special_use == special_use, + ) + .order_by(Message.internal_date.desc()) + .limit(max(0, min(limit, 5000))) + ) + if older_than_days is not None: + cutoff = dt.datetime.now(dt.UTC) - dt.timedelta(days=older_than_days) + stmt = stmt.where(Message.internal_date < cutoff) + async with session_scope() as session: + return list(await session.scalars(stmt)) + async def threads_of( self, user_id: uuid.UUID, message_ids: Iterable[str] ) -> dict[str, str | None]: diff --git a/backend/app/sync/actions.py b/backend/app/sync/actions.py index 27c09a0..39c0062 100644 --- a/backend/app/sync/actions.py +++ b/backend/app/sync/actions.py @@ -126,6 +126,10 @@ async def move_job( return src_path = src.path uid, dst_path, user_id = message.uid, dst.path, message.user_id + # Carried into the event below: a consumer usually cares about the + # transition, not the destination. "Now in Spam" and "moved out of Spam" + # are different facts, and only the pair distinguishes them. + src_special, dst_special = src.special_use, dst.special_use provider = build_provider(account) try: @@ -145,7 +149,14 @@ async def move_job( message.mailbox_id = dst_id await get_event_bus().publish( "message.changed", - {"user_id": str(user_id), "message_id": str(mid), "mailbox_id": str(dst_id)}, + { + "user_id": str(user_id), + "message_id": str(mid), + "mailbox_id": str(dst_id), + "previous_mailbox_id": src_mailbox_id, + "special_use": dst_special, + "previous_special_use": src_special, + }, ) diff --git a/backend/tests/plugins/test_antispam_learning.py b/backend/tests/plugins/test_antispam_learning.py new file mode 100644 index 0000000..c239b67 --- /dev/null +++ b/backend/tests/plugins/test_antispam_learning.py @@ -0,0 +1,190 @@ +"""C2 learning loop: what teaches, what doesn't, and what only teaches once. + +Passive learning (SPEC §C.5) reads the user's own filing as a verdict. The value +is in the restraint — most folder moves say nothing about spam, and teaching the +same message twice skews the corpus. +""" + +from __future__ import annotations + +import uuid +from typing import Any + +import pytest +from webmail_plugin_antispam import hook, jobs, policy + +SETTINGS = policy.Settings( + enabled_accounts=[], + move_threshold=12.0, + banner_threshold=6.0, + network_checks=False, + learn_from_provider_junk=True, +) + + +class _FakeSdk: + def __init__(self, learned: str | None = None) -> None: + self.learned_before = learned + self.published: list[tuple[str, dict[str, Any]]] = [] + self.recorded: list[dict[str, Any]] = [] + self.enqueued: list[dict[str, Any]] = [] + + outer = self + + class _M: + async def envelope(self, *_a: object, **_k: object) -> dict[str, Any]: + return {"account_id": "acc-1"} + + async def raw(self, *_a: object, **_k: object) -> bytes: + return b"From: x@y.test\r\n\r\nbody\r\n" + + self.messages = _M() + + async def _enqueue(_f: object, **kw: Any) -> None: + outer.enqueued.append(kw) + + self.enqueue = _enqueue + + async def publish(self, event: str, payload: dict[str, Any]) -> None: + self.published.append((event, payload)) + + +@pytest.fixture +def wired(monkeypatch: pytest.MonkeyPatch): # type: ignore[no-untyped-def] + def _install(*, learned: str | None = None, engine_up: bool = True) -> _FakeSdk: + fake = _FakeSdk(learned) + taught: list[bool] = [] + + async def _learn(_raw: bytes, *, user_key: str, spam: bool) -> bool: + taught.append(spam) + return engine_up + + async def _already(_u: object, _m: object) -> str | None: + return fake.learned_before + + async def _remember(_u: object, _m: object, **kw: Any) -> None: + fake.recorded.append(kw) + + async def _settings(_u: object) -> policy.Settings: + return SETTINGS + + monkeypatch.setattr(hook, "_SDK", fake) + monkeypatch.setattr(hook.rspamd, "learn", _learn) + monkeypatch.setattr(hook, "_already_learned", _already) + monkeypatch.setattr(hook, "_remember_learned", _remember) + monkeypatch.setattr(policy, "load_settings", _settings) + fake.taught = taught # type: ignore[attr-defined] + return fake + + return _install + + +async def _move(before: str | None, now: str | None) -> None: + await hook.on_message_changed( + { + "user_id": str(uuid.uuid4()), + "message_id": str(uuid.uuid4()), + "previous_special_use": before, + "special_use": now, + } + ) + + +async def test_filing_into_spam_teaches_spam(wired) -> None: # type: ignore[no-untyped-def] + fake = wired() + await _move("inbox", "junk") + assert fake.taught == [True] + assert fake.recorded[0]["verdict"] == "spam" + + +async def test_rescuing_from_spam_teaches_ham(wired) -> None: # type: ignore[no-untyped-def] + fake = wired() + await _move("junk", "inbox") + assert fake.taught == [False] + assert fake.recorded[0]["verdict"] == "ham" + + +@pytest.mark.parametrize( + ("before", "now"), + [ + ("inbox", "archive"), # tidying up + ("junk", "trash"), # deleting spam is not a verdict about ham + ("archive", "inbox"), # unarchiving says nothing about spam + ("sent", "archive"), + ("inbox", "inbox"), # not a move at all + ], +) +async def test_other_moves_teach_nothing(wired, before, now) -> None: # type: ignore[no-untyped-def] + fake = wired() + await _move(before, now) + assert fake.taught == [] + assert fake.recorded == [] + + +async def test_the_same_lesson_is_not_taught_twice(wired) -> None: # type: ignore[no-untyped-def] + """Users move things back and forth; the corpus should not count that twice.""" + fake = wired(learned="spam") + await _move("inbox", "junk") + assert fake.taught == [] + + +async def test_reversing_a_lesson_does_teach(wired) -> None: # type: ignore[no-untyped-def] + """Having taught spam, moving it back out must be able to teach ham.""" + fake = wired(learned="spam") + await _move("junk", "inbox") + assert fake.taught == [False] + + +async def test_nothing_is_recorded_when_the_engine_is_down(wired) -> None: # type: ignore[no-untyped-def] + """`learned` stays unset so a later move can teach it again.""" + fake = wired(engine_up=False) + await _move("inbox", "junk") + assert fake.taught == [True] + assert fake.recorded == [] + + +# --- bootstrap --------------------------------------------------------------- + + +async def test_going_live_enqueues_the_bootstrap(wired) -> None: # type: ignore[no-untyped-def] + fake = wired() + await hook.on_account_state_changed( + {"user_id": str(uuid.uuid4()), "account_id": "acc-1", "state": "live"} + ) + assert fake.enqueued == [{"user_id": fake.enqueued[0]["user_id"], "account_id": "acc-1"}] + + +async def test_other_states_do_not_enqueue(wired) -> None: # type: ignore[no-untyped-def] + fake = wired() + for state in ("backfilling", "error", "paused"): + await hook.on_account_state_changed( + {"user_id": str(uuid.uuid4()), "account_id": "acc-1", "state": state} + ) + assert fake.enqueued == [] + + +async def test_bootstrap_respects_the_setting(monkeypatch: pytest.MonkeyPatch) -> None: + """Opting out of provider-Junk learning means no training pass at all.""" + sampled: list[str] = [] + + async def _settings(_u: object) -> policy.Settings: + return policy.Settings( + enabled_accounts=[], + move_threshold=12.0, + banner_threshold=6.0, + network_checks=False, + learn_from_provider_junk=False, + ) + + class _M: + async def sample(self, *_a: object, **kw: object) -> list[uuid.UUID]: + sampled.append(str(kw.get("special_use"))) + return [] + + class _Fake: + messages = _M() + + monkeypatch.setattr(policy, "load_settings", _settings) + monkeypatch.setattr(jobs, "_SDK", _Fake()) + await jobs.bootstrap_account({}, user_id=str(uuid.uuid4()), account_id=str(uuid.uuid4())) + assert sampled == [] diff --git a/docs/open-questions.md b/docs/open-questions.md index f2ba44d..4fdc8fe 100644 --- a/docs/open-questions.md +++ b/docs/open-questions.md @@ -31,6 +31,18 @@ mail path: - `PluginMessages.raw(user_id, message_id) -> bytes | None`. - `envelope()` now also returns `account_id` and `mailbox_special_use`. +**C2 needed two more, on the same reasoning:** + +4. **§C.5 passive learning** reads the user's own filing as a verdict, which + needs the *transition* — "now in Junk" and "moved out of Junk" are different + facts. `message.changed` carried only the destination, so `move_job` now also + publishes `previous_mailbox_id`, `special_use` and `previous_special_use`. +5. **§C.5 provider bootstrap** samples up to 500 provider-Junk messages and 500 + aged Inbox messages. The SDK could fetch a message by id but not enumerate a + folder, so `PluginMessages.sample(account_id, special_use, limit, + older_than_days)` returns ids only — the caller pulls bodies one at a time, + so a 500-message sample never materialises 500 bodies. + **Reading of "zero core diffs":** taken to mean the plugin must not require changes to core *mail flow or behaviour* — with the plugin absent, mail handling is byte-identical, which remains true. Extending the plugin SDK is what the SDK diff --git a/plugins/webmail_plugin_antispam/webmail_plugin_antispam/__init__.py b/plugins/webmail_plugin_antispam/webmail_plugin_antispam/__init__.py index 16aa3ef..a280bac 100644 --- a/plugins/webmail_plugin_antispam/webmail_plugin_antispam/__init__.py +++ b/plugins/webmail_plugin_antispam/webmail_plugin_antispam/__init__.py @@ -19,7 +19,13 @@ from __future__ import annotations from pathlib import Path from app.plugins.sdk import Plugin, Slot, UiManifest -from webmail_plugin_antispam.hook import on_message_new +from app.plugins.sdk import JobSpec +from webmail_plugin_antispam.hook import ( + on_account_state_changed, + on_message_changed, + on_message_new, +) +from webmail_plugin_antispam.jobs import bootstrap_account from webmail_plugin_antispam.router import router _HERE = Path(__file__).parent @@ -28,7 +34,16 @@ _UI = "/api/plugins/antispam/ui" plugin = Plugin( name="antispam", version="0.1.0", - hooks={"message.new": on_message_new}, + hooks={ + "message.new": on_message_new, + # Passive learning: where the user files things teaches the filter (§C.5). + "message.changed": on_message_changed, + # An account going live is the cue to warm Bayes from its history (§C.5). + "account.state_changed": on_account_state_changed, + }, + # Enqueue-on-demand (no schedule); singleton so a flapping account cannot + # start two training passes over the same mailbox. + jobs=[JobSpec(func=bootstrap_account, schedule=None, singleton=True)], router=router, migrations=str(_HERE / "migrations" / "versions"), ui_manifest=UiManifest( diff --git a/plugins/webmail_plugin_antispam/webmail_plugin_antispam/hook.py b/plugins/webmail_plugin_antispam/webmail_plugin_antispam/hook.py index 70ae1a6..f0ea327 100644 --- a/plugins/webmail_plugin_antispam/webmail_plugin_antispam/hook.py +++ b/plugins/webmail_plugin_antispam/webmail_plugin_antispam/hook.py @@ -139,3 +139,102 @@ async def on_message_new(payload: dict[str, Any]) -> None: "action": action, }, ) + + +async def _already_learned(user_id: uuid.UUID, message_id: uuid.UUID) -> str | None: + async with _SDK.session(user_id) as s: + row = ( + await s.execute( + text( + "SELECT learned FROM plg_antispam_verdicts " + "WHERE message_id = :m AND user_id = :u" + ), + {"m": str(message_id), "u": str(user_id)}, + ) + ).first() + return None if row is None else row[0] + + +async def _remember_learned( + user_id: uuid.UUID, message_id: uuid.UUID, *, verdict: str, action: str +) -> None: + """Record the teaching. Upsert, because the message may never have been + classified by us — provider-filed spam the user drags back has no verdict row.""" + async with _SDK.session(user_id) as s: + await s.execute( + text( + "INSERT INTO plg_antispam_verdicts " + " (message_id, user_id, score, symbols, action, learned) " + "VALUES (:m, :u, 0, '[]'::jsonb, :a, :l) " + "ON CONFLICT (message_id) DO UPDATE SET learned = EXCLUDED.learned" + ), + {"m": str(message_id), "u": str(user_id), "a": action, "l": verdict}, + ) + await s.commit() + + +async def on_message_changed(payload: dict[str, Any]) -> None: + """Passive learning from where the user files things (SPEC §C.5). + + Moving mail into Spam says "this is spam" more honestly than any button, + because the user is doing it for their own reasons rather than to train a + filter. Moving it back out of Spam says the opposite. + + Only those two transitions teach. Spam -> Trash is a user tidying up, not a + verdict; Inbox -> Archive says nothing about spam at all. + """ + uid_raw, mid_raw = payload.get("user_id"), payload.get("message_id") + if not isinstance(uid_raw, str) or not isinstance(mid_raw, str): + return + now, before = payload.get("special_use"), payload.get("previous_special_use") + if now == before: + return + + if now == "junk": + verdict, action = "spam", "moved" + elif before == "junk" and now == "inbox": + verdict, action = "ham", "none" + else: + return + + user_id, message_id = uuid.UUID(uid_raw), uuid.UUID(mid_raw) + + # Teaching the same message the same thing twice skews the corpus, and users + # do move things back and forth. + if await _already_learned(user_id, message_id) == verdict: + return + + settings = await policy.load_settings(user_id) + env = await _SDK.messages.envelope(user_id, message_id) + if env is None or not settings.covers(str(env.get("account_id") or "")): + return + + raw = await _SDK.messages.raw(user_id, message_id) + if raw is None: + return + if not await rspamd.learn(raw, user_key=str(user_id), spam=(verdict == "spam")): + # Engine down: leave `learned` unset so a later move can teach it again. + return + + await _remember_learned(user_id, message_id, verdict=verdict, action=action) + await _SDK.publish( + "plugin.antispam.learned", + {"user_id": str(user_id), "message_id": str(message_id), "verdict": verdict}, + ) + + +async def on_account_state_changed(payload: dict[str, Any]) -> None: + """When an account finishes its initial import, warm Bayes from what it holds. + + This is the constructive use of the backfill that classification deliberately + skips (§C.4): the same history we refuse to act on is exactly what we want to + learn from. + """ + if payload.get("state") != "live": + return + user_id, account_id = payload.get("user_id"), payload.get("account_id") + if not isinstance(user_id, str) or not isinstance(account_id, str): + return + from webmail_plugin_antispam.jobs import bootstrap_account + + await _SDK.enqueue(bootstrap_account, user_id=user_id, account_id=account_id) diff --git a/plugins/webmail_plugin_antispam/webmail_plugin_antispam/jobs.py b/plugins/webmail_plugin_antispam/webmail_plugin_antispam/jobs.py new file mode 100644 index 0000000..52b1455 --- /dev/null +++ b/plugins/webmail_plugin_antispam/webmail_plugin_antispam/jobs.py @@ -0,0 +1,112 @@ +"""Provider-Junk bootstrap (SPEC §C.5). + +A per-user Bayes classifier knows nothing on day one, which is exactly when the +filter is asked to judge the most mail. But the user already owns a labelled +corpus: whatever their provider filed as Junk is spam they agreed to lose, and +old Inbox mail they never moved is ham they agreed to keep. Feeding both gives +Bayes a warm start without asking the user to train anything. + +Deliberately unhurried. This is a background nicety, not a deadline: it yields +between messages so it cannot monopolise the worker or the engine, and it gives +up quietly on anything that looks wrong rather than retrying into a wall. +""" + +from __future__ import annotations + +import asyncio +import logging +import uuid +from typing import Any + +from app.plugins.sdk import sdk +from webmail_plugin_antispam import policy, rspamd + +logger = logging.getLogger("plg.antispam") + +_SDK = sdk("antispam") + +# §C.5: up to 500 of each. Enough to shape a classifier, small enough that the +# whole pass is minutes rather than hours. +SPAM_SAMPLE = 500 +HAM_SAMPLE = 500 +# Ham comes from mail old enough that the user has plainly not triaged it — a +# message from this morning is not yet evidence of anything. +HAM_MIN_AGE_DAYS = 30 +# A breath between messages: the engine is shared with live classification, and +# live mail must never queue behind a training backlog. +PAUSE_SECONDS = 0.05 + +_DONE_KEY = "_bootstrapped" + + +async def _teach(user_id: uuid.UUID, ids: list[uuid.UUID], *, spam: bool) -> int: + taught = 0 + for message_id in ids: + raw = await _SDK.messages.raw(user_id, message_id) + if raw is None: + continue + if not await rspamd.learn(raw, user_key=str(user_id), spam=spam): + # Engine gone. Stop rather than grind through hundreds of failures; + # the account stays unmarked, so a later run picks it up again. + return -taught + taught += 1 + await asyncio.sleep(PAUSE_SECONDS) + return taught + + +async def bootstrap_account( + ctx: dict[str, Any], *, user_id: str, account_id: str +) -> None: + """Teach Bayes from one account's existing mail. Safe to call more than once.""" + uid, acc = uuid.UUID(user_id), uuid.UUID(account_id) + + settings = await policy.load_settings(uid) + if not settings.learn_from_provider_junk or not settings.covers(account_id): + return + + done = await _SDK.kv(uid).get(_DONE_KEY, []) + if account_id in done: + return + + spam_ids = await _SDK.messages.sample( + uid, account_id=acc, special_use="junk", limit=SPAM_SAMPLE + ) + ham_ids = await _SDK.messages.sample( + uid, + account_id=acc, + special_use="inbox", + limit=HAM_SAMPLE, + older_than_days=HAM_MIN_AGE_DAYS, + ) + if not spam_ids and not ham_ids: + return + + spam_taught = await _teach(uid, spam_ids, spam=True) + if spam_taught < 0: + logger.warning( + "antispam bootstrap for %s stopped: engine unavailable", account_id + ) + return + ham_taught = await _teach(uid, ham_ids, spam=False) + if ham_taught < 0: + logger.warning( + "antispam bootstrap for %s stopped: engine unavailable", account_id + ) + return + + await _SDK.kv(uid).set(_DONE_KEY, [*done, account_id]) + logger.info( + "antispam bootstrap for %s: learned %d spam, %d ham", + account_id, + spam_taught, + ham_taught, + ) + await _SDK.publish( + "plugin.antispam.bootstrapped", + { + "user_id": user_id, + "account_id": account_id, + "spam": spam_taught, + "ham": ham_taught, + }, + )