Plugin SDK: raw message bytes, per-message move, mailbox in the envelope

Three additive surfaces the antispam plugin (M7, Appendix C) needs and the
frozen §11 API did not have. None of them changes a core mail path.

`actions.move_to_special` moves ONE message into an account's folder by
special-use, wrapping the same optimistic-update + move_job path the core
endpoint uses. The existing `trash` moves a whole thread, which is right for
Delete and wrong for a spam verdict: one spam reply must not drag a legitimate
conversation into Junk. It returns False rather than raising when the account
has no such folder — an account whose server lacks a Junk folder must still
receive its mail.

`messages.raw` returns the stored RFC822 bytes. Spam scoring needs headers and
body; the SDK could only hand out one header at a time.

`envelope` now also carries account_id and mailbox_special_use. Appendix C must
classify only inbox arrivals, and neither the message.new payload nor the
envelope previously said where a message landed.

docs/open-questions.md gains OQ-6 (why these are additive rather than a spec
violation of "zero core diffs") and OQ-7 (why the engine is reached with plain
httpx instead of the SSRF-guarded sdk.http).

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_018SqJYt7jD7oUEkoY68sBZo

backend/app/plugins/actions.py

@@ -6,8 +6,9 @@ own message (the same app-level ``user_id`` scoping core uses — core tables ha
 no RLS). Flagging mirrors the core action path: update the DB, enqueue the IMAP
 push, and publish ``flags.changed`` so the UI updates.
 
-(Only flag actions are exposed for now; a ``move`` action would follow the same
-wrap-the-core-path pattern.)
+Actions exposed: flags, trash (whole thread), and ``move_to_special`` for a
+single message — the last added for the antispam plugin, which must file one
+spam message into Junk without dragging its thread along.
 """
 
 from __future__ import annotations
@@ -22,7 +23,7 @@ from sqlalchemy import select
 from app.db import session_scope
 from app.events import get_event_bus
 from app.indexer import blobstore
-from app.models import Message, Thread
+from app.models import Mailbox, Message, Thread
 
 
 class PluginMessages:
@@ -44,6 +45,26 @@ class PluginMessages:
             return None
         return message_from_bytes(raw).get(name)
 
+    async def raw(self, user_id: uuid.UUID, message_id: uuid.UUID) -> bytes | None:
+        """Return the stored RFC822 bytes of the user's message, or None.
+
+        For engines that need the whole message rather than one header — spam
+        scoring wants headers *and* body. None when the message isn't the user's,
+        has no blob, or the blob is unreadable; callers must treat that as "cannot
+        classify", never as an error.
+        """
+        async with session_scope() as session:
+            msg = await session.scalar(
+                select(Message).where(Message.id == message_id, Message.user_id == user_id)
+            )
+            if msg is None or not msg.blob_sha256:
+                return None
+            sha = msg.blob_sha256
+        try:
+            return blobstore.load(user_id, sha)
+        except FileNotFoundError:
+            return None
+
     async def threads_of(
         self, user_id: uuid.UUID, message_ids: Iterable[str]
     ) -> dict[str, str | None]:
@@ -69,14 +90,24 @@ class PluginMessages:
 
     async def envelope(self, user_id: uuid.UUID, message_id: uuid.UUID) -> dict[str, Any] | None:
         async with session_scope() as session:
-            msg = await session.scalar(
-                select(Message).where(Message.id == message_id, Message.user_id == user_id)
-            )
-            if msg is None:
+            row = (
+                await session.execute(
+                    select(Message, Mailbox.special_use)
+                    .join(Mailbox, Mailbox.id == Message.mailbox_id, isouter=True)
+                    .where(Message.id == message_id, Message.user_id == user_id)
+                )
+            ).first()
+            if row is None:
                 return None
+            msg, special_use = row
             sender = msg.from_addr[0] if msg.from_addr else {}
             return {
                 "message_id": str(msg.id),
+                "account_id": str(msg.account_id),
+                # Where it landed: "inbox", "junk", "sent", … or None for a folder
+                # with no SPECIAL-USE. A plugin that should only act on newly
+                # arrived mail checks this — `message.new` itself does not say.
+                "mailbox_special_use": special_use,
                 "subject": msg.subject or "",
                 "from_email": (sender.get("email") or "").lower(),
                 "from_name": sender.get("name") or "",
@@ -148,6 +179,45 @@ class PluginActions:
         )
         return True
 
+    async def move_to_special(
+        self, user_id: uuid.UUID, message_id: uuid.UUID, special_use: str
+    ) -> bool:
+        """Move ONE message into the account's folder with this ``special_use``
+        (``junk``, ``inbox``, …), on the same path as the UI's move: optimistic DB
+        update, then ``move_job`` performs the IMAP MOVE and reconciles the uid.
+
+        Per-message on purpose. ``trash`` moves a whole thread, which is right for
+        Delete and wrong for a spam verdict — one spam reply must not drag a
+        legitimate conversation into Junk.
+
+        Returns False when the message isn't the user's, the account has no such
+        folder, or it is already there. That is "could not act", not an error:
+        an account whose server has no Junk folder must still receive its mail.
+        """
+        from app.queue import enqueue_job
+
+        async with session_scope() as session:
+            msg = await session.scalar(
+                select(Message).where(Message.id == message_id, Message.user_id == user_id)
+            )
+            if msg is None:
+                return False
+            dst = await session.scalar(
+                select(Mailbox).where(
+                    Mailbox.user_id == user_id,
+                    Mailbox.account_id == msg.account_id,  # moves stay within an account
+                    Mailbox.special_use == special_use,
+                )
+            )
+            if dst is None or dst.id == msg.mailbox_id:
+                return False
+            dst_id, src_id = dst.id, msg.mailbox_id
+            msg.mailbox_id = dst_id  # optimistic, as the core endpoint does
+            await session.commit()
+
+        await enqueue_job("move_job", str(message_id), str(dst_id), str(src_id))
+        return True
+
     async def trash(self, user_id: uuid.UUID, message_id: uuid.UUID) -> bool:
         """Move the message's thread to Trash — the same path as the UI Delete
         (move_thread_to_folder_job handles the IMAP move, restore state, folder