M3 plugin platform: SDK + loader, RLS-isolated tables, migrations, jobs

Backend foundation for the plugin platform (SPEC §11), Phases 1-3.

Phase 1 — SDK + loader:
- app/plugins/types.py: frozen Plugin/UiManifest/Slot/Page/JobSpec contract.
- app/plugins/sdk.py: sdk(name) facade — the only app.* import allowed for plugins
  (scoped session, plugin.{name}.* publish, blob, outbox, settings, SSRF http_get, KV, enqueue).
- app/plugins/loader.py: entry-point discovery (webmail.plugins), hook forwarding
  (API + worker), router mounting at /api/plugins/{name}, ui-manifest aggregation.
- plugin_kv core table (migration 0013) + GET /api/plugins/ui-manifest.

Phase 2 — non-removable core plugins + DB-enforced privacy:
- builtin flag (calendar/contacts will be builtin=True; uninstall refuses them).
- RLS per-user isolation on plg_* tables: least-privilege webmail_plugin role
  (migration 0014), plugin_session(user_id) sets LOCAL ROLE + app.current_user_id,
  rls.enable_rls() (ENABLE+FORCE RLS + USING/WITH CHECK policy). Mail untouched.
- Per-plugin Alembic runner (shared env, dedicated version table) + install/uninstall
  --drop-data CLI; skeleton fixture proves install -> run -> uninstall --drop-data
  leaves zero residue (§12 M3 acceptance).

Phase 3 — plugin jobs (§11.2):
- JobSpec registered as arq function plg.{plugin}.{func} + cron (5-field parser);
  singleton=True guarded by a per-(plugin,user) Redis lock; wired onto the shared
  interactive worker.

Gates: ruff, mypy --strict (94 files + tests), full suite 326 passed. RLS isolation,
migration round-trip, and singleton lock covered by tests.

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

backend/app/plugins/__init__.py

@@ -0,0 +1,6 @@
+"""Plugin platform (SPEC §11).
+
+Public surface for plugin authors is :mod:`app.plugins.sdk`; the loader and
+core wiring live here but plugins must not import them directly (SPEC §11:
+"plugins may not import from ``app.*`` internals other than ``app.plugins.sdk``").
+"""

backend/app/plugins/alembic/env.py

@@ -0,0 +1,39 @@
+"""Shared Alembic environment for plugin migration branches (SPEC §11.1).
+
+Every plugin reuses this env; the runner (:mod:`app.plugins.migrate`) points
+``version_locations`` at the plugin's own revision directory and gives it a
+dedicated ``version_table`` (``alembic_version_plg_{name}``) so each plugin's
+migration history is independent of core and of every other plugin.
+"""
+
+from __future__ import annotations
+
+import asyncio
+
+from alembic import context
+from sqlalchemy.ext.asyncio import async_engine_from_config
+from sqlalchemy.pool import NullPool
+
+from app.config import get_settings
+
+config = context.config
+config.set_main_option("sqlalchemy.url", get_settings().database_url)
+_version_table = config.get_main_option("version_table") or "alembic_version"
+
+
+def _do_run_migrations(connection: object) -> None:
+    context.configure(connection=connection, version_table=_version_table)  # type: ignore[arg-type]
+    with context.begin_transaction():
+        context.run_migrations()
+
+
+async def _run_online() -> None:
+    engine = async_engine_from_config(
+        config.get_section(config.config_ini_section, {}), prefix="sqlalchemy.", poolclass=NullPool
+    )
+    async with engine.connect() as connection:
+        await connection.run_sync(_do_run_migrations)
+    await engine.dispose()
+
+
+asyncio.run(_run_online())

backend/app/plugins/alembic/script.py.mako

@@ -0,0 +1,26 @@
+"""${message}
+
+Revision ID: ${up_revision}
+Revises: ${down_revision | comma,n}
+Create Date: ${create_date}
+"""
+from __future__ import annotations
+
+from collections.abc import Sequence
+
+from alembic import op
+import sqlalchemy as sa
+${imports if imports else ""}
+
+revision: str = ${repr(up_revision)}
+down_revision: str | None = ${repr(down_revision)}
+branch_labels: str | Sequence[str] | None = ${repr(branch_labels)}
+depends_on: str | Sequence[str] | None = ${repr(depends_on)}
+
+
+def upgrade() -> None:
+    ${upgrades if upgrades else "pass"}
+
+
+def downgrade() -> None:
+    ${downgrades if downgrades else "pass"}

backend/app/plugins/cli.py

@@ -0,0 +1,58 @@
+"""Plugin management CLI (SPEC §11.1).
+
+    python -m app.plugins.cli list
+    python -m app.plugins.cli install <name>
+    python -m app.plugins.cli uninstall <name> [--drop-data]
+
+Builtin (core) plugins refuse uninstall.
+"""
+
+from __future__ import annotations
+
+import argparse
+
+from app.plugins import loader, migrate
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(
+        prog="python -m app.plugins.cli", description="Manage webmail plugins"
+    )
+    sub = parser.add_subparsers(dest="cmd", required=True)
+    sub.add_parser("list", help="list discovered plugins")
+    p_install = sub.add_parser("install", help="apply a plugin's migrations")
+    p_install.add_argument("name")
+    p_uninstall = sub.add_parser("uninstall", help="remove a plugin (refuses builtin)")
+    p_uninstall.add_argument("name")
+    p_uninstall.add_argument(
+        "--drop-data", action="store_true", help="also drop the plugin's tables"
+    )
+    args = parser.parse_args(argv)
+
+    if args.cmd == "list":
+        for plugin in loader.get_plugins():
+            kind = "builtin" if plugin.builtin else "removable"
+            migr = "migrations" if plugin.migrations else "no-migrations"
+            print(f"{plugin.name}\t{plugin.version}\t{kind}\t{migr}")
+        return 0
+
+    if args.cmd == "install":
+        migrate.install(args.name)
+        print(f"installed {args.name}")
+        return 0
+
+    if args.cmd == "uninstall":
+        try:
+            migrate.uninstall(args.name, drop_data=args.drop_data)
+        except loader.PluginRemovalError as exc:
+            print(f"error: {exc}")
+            return 1
+        suffix = " (data dropped)" if args.drop_data else " (data kept)"
+        print(f"uninstalled {args.name}{suffix}")
+        return 0
+
+    return 1
+
+
+if __name__ == "__main__":  # pragma: no cover
+    raise SystemExit(main())

backend/app/plugins/db.py

@@ -0,0 +1,44 @@
+"""RLS-enforced database access for plugins (SPEC §11.1 + privacy hardening).
+
+``plugin_session(user_id)`` yields a normal SQLAlchemy session whose transaction
+has been dropped to the least-privilege :data:`~app.plugins.rls.PLUGIN_ROLE` and
+bound to the per-user RLS context. Both are set with transaction-local scope
+(``SET LOCAL`` / ``set_config(..., is_local=true)``), so they auto-revert on
+commit/rollback and cannot leak across pooled connections.
+
+Because the role is non-superuser / non-owner / NOBYPASSRLS, the row-level policy
+from :mod:`app.plugins.rls` is actually enforced: a plugin query that forgets to
+filter ``user_id`` still returns only the current user's rows, and it cannot write
+rows for anyone else (policy ``WITH CHECK``).
+"""
+
+from __future__ import annotations
+
+import uuid
+from collections.abc import AsyncIterator
+from contextlib import asynccontextmanager
+
+from sqlalchemy import text
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.db import get_sessionmaker
+from app.plugins.rls import PLUGIN_ROLE
+
+
+@asynccontextmanager
+async def plugin_session(user_id: uuid.UUID) -> AsyncIterator[AsyncSession]:
+    """A transactional session scoped to ``user_id`` under RLS (SPEC §11.1)."""
+    async with get_sessionmaker()() as session:
+        try:
+            # Bind the RLS user context first (while still the privileged role), then
+            # drop into the restricted role for the rest of the transaction.
+            await session.execute(
+                text("SELECT set_config('app.current_user_id', :uid, true)"),
+                {"uid": str(user_id)},
+            )
+            await session.execute(text(f"SET LOCAL ROLE {PLUGIN_ROLE}"))
+            yield session
+            await session.commit()
+        except Exception:
+            await session.rollback()
+            raise

backend/app/plugins/jobs.py

@@ -0,0 +1,130 @@
+"""Plugin background jobs on the shared arq worker (SPEC §11.2).
+
+Each plugin ``JobSpec`` is registered as an arq function under a namespaced name
+``plg.{plugin}.{func}`` (enqueue on demand from hooks/routes via ``sdk.enqueue``),
+and, when it has a ``schedule``, also as an arq cron job. ``singleton=True`` wraps
+the job in a Redis NX lock keyed by (plugin, func, user) so at most one instance
+runs per user (or globally for scheduleds with no user). Jobs share the worker
+pool + retry semantics with core jobs and log under the ``plg`` logger.
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+from arq import cron
+from arq.cron import CronJob
+from arq.worker import Function, func
+
+from app.plugins import loader
+from app.plugins.types import JobSpec, Plugin
+from app.redis_client import get_redis
+
+logger = logging.getLogger("plg")
+
+# Safety expiry on the singleton lock so a crashed job can't wedge a user forever.
+# A job that runs longer than this could overlap; acceptable for M3.
+_LOCK_TTL = 3600
+
+# Standard 5-field cron order and each field's inclusive value bounds.
+_FIELD_ORDER = ("minute", "hour", "day", "month", "weekday")
+_FIELD_BOUNDS = {
+    "minute": (0, 59),
+    "hour": (0, 23),
+    "day": (1, 31),
+    "month": (1, 12),
+    "weekday": (0, 6),
+}
+
+
+def job_name(plugin_name: str, spec: JobSpec) -> str:
+    """The arq function name a plugin job is registered under."""
+    return f"plg.{plugin_name}.{spec.func.__name__}"
+
+
+def _lock_key(plugin_name: str, func_name: str, user: object) -> str:
+    return f"plglock:{plugin_name}:{func_name}:{user}"
+
+
+def _make_runner(plugin: Plugin, spec: JobSpec) -> tuple[Any, str]:
+    """Wrap a job func with the singleton lock (when requested) + namespaced name."""
+    func_name = spec.func.__name__
+    name = job_name(plugin.name, spec)
+
+    async def runner(ctx: dict[str, Any], *args: Any, **kwargs: Any) -> Any:
+        if not spec.singleton:
+            return await spec.func(ctx, *args, **kwargs)
+        user = kwargs.get("user_id", "global")
+        key = _lock_key(plugin.name, func_name, user)
+        acquired = await get_redis().set(key, b"1", nx=True, ex=_LOCK_TTL)
+        if not acquired:
+            logger.info("%s skipped: singleton already running (user=%s)", name, user)
+            return None
+        try:
+            return await spec.func(ctx, *args, **kwargs)
+        finally:
+            await get_redis().delete(key)
+
+    return runner, name
+
+
+def plugin_arq_functions() -> list[Function]:
+    """Every plugin job as an arq function (enqueue-on-demand), namespaced."""
+    funcs: list[Function] = []
+    for plugin in loader.get_plugins():
+        for spec in plugin.jobs:
+            runner, name = _make_runner(plugin, spec)
+            funcs.append(func(runner, name=name))
+    return funcs
+
+
+def _parse_field(token: str, lo: int, hi: int) -> set[int] | None:
+    """Parse one cron field to a value set (``*`` -> None = every)."""
+    if token == "*":
+        return None
+    values: set[int] = set()
+    for part in token.split(","):
+        if part.startswith("*/"):
+            values.update(range(lo, hi + 1, int(part[2:])))
+        elif "-" in part:
+            start, end = part.split("-", 1)
+            values.update(range(int(start), int(end) + 1))
+        else:
+            values.add(int(part))
+    return values
+
+
+def parse_cron(expr: str) -> dict[str, set[int] | None]:
+    """Parse a standard 5-field cron string into arq ``cron`` field kwargs."""
+    parts = expr.split()
+    if len(parts) != 5:
+        raise ValueError(f"cron expression must have 5 fields (min hour dom mon dow): {expr!r}")
+    return {
+        name: _parse_field(token, *_FIELD_BOUNDS[name])
+        for name, token in zip(_FIELD_ORDER, parts, strict=True)
+    }
+
+
+def plugin_cron_jobs() -> list[CronJob]:
+    """Cron-scheduled plugin jobs as arq ``CronJob``s (run on the shared worker)."""
+    jobs: list[CronJob] = []
+    for plugin in loader.get_plugins():
+        for spec in plugin.jobs:
+            if spec.schedule is None:
+                continue
+            runner, name = _make_runner(plugin, spec)
+            f = parse_cron(spec.schedule)
+            jobs.append(
+                cron(
+                    runner,
+                    name=f"{name}.cron",
+                    second=0,
+                    minute=f["minute"],
+                    hour=f["hour"],
+                    day=f["day"],
+                    month=f["month"],
+                    weekday=f["weekday"],
+                )
+            )
+    return jobs

backend/app/plugins/kv.py

@@ -0,0 +1,63 @@
+"""Namespaced key/value store handed to plugins (SPEC §11 ``plugin_kv``).
+
+A :class:`PluginKv` is bound to one (plugin, user_id) slice of the core
+``plugin_kv`` table; values are arbitrary JSON. This is the light-state store
+every plugin gets — heavy plugins use their own ``plg_``-prefixed tables (§11.1).
+"""
+
+from __future__ import annotations
+
+import uuid
+from typing import Any
+
+from sqlalchemy import delete, select
+from sqlalchemy.dialects.postgresql import insert as pg_insert
+
+from app.db import session_scope
+from app.models import PluginKv
+
+
+class PluginKvStore:
+    """CRUD over one plugin's per-user KV slice. Instantiate via ``sdk.kv(user_id)``."""
+
+    def __init__(self, plugin: str, user_id: uuid.UUID) -> None:
+        self._plugin = plugin
+        self._user_id = user_id
+
+    async def get(self, key: str, default: Any = None) -> Any:
+        async with session_scope() as session:
+            row = await session.get(PluginKv, (self._plugin, self._user_id, key))
+            return row.value if row is not None else default
+
+    async def set(self, key: str, value: Any) -> None:
+        """Upsert ``key`` -> ``value`` (JSON-serialisable)."""
+        stmt = (
+            pg_insert(PluginKv)
+            .values(plugin=self._plugin, user_id=self._user_id, key=key, value=value)
+            .on_conflict_do_update(
+                index_elements=[PluginKv.plugin, PluginKv.user_id, PluginKv.key],
+                set_={"value": value},
+            )
+        )
+        async with session_scope() as session:
+            await session.execute(stmt)
+
+    async def delete(self, key: str) -> None:
+        async with session_scope() as session:
+            await session.execute(
+                delete(PluginKv).where(
+                    PluginKv.plugin == self._plugin,
+                    PluginKv.user_id == self._user_id,
+                    PluginKv.key == key,
+                )
+            )
+
+    async def items(self) -> dict[str, Any]:
+        """Return every key/value in this plugin+user slice."""
+        async with session_scope() as session:
+            rows = await session.execute(
+                select(PluginKv.key, PluginKv.value).where(
+                    PluginKv.plugin == self._plugin, PluginKv.user_id == self._user_id
+                )
+            )
+            return dict(rows.all())  # type: ignore[arg-type]

backend/app/plugins/loader.py

@@ -0,0 +1,148 @@
+"""Plugin discovery + core wiring (SPEC §11).
+
+Discovers plugins via the ``webmail.plugins`` entry-point group, then:
+- mounts each plugin router at ``/api/plugins/{name}`` (API process),
+- forwards §8.3 events to plugin hooks (both API and worker processes),
+- aggregates UI manifests for ``GET /api/plugins/ui-manifest``.
+
+Plugin background jobs (§11.2) and migrations (§11.1) are wired in later M3 phases.
+"""
+
+from __future__ import annotations
+
+import logging
+from importlib.metadata import entry_points
+from typing import TYPE_CHECKING, Any
+
+from app.events import EventBus
+from app.plugins.types import Page, Plugin, Slot, UiManifest
+
+if TYPE_CHECKING:
+    from fastapi import FastAPI
+
+logger = logging.getLogger(__name__)
+
+ENTRYPOINT_GROUP = "webmail.plugins"
+
+_plugins: list[Plugin] | None = None
+# Plugin names whose hooks are already subscribed on a given bus, so repeated
+# create_app()/worker starts in one process don't double-register (tests, reload).
+_hooks_registered: set[tuple[int, str]] = set()
+
+
+def _coerce_plugin(obj: Any) -> Plugin:
+    """An entry point may resolve to a ``Plugin`` or a module exposing ``plugin``."""
+    if isinstance(obj, Plugin):
+        return obj
+    candidate = getattr(obj, "plugin", None)
+    if isinstance(candidate, Plugin):
+        return candidate
+    raise TypeError(f"entry point {obj!r} does not expose a Plugin (SPEC §11)")
+
+
+def discover_plugins() -> list[Plugin]:
+    """Load every plugin registered under the ``webmail.plugins`` group."""
+    found: list[Plugin] = []
+    seen: set[str] = set()
+    for ep in entry_points(group=ENTRYPOINT_GROUP):
+        try:
+            plugin = _coerce_plugin(ep.load())
+        except Exception:  # noqa: BLE001 - one bad plugin must not break the app
+            logger.exception("failed to load plugin entry point %s", ep.name)
+            continue
+        if plugin.name in seen:
+            logger.error("duplicate plugin name %r; skipping", plugin.name)
+            continue
+        seen.add(plugin.name)
+        found.append(plugin)
+        logger.info("loaded plugin %s v%s", plugin.name, plugin.version)
+    return found
+
+
+def get_plugins() -> list[Plugin]:
+    """Return the process-wide discovered plugin list (cached)."""
+    global _plugins
+    if _plugins is None:
+        _plugins = discover_plugins()
+    return _plugins
+
+
+def set_plugins_for_test(plugins: list[Plugin] | None) -> None:
+    """Override (or reset with ``None``) the plugin registry — tests only."""
+    global _plugins
+    _plugins = plugins
+    _hooks_registered.clear()
+
+
+def register_hooks(bus: EventBus) -> None:
+    """Subscribe each plugin's hooks to the bus (idempotent per process/bus).
+
+    Called in both the API and the worker so hooks fire wherever the event is
+    published (e.g. ``message.new`` originates in the worker)."""
+    for plugin in get_plugins():
+        marker = (id(bus), plugin.name)
+        if marker in _hooks_registered:
+            continue
+        for event, handler in plugin.hooks.items():
+            bus.subscribe(event, handler)
+        _hooks_registered.add(marker)
+
+
+def mount_routers(app: FastAPI) -> None:
+    """Mount each plugin's router at ``/api/plugins/{name}`` (SPEC §11)."""
+    for plugin in get_plugins():
+        if plugin.router is not None:
+            app.include_router(plugin.router, prefix=f"/api/plugins/{plugin.name}")
+
+
+def _slot_dict(slot: Slot) -> dict[str, str]:
+    return {"name": slot.name, "bundle": slot.bundle}
+
+
+def _page_dict(page: Page) -> dict[str, Any]:
+    return {
+        "route": page.route,
+        "bundle": page.bundle,
+        "nav": {"icon": page.nav.icon, "label": page.nav.label, "order": page.nav.order},
+    }
+
+
+def _manifest_dict(plugin: Plugin, manifest: UiManifest) -> dict[str, Any]:
+    return {
+        "name": plugin.name,
+        "version": plugin.version,
+        "builtin": plugin.builtin,  # core plugins the UI must not offer to remove
+        "slots": [_slot_dict(s) for s in manifest.slots],
+        "pages": [_page_dict(p) for p in manifest.pages],
+        "capabilities": dict(manifest.capabilities),
+    }
+
+
+def build_ui_manifest() -> dict[str, Any]:
+    """Aggregate every plugin's UI manifest for the frontend loader (SPEC §11.3)."""
+    plugins = [
+        _manifest_dict(p, p.ui_manifest) for p in get_plugins() if p.ui_manifest is not None
+    ]
+    return {"plugins": plugins}
+
+
+def get_plugin(name: str) -> Plugin | None:
+    """Return the loaded plugin named ``name``, or None."""
+    return next((p for p in get_plugins() if p.name == name), None)
+
+
+class PluginRemovalError(Exception):
+    """Raised when trying to uninstall a builtin (core) plugin (SPEC §11 / §12)."""
+
+
+def assert_removable(name: str) -> Plugin:
+    """Return the plugin, or raise if it's unknown or builtin (non-removable).
+
+    Core plugins (calendar, contacts) ship ``builtin=True`` and may never be
+    uninstalled by a user; the uninstall path calls this before touching data."""
+    plugin = get_plugin(name)
+    if plugin is None:
+        raise PluginRemovalError(f"unknown plugin {name!r}")
+    if plugin.builtin:
+        raise PluginRemovalError(f"plugin {name!r} is builtin and cannot be removed")
+    return plugin

backend/app/plugins/migrate.py

@@ -0,0 +1,98 @@
+"""Per-plugin Alembic migration runner + install/uninstall lifecycle (SPEC §11.1).
+
+Each plugin ships its own revision directory (``Plugin.migrations``) and runs
+against a dedicated version table, so its schema history is independent of core
+and of other plugins. ``install`` upgrades to head; ``uninstall --drop-data``
+runs the plugin's downgrade chain and removes the version table, leaving zero
+residue (the §12 M3 acceptance round-trip). Builtin (core) plugins refuse removal.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+from pathlib import Path
+
+from alembic import command
+from alembic.config import Config
+from sqlalchemy import text
+from sqlalchemy.ext.asyncio import create_async_engine
+from sqlalchemy.pool import NullPool
+
+from app.config import get_settings
+from app.plugins import loader
+from app.plugins.types import Plugin
+
+logger = logging.getLogger(__name__)
+
+_SHARED_ENV = str(Path(__file__).parent / "alembic")
+
+
+class PluginMigrationError(Exception):
+    """Raised when a plugin's migrations can't be run."""
+
+
+def _version_table(plugin: Plugin) -> str:
+    # plugin.name is validated [a-z][a-z0-9_]* at construction, so this is safe to
+    # interpolate into DDL.
+    return f"alembic_version_plg_{plugin.name}"
+
+
+def _config(plugin: Plugin) -> Config:
+    if not plugin.migrations:
+        raise PluginMigrationError(f"plugin {plugin.name!r} declares no migrations")
+    cfg = Config()
+    cfg.set_main_option("script_location", _SHARED_ENV)
+    cfg.set_main_option("version_locations", plugin.migrations)
+    cfg.set_main_option("path_separator", "os")  # split version_locations on os.pathsep only
+    cfg.set_main_option("version_table", _version_table(plugin))
+    return cfg
+
+
+def upgrade(plugin: Plugin, revision: str = "head") -> None:
+    """Run the plugin's migrations up to ``revision`` (default: head)."""
+    command.upgrade(_config(plugin), revision)
+
+
+def downgrade(plugin: Plugin, revision: str = "base") -> None:
+    """Run the plugin's downgrades to ``revision`` (default: base = drop all)."""
+    command.downgrade(_config(plugin), revision)
+
+
+async def _drop_version_table(plugin: Plugin) -> None:
+    # Fresh engine (not the shared global one) so this is safe to call from any loop.
+    engine = create_async_engine(get_settings().database_url, poolclass=NullPool)
+    try:
+        async with engine.begin() as conn:
+            await conn.execute(text(f"DROP TABLE IF EXISTS {_version_table(plugin)}"))
+    finally:
+        await engine.dispose()
+
+
+def install(name: str) -> None:
+    """Apply a plugin's migrations (create its ``plg_*`` tables). No-op if none."""
+    plugin = loader.get_plugin(name)
+    if plugin is None:
+        raise PluginMigrationError(f"unknown plugin {name!r}")
+    if plugin.migrations is None:
+        logger.info("plugin %s has no migrations; nothing to install", name)
+        return
+    logger.info("installing plugin %s: upgrade -> head", name)
+    upgrade(plugin, "head")
+
+
+def uninstall(name: str, *, drop_data: bool = False) -> None:
+    """Uninstall a plugin. Refuses builtin plugins.
+
+    Without ``drop_data`` the schema/data are kept (a reversible disable). With
+    ``drop_data`` the downgrade chain runs and the version table is removed, so
+    nothing remains in the database (SPEC §12 M3 acceptance)."""
+    plugin = loader.assert_removable(name)  # raises PluginRemovalError for builtin/unknown
+    if not drop_data:
+        logger.info("uninstalling plugin %s (data kept)", name)
+        return
+    if plugin.migrations is None:
+        return
+    logger.info("uninstalling plugin %s --drop-data: downgrade -> base", name)
+    downgrade(plugin, "base")
+    asyncio.run(_drop_version_table(plugin))

backend/app/plugins/rls.py

@@ -0,0 +1,46 @@
+"""Row-Level Security for plugin-owned tables (SPEC §11.1 + privacy hardening).
+
+Per-user isolation for ``plg_*`` tables is enforced by Postgres, not by remembering
+a ``WHERE user_id = ?`` in every query. Each user-data table gets RLS enabled +
+forced, with a policy comparing ``user_id`` to the transaction-local GUC
+``app.current_user_id``. Plugin code reaches these tables only through
+``sdk.session(user_id)`` (see :mod:`app.plugins.db`), which drops to the
+least-privilege :data:`PLUGIN_ROLE` and binds that GUC — so even a query that
+forgets to filter cannot see or write another user's rows.
+
+Plugin migrations call :func:`enable_rls` right after creating a user-data table.
+"""
+
+from __future__ import annotations
+
+# NOLOGIN, non-superuser, NOBYPASSRLS role plugin sessions SET LOCAL ROLE into
+# (created by core migration 0014). RLS is skipped for superusers / table owners /
+# BYPASSRLS roles, so plugin access must run as this restricted role to be enforced.
+PLUGIN_ROLE = "webmail_plugin"
+
+_USER_CTX = "current_setting('app.current_user_id', true)::uuid"
+
+
+def rls_ddl(table: str, *, user_column: str = "user_id") -> list[str]:
+    """SQL statements that lock ``table`` to the current user (SPEC §11.1).
+
+    Pure/testable: returns the DDL rather than executing it. ``FORCE`` makes even
+    the table owner subject to the policy; ``WITH CHECK`` blocks inserting/updating
+    rows for a different user."""
+    policy = f"{table}_user_isolation"
+    return [
+        f"ALTER TABLE {table} ENABLE ROW LEVEL SECURITY",
+        f"ALTER TABLE {table} FORCE ROW LEVEL SECURITY",
+        f"CREATE POLICY {policy} ON {table} "
+        f"USING ({user_column} = {_USER_CTX}) "
+        f"WITH CHECK ({user_column} = {_USER_CTX})",
+        f"GRANT SELECT, INSERT, UPDATE, DELETE ON {table} TO {PLUGIN_ROLE}",
+    ]
+
+
+def enable_rls(table: str, *, user_column: str = "user_id") -> None:
+    """Apply :func:`rls_ddl` via Alembic ``op`` — call from a plugin migration."""
+    from alembic import op
+
+    for statement in rls_ddl(table, user_column=user_column):
+        op.execute(statement)

backend/app/plugins/sdk.py

@@ -0,0 +1,116 @@
+"""The plugin SDK — the ONLY ``app.*`` surface a plugin may import (SPEC §11).
+
+Everything a plugin needs is reached through :func:`sdk`, which returns a
+:class:`PluginSdk` bound to the plugin's name: a plugin-scoped DB session, the
+event bus (publishing under ``plugin.{name}.*``), the blob store, outbox
+draft/send helpers, the settings accessor, an SSRF-guarded HTTP client
+(§10.6), and the namespaced KV store (§11).
+
+The frozen contract types (:class:`Plugin`, :class:`UiManifest`, :class:`JobSpec`,
+…) are re-exported here so plugin packages import only from ``app.plugins.sdk``.
+"""
+
+from __future__ import annotations
+
+import uuid
+from contextlib import AbstractAsyncContextManager
+from dataclasses import dataclass
+from typing import Any
+
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.accounts.ssrf import guarded_get
+from app.config import Settings, get_settings
+from app.events import get_event_bus
+from app.indexer import blobstore
+from app.outbox import service as outbox_service
+from app.plugins.db import plugin_session
+from app.plugins.kv import PluginKvStore
+from app.plugins.types import (
+    Hook,
+    JobFunc,
+    JobSpec,
+    Page,
+    PageNav,
+    Plugin,
+    Slot,
+    UiManifest,
+)
+from app.queue import enqueue_job
+
+__all__ = [
+    "Hook",
+    "JobFunc",
+    "JobSpec",
+    "Page",
+    "PageNav",
+    "Plugin",
+    "PluginSdk",
+    "Slot",
+    "UiManifest",
+    "blobstore",
+    "sdk",
+]
+
+
+@dataclass(frozen=True)
+class PluginSdk:
+    """A lightweight facade over shared services, scoped to one plugin.
+
+    Cheap to construct (wraps process-wide singletons + the plugin name), so
+    handlers/routes obtain it on demand via ``sdk("<name>")``."""
+
+    name: str
+
+    # --- config -------------------------------------------------------------
+    @property
+    def settings(self) -> Settings:
+        return get_settings()
+
+    # --- database -----------------------------------------------------------
+    def session(self, user_id: uuid.UUID) -> AbstractAsyncContextManager[AsyncSession]:
+        """An RLS-enforced transactional session scoped to ``user_id`` (SPEC §11.1).
+
+        Row-level security limits every query to this user's rows on ``plg_*``
+        tables — even a query that forgets ``WHERE user_id`` cannot leak or write
+        across users (see :mod:`app.plugins.db`)."""
+        return plugin_session(user_id)
+
+    def kv(self, user_id: uuid.UUID) -> PluginKvStore:
+        """The namespaced KV store for this plugin and ``user_id`` (SPEC §11)."""
+        return PluginKvStore(self.name, user_id)
+
+    # --- events -------------------------------------------------------------
+    async def publish(self, event: str, payload: dict[str, object]) -> None:
+        """Publish ``plugin.{name}.{event}`` on the bus (payload needs ``user_id``)."""
+        await get_event_bus().publish(f"plugin.{self.name}.{event}", payload)
+
+    # --- background jobs ----------------------------------------------------
+    async def enqueue(self, job_func: JobFunc, **kwargs: Any) -> None:
+        """Enqueue one of this plugin's jobs on the shared worker (SPEC §11.2).
+
+        ``job_func`` must be a job declared in this plugin's ``jobs=[...]``. Pass
+        ``user_id=`` for singleton jobs so the per-user lock applies."""
+        await enqueue_job(f"plg.{self.name}.{job_func.__name__}", **kwargs)
+
+    # --- blob store ---------------------------------------------------------
+    @property
+    def blob(self) -> object:
+        """The blob store (``store``/``load``/``exists``/``remove``), keyed by user."""
+        return blobstore
+
+    # --- outbox -------------------------------------------------------------
+    @property
+    def outbox(self) -> object:
+        """Draft/send helpers (``upsert_draft``, ``create_outbox``, …)."""
+        return outbox_service
+
+    # --- outbound HTTP ------------------------------------------------------
+    async def http_get(self, url: str) -> str | None:
+        """GET ``url`` through the §10.6 SSRF guard (blocks private IPs/oversize)."""
+        return await guarded_get(url)
+
+
+def sdk(name: str) -> PluginSdk:
+    """Return the SDK facade bound to plugin ``name`` (SPEC §11)."""
+    return PluginSdk(name)

backend/app/plugins/types.py

@@ -0,0 +1,106 @@
+# INTERFACE VERSION: 1
+"""Frozen plugin contract types (SPEC §11).
+
+A plugin is a Python package ``webmail_plugin_<name>`` exposing a module-level
+``plugin = Plugin(...)``, discovered via the ``webmail.plugins`` entry-point group.
+
+These dataclasses are the frozen surface (SPEC §11 "freeze in M3"): fields may be
+added with defaults, but existing names/meanings may not change without bumping
+the INTERFACE VERSION above.
+"""
+
+from __future__ import annotations
+
+import re
+from collections.abc import Awaitable, Callable
+from dataclasses import dataclass, field
+from typing import TYPE_CHECKING, Any
+
+if TYPE_CHECKING:
+    from fastapi import APIRouter
+
+# A hook handler receives an event payload (always carrying ``user_id``, SPEC §5.3).
+Hook = Callable[[dict[str, object]], Awaitable[None]]
+# A background job body: receives the arq context dict plus job-specific kwargs.
+JobFunc = Callable[..., Awaitable[Any]]
+
+# The frozen slot catalog (SPEC §11.3). Bundles are injected into these UI points.
+SLOT_NAMES: frozenset[str] = frozenset(
+    {"thread.toolbar", "message.actions", "message.banner", "sidebar.section", "settings.page"}
+)
+
+_NAME_RE = re.compile(r"[a-z][a-z0-9_]*")
+
+
+@dataclass(frozen=True)
+class Slot:
+    """A lazy-loaded ESM bundle injected into a named mail-UI slot (SPEC §11.3)."""
+
+    name: str
+    bundle: str  # path served by the plugin router, resolved by the frontend loader
+
+    def __post_init__(self) -> None:
+        if self.name not in SLOT_NAMES:
+            raise ValueError(f"unknown slot {self.name!r} (SPEC §11.3 slot catalog)")
+
+
+@dataclass(frozen=True)
+class PageNav:
+    """Nav entry for a page-plugin (SPEC §11.3)."""
+
+    icon: str
+    label: str
+    order: int = 100
+
+
+@dataclass(frozen=True)
+class Page:
+    """A full-content-area view mounted at ``route`` with a nav entry (SPEC §11.3)."""
+
+    route: str
+    nav: PageNav
+    bundle: str
+
+
+@dataclass(frozen=True)
+class UiManifest:
+    """A plugin's declared frontend surfaces (SPEC §11.3)."""
+
+    slots: list[Slot] = field(default_factory=list)
+    pages: list[Page] = field(default_factory=list)
+    capabilities: dict[str, str] = field(default_factory=dict)  # name -> endpoint path
+
+
+@dataclass(frozen=True)
+class JobSpec:
+    """A background task on the shared arq worker (SPEC §11.2).
+
+    ``schedule`` is a cron string for periodic jobs (None = enqueue on demand);
+    ``singleton`` caps it to one running instance per (plugin, user) via a Redis
+    lock. Registration onto the worker is done by the loader (M3 §11.2)."""
+
+    func: JobFunc
+    schedule: str | None = None
+    singleton: bool = False
+
+
+@dataclass(frozen=True)
+class Plugin:
+    """The object a plugin package exposes as module-level ``plugin`` (SPEC §11)."""
+
+    name: str
+    version: str
+    hooks: dict[str, Hook] = field(default_factory=dict)
+    router: APIRouter | None = None
+    ui_manifest: UiManifest | None = None
+    migrations: str | None = None  # path to the plugin's Alembic version branch (11.1)
+    jobs: list[JobSpec] = field(default_factory=list)
+    # Core plugins (calendar, contacts) ship builtin=True: always installed, never
+    # offered for removal in the UI, and the uninstall path refuses --drop-data.
+    builtin: bool = False
+
+    def __post_init__(self) -> None:
+        if not _NAME_RE.fullmatch(self.name):
+            raise ValueError(
+                f"invalid plugin name {self.name!r}: must match [a-z][a-z0-9_]* (SPEC §11)"
+            )