chore(email): sharpen tool descriptions and rename email_list_inbox

- Rename email_list_inbox → email_list_folder (works on any folder, not just inbox)
- email_list_folder / email_search: distinguish browse-by-recency vs filter-by-query
- email_move / email_delete: clarify single-email scope; point to email_bulk_action for multiple
- email_bulk_action: clarify it handles multiple emails; point to move/delete for single
- email_create_folder: remove redundant "ask user" instruction (requires_approval handles it)
- Update tests to reflect renamed tool

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

CLAUDE.md

@@ -8,7 +8,8 @@ a plugin/integration system (Stage 2+) and an encrypted vault (Stage 3+).
 ## Current Status
 
 **Stage 3 — Memory Database: complete** (2026-05-18)
-Next: Stage 4 — Vault Encryption
+**Stage 6 — Daemon infrastructure: in progress** (`feat/daemon` branch)
+Next: Stage 4 — Vault Encryption (skipped for now); messaging bots (Stage 6 remainder)
 
 ## Project Roadmap
 
@@ -19,11 +20,11 @@ memory in `~/.pyra/memory/`, and hard security boundaries around the vault.
 ### Stage 2 — Plugin Framework ✅ COMPLETE
 - `src/pyra/plugins/` package: `base.py`, `loader.py`, `registry.py`, `executor.py`, `install.py`
 - `src/pyra/bundled_plugins/` — ships bundled plugin scripts with pyra
-- `src/pyra/daemon/` stub (CLI surface only)
+- `src/pyra/daemon/` stub (CLI surface only; daemon itself is Stage 6)
 - Config: `PluginConfig` + `DaemonConfig` added to `PyraConfig`
 - Bootstrap: `~/.pyra/plugins/` and `~/.pyra/logs/` created on startup
 - Chat session: AI tool-use loop (up to 10 iterations), approval gate, plugin slash commands
-- CLI: `pyra plugin list/install/enable/disable/setup`, `pyra daemon *` stubs
+- CLI: `pyra plugin list/install/enable/disable/setup`, `pyra daemon *` (stubs at Stage 2; implemented in Stage 6)
 
 ### Stage 3 — Memory Database ✅ COMPLETE
 - `src/pyra/memory/database.py`: SQLite + FTS5 via `memory_meta` + `memory_fts` tables
@@ -117,7 +118,11 @@ the vault under namespaced keys (`plugin:{name}:{key}`).
 | `plugins/executor.py` | Approval gate: scan args → prompt → execute → scan result → log |
 | `plugins/install.py` | Copies bundled plugins to `~/.pyra/plugins/` |
 | `bundled_plugins/` | Standalone plugin scripts shipped with pyra (installed on demand) |
-| `daemon/__init__.py` | Daemon package stub (implementation in Stage 2.4) |
+| `daemon/pid.py` | Atomic PID file — write, read, stale detection (POSIX + Windows), context manager |
+| `daemon/ipc.py` | IPC transport — Unix socket chmod 600 + UID-check (Linux/macOS) or TCP loopback + port file (Windows); newline-delimited JSON protocol |
+| `daemon/service.py` | OS service file generation + install/uninstall — launchd plist (macOS), systemd user unit (Linux), schtasks XML (Windows) |
+| `daemon/core.py` | asyncio event loop entry point, `PluginSupervisor` (per-task restart, max 10×, 5s back-off, reload), IPC command dispatch, signal handling |
+| `daemon/__init__.py` | Public daemon API exports |
 
 ### Runtime: `~/.pyra/`
 
@@ -493,6 +498,40 @@ Dataclass: `MemoryFile(name, path, category, size_bytes, modified)`
 | `list_bundled_plugins` | `(bundled_dir: Path) -> list[str]` | Names of all bundled plugins that have a `manifest.json` |
 | `read_manifest` | `(plugin_dir: Path) -> dict` | Reads `manifest.json`; returns `{}` if missing |
 
+#### `daemon.core`
+
+| Function | Signature | Purpose |
+|----------|-----------|---------|
+| `run_foreground` | `() -> None` | Entry point for `pyra daemon run` — loads config + plugins, writes PID file, runs asyncio loop |
+| `start_background` | `() -> None` | Spawns `pyra daemon run` as a detached subprocess (`start_new_session` on POSIX, `DETACHED_PROCESS` on Windows) |
+
+#### `daemon.pid`
+
+| Function | Signature | Purpose |
+|----------|-----------|---------|
+| `resolve_pid_path` | `(cfg_path: str) -> Path` | Expand `~` and resolve to absolute Path |
+
+#### `daemon.ipc`
+
+| Function | Signature | Purpose |
+|----------|-----------|---------|
+| `send_command` | `(address, msg, timeout=5.0) -> IpcResponse` | Synchronous CLI helper — `asyncio.run(IpcClient.send(...))` |
+| `get_socket_path` | `(cfg: str) -> Path` | Expand `~` and return Unix socket path |
+| `is_unix_socket` | `() -> bool` | True on Linux/macOS (`sys.platform != 'nt'`) |
+| `get_port_file_path` | `() -> Path` | Path to `~/.pyra/daemon.port` (Windows TCP port file) |
+
+#### `daemon.service`
+
+| Function | Signature | Purpose |
+|----------|-----------|---------|
+| `detect_platform` | `() -> Literal["macos","linux","windows"]` | Detect current OS |
+| `find_pyra_executable` | `() -> str` | `shutil.which("pyra")` → sibling fallback → `sys.executable -m pyra` |
+| `install_service` | `() -> None` | Generate + register OS service (reads config for log/pid paths) |
+| `uninstall_service` | `() -> None` | Deregister OS service |
+| `render_launchd_plist` | `(exe, log_file, pid_file) -> str` | macOS plist template |
+| `render_systemd_unit` | `(exe, log_file) -> str` | Linux systemd unit template |
+| `render_schtasks_xml` | `(exe) -> str` | Windows Task Scheduler XML template (write as UTF-16) |
+
 #### `chat.renderer` — rendering functions and shared `console`
 
 Import `console` from here; do not create a second `rich.Console()` in new code.
@@ -515,7 +554,7 @@ Import `console` from here; do not create a second `rich.Console()` in new code.
 | `GeneralConfig` | `config.schema` | `general:` block — `user_name`, `assistant_name` |
 | `ProviderConfig` | `config.schema` | `ai:` block — `provider_id`, `model`, `base_url` |
 | `PluginConfig` | `config.schema` | `plugins:` block — `enabled`, `require_approval`, `log_executions` |
-| `DaemonConfig` | `config.schema` | `daemon:` block |
+| `DaemonConfig` | `config.schema` | `daemon:` block — `enabled`, `socket_path`, `log_file`, `pid_file`, `ipc_port` |
 | `MemoryConfig` | `config.schema` | `memory:` block — `max_tokens_in_context`, `auto_load` |
 | `SecurityConfig` | `config.schema` | `security:` block — `injection_detection`, `log_injections` |
 | `ConversationHistory` | `chat.history` | Holds message list; builds API payload via `build_for_api()`; trims to token budget |
@@ -526,3 +565,5 @@ Import `console` from here; do not create a second `rich.Console()` in new code.
 | `PyraPlugin` | `plugins.base` | `@runtime_checkable` Protocol — the plugin interface |
 | `BasePlugin` | `plugins.base` | Concrete base with no-op defaults; plugins should inherit this |
 | `TaskPlanner` | `chat.planner` | Multi-step plan runner; `make_tool_handler()` returns the callable wired into the chat session; presents plan for user approval, executes each step via litellm with up to 5 tool-use iterations, verifies output before proceeding |
+| `PluginSupervisor` | `daemon.core` | asyncio supervisor — `add_task(name, factory)`, `start()`, `stop()`, `reload()`, `status()`; restarts crashed tasks up to 10× with 5s back-off |
+| `PidFile` | `daemon.pid` | `write()` (atomic), `read()`, `is_stale()`, `remove()`, context manager; `PidFileError(OSError)` raised when live PID already exists |

pyproject.toml

@@ -35,6 +35,12 @@ gdrive    = ["google-api-python-client>=2.120.0", "google-auth-oauthlib>=1.2.0"]
 onedrive  = ["msal>=1.28.0"]
 dropbox   = ["dropbox>=12.0.0"]
 daemon    = ["aiofiles>=23.0.0"]
+email     = [
+    "imap-tools>=1.7.0",
+    "google-api-python-client>=2.120.0",
+    "google-auth-oauthlib>=1.2.0",
+    "O365>=2.0.36",
+]
 all-plugins = [
     "caldav>=1.3.0", "webdav4>=0.9.0", "vobject>=0.9.6",
     "matrix-nio>=0.24.0", "aiofiles>=23.0.0",
@@ -44,6 +50,8 @@ all-plugins = [
     "google-api-python-client>=2.120.0", "google-auth-oauthlib>=1.2.0",
     "msal>=1.28.0",
     "dropbox>=12.0.0",
+    "imap-tools>=1.7.0",
+    "O365>=2.0.36",
 ]
 
 [project.scripts]

src/pyra/bundled_plugins/email/manifest.json

@@ -0,0 +1,12 @@
+{
+  "name": "email",
+  "version": "1.0.0",
+  "description": "Full email management — read, send, search, sort, and create filter rules. Supports Gmail, Microsoft 365, ProtonMail (Bridge), and any IMAP provider. Background monitoring pushes new-email summaries to your configured messaging bot.",
+  "author": "pyra",
+  "requires": [
+    "imap-tools>=1.7.0",
+    "google-api-python-client>=2.120.0",
+    "google-auth-oauthlib>=1.2.0",
+    "O365>=2.0.36"
+  ]
+}

src/pyra/daemon/__init__.py

@@ -1,6 +1,7 @@
 """Pyra background daemon package."""
 
 from pyra.daemon.core import PluginSupervisor, run_foreground, start_background
+from pyra.daemon.events import publish, subscribe_forever
 from pyra.daemon.ipc import IpcClient, IpcServer, send_command
 from pyra.daemon.pid import PidFile, PidFileError, resolve_pid_path
 from pyra.daemon.service import detect_platform, install_service, uninstall_service
@@ -9,6 +10,8 @@ __all__ = [
     "run_foreground",
     "start_background",
     "PluginSupervisor",
+    "publish",
+    "subscribe_forever",
     "IpcClient",
     "IpcServer",
     "send_command",

src/pyra/daemon/core.py

@@ -68,6 +68,22 @@ class PluginSupervisor:
         if tasks:
             await asyncio.gather(*tasks, return_exceptions=True)
 
+    async def reload(self) -> None:
+        """Cancel all running tasks and restart them with fresh coroutines."""
+        for record in self._records:
+            if record.task and not record.task.done():
+                record.task.cancel()
+                try:
+                    await record.task
+                except (asyncio.CancelledError, Exception):
+                    pass
+            record.restart_count = 0
+            record.last_error = None
+            record.task = asyncio.create_task(
+                self._supervise(record), name=record.name
+            )
+        _log.info("Reloaded %d plugin task(s).", len(self._records))
+
     def status(self) -> list[dict]:
         return [
             {
@@ -131,8 +147,8 @@ def _make_ipc_handler(supervisor: PluginSupervisor):
                 supervisor.request_shutdown()
                 return {"ok": True, "data": {}}
             case "reload":
-                _log.info("Reload requested via IPC.")
+                await supervisor.reload()
-                return {"ok": True, "data": {}}
+                return {"ok": True, "data": {"tasks_reloaded": len(supervisor._records)}}
             case _:
                 return {"ok": False, "data": {"error": f"unknown command: {cmd}"}}
 
@@ -144,6 +160,9 @@ def _make_ipc_handler(supervisor: PluginSupervisor):
 async def _run_daemon(cfg, supervisor: PluginSupervisor) -> None:
     from pyra.daemon.ipc import IpcServer, get_socket_path, is_unix_socket
 
+    # Install signal handlers now that the event loop is running.
+    _install_signal_handlers(supervisor)
+
     if is_unix_socket():
         address = get_socket_path(cfg.daemon.socket_path)
     else:
@@ -193,14 +212,17 @@ def run_foreground() -> None:
 
     _start_time = time.monotonic()
 
-    with pid_file:
+    try:
-        _install_signal_handlers(supervisor)
+        with pid_file:
-        _log.info("Pyra daemon starting (PID %d).", os.getpid())
+            _log.info("Pyra daemon starting (PID %d).", os.getpid())
-        try:
+            try:
-            asyncio.run(_run_daemon(cfg, supervisor))
+                asyncio.run(_run_daemon(cfg, supervisor))
-        except KeyboardInterrupt:
+            except KeyboardInterrupt:
-            pass
+                pass
-        _log.info("Pyra daemon stopped.")
+            _log.info("Pyra daemon stopped.")
+    except PidFileError as exc:
+        _log.error("Could not acquire PID file: %s", exc)
+        sys.exit(1)
 
 
 # ── Background spawn (pyra daemon start) ─────────────────────────────────────
@@ -266,7 +288,7 @@ def _install_signal_handlers(supervisor: PluginSupervisor) -> None:
         signal.signal(signal.SIGTERM, lambda *_: supervisor.request_shutdown())
         return
 
-    loop = asyncio.get_event_loop()
+    loop = asyncio.get_running_loop()
     loop.add_signal_handler(signal.SIGTERM, supervisor.request_shutdown)
     loop.add_signal_handler(signal.SIGHUP, supervisor.request_shutdown)
 

src/pyra/daemon/events.py

@@ -0,0 +1,46 @@
+"""Async notification bus for inter-plugin communication in the daemon.
+
+Plugins publish events to a shared asyncio.Queue; other plugins (e.g. messaging
+bots) consume them via subscribe_forever().  No direct plugin-to-plugin imports
+are needed — both sides just use this module.
+
+Event shape (by convention):
+    {"type": "new_email", "priority": int, "from": str, "subject": str,
+     "summary": str, "uid": str, "folder": str}
+    {"type": "new_message", "bot": str, "user_id": str, "text": str}
+"""
+from __future__ import annotations
+
+import asyncio
+from typing import Any, AsyncGenerator
+
+_queue: asyncio.Queue[dict[str, Any]] | None = None
+
+
+def get_queue() -> asyncio.Queue[dict[str, Any]]:
+    global _queue
+    if _queue is None:
+        _queue = asyncio.Queue(maxsize=200)
+    return _queue
+
+
+async def publish(event: dict[str, Any]) -> None:
+    """Emit an event. Drops silently if the queue is full (daemon is overloaded)."""
+    q = get_queue()
+    try:
+        q.put_nowait(event)
+    except asyncio.QueueFull:
+        pass
+
+
+async def subscribe_forever() -> AsyncGenerator[dict[str, Any], None]:
+    """Async generator — yields events as they arrive. Intended for daemon tasks."""
+    q = get_queue()
+    while True:
+        yield await q.get()
+
+
+def reset() -> None:
+    """Discard the current queue and create a fresh one. FOR TESTS ONLY."""
+    global _queue
+    _queue = None

src/pyra/plugins/registry.py

@@ -87,7 +87,10 @@ class PluginRegistry:
         factories: list[tuple[str, Callable[[], Coroutine]]] = []  # type: ignore[type-arg]
         for plugin in self._plugins.values():
             try:
-                n_tasks = len(plugin.daemon_tasks())
+                initial = plugin.daemon_tasks()
+                n_tasks = len(initial)
+                for c in initial:
+                    c.close()  # prevent "coroutine never awaited" RuntimeWarning
             except Exception:
                 continue
             for i in range(n_tasks):

tests/unit/test_daemon_core.py

@@ -0,0 +1,226 @@
+"""Unit tests for the daemon core — PluginSupervisor and IPC handler dispatch."""
+
+from __future__ import annotations
+
+import asyncio
+
+import pytest
+
+from pyra.daemon.core import PluginSupervisor, _make_ipc_handler
+
+
+# ── Helpers ───────────────────────────────────────────────────────────────────
+
+async def _drain(n: int = 20) -> None:
+    """Yield to the event loop n times to let scheduled tasks run."""
+    for _ in range(n):
+        await asyncio.sleep(0)
+
+
+# ── PluginSupervisor — lifecycle ──────────────────────────────────────────────
+
+async def test_supervisor_empty_starts_and_stops_cleanly() -> None:
+    sup = PluginSupervisor()
+    await sup.start()
+    await sup.stop()
+    assert sup.status() == []
+
+
+async def test_supervisor_runs_task_to_completion() -> None:
+    done = asyncio.Event()
+
+    async def task():
+        done.set()
+
+    sup = PluginSupervisor()
+    sup._RESTART_DELAY = 0.0
+    sup.add_task("t", task)
+    await sup.start()
+
+    await asyncio.wait_for(done.wait(), timeout=1.0)
+    await sup.stop()
+
+    assert sup._records[0].restart_count == 0
+    assert sup._records[0].last_error is None
+
+
+async def test_supervisor_restarts_crashed_task() -> None:
+    call_count = 0
+    completed = asyncio.Event()
+
+    async def flaky():
+        nonlocal call_count
+        call_count += 1
+        if call_count == 1:
+            raise RuntimeError("first call fails")
+        completed.set()
+
+    sup = PluginSupervisor()
+    sup._RESTART_DELAY = 0.0
+    sup.add_task("flaky", flaky)
+    await sup.start()
+
+    await asyncio.wait_for(completed.wait(), timeout=1.0)
+    await sup.stop()
+
+    assert sup._records[0].restart_count == 1
+    assert "RuntimeError" in (sup._records[0].last_error or "")
+
+
+async def test_supervisor_gives_up_after_max_restarts() -> None:
+    async def always_fails():
+        raise ValueError("always")
+
+    sup = PluginSupervisor()
+    sup._RESTART_DELAY = 0.0
+    sup._MAX_RESTARTS = 3
+    sup.add_task("failing", always_fails)
+    await sup.start()
+
+    # Allow enough iterations for 3 restarts + give-up.
+    for _ in range(200):
+        await asyncio.sleep(0)
+        if sup._records[0].task and sup._records[0].task.done():
+            break
+
+    await sup.stop()
+
+    assert sup._records[0].restart_count == 3
+    assert sup._records[0].last_error is not None
+
+
+# ── PluginSupervisor — status ─────────────────────────────────────────────────
+
+async def test_supervisor_status_returns_correct_shape() -> None:
+    sup = PluginSupervisor()
+    sup._RESTART_DELAY = 0.0
+
+    async def noop():
+        pass
+
+    sup.add_task("noop", noop)
+    await sup.start()
+    await _drain()
+
+    statuses = sup.status()
+    assert len(statuses) == 1
+    s = statuses[0]
+    assert set(s.keys()) == {"name", "alive", "restart_count", "last_error"}
+    assert s["name"] == "noop"
+    assert isinstance(s["alive"], bool)
+    assert isinstance(s["restart_count"], int)
+
+    await sup.stop()
+
+
+async def test_supervisor_status_empty_when_no_tasks() -> None:
+    sup = PluginSupervisor()
+    await sup.start()
+    assert sup.status() == []
+    await sup.stop()
+
+
+# ── PluginSupervisor — reload ─────────────────────────────────────────────────
+
+async def test_supervisor_reload_restarts_tasks() -> None:
+    call_count = 0
+
+    async def counting():
+        nonlocal call_count
+        call_count += 1
+        # Hang until cancelled so reload can cancel it.
+        await asyncio.sleep(10)
+
+    sup = PluginSupervisor()
+    sup._RESTART_DELAY = 0.0
+    sup.add_task("c", counting)
+    await sup.start()
+
+    await _drain()
+    assert call_count == 1
+
+    await sup.reload()
+    await _drain()
+
+    # After reload, the task should have been restarted (called a second time).
+    assert call_count == 2
+    assert sup._records[0].restart_count == 0  # reset by reload
+
+    await sup.stop()
+
+
+async def test_supervisor_reload_resets_restart_count() -> None:
+    call_count = 0
+
+    async def flaky():
+        nonlocal call_count
+        call_count += 1
+        if call_count <= 2:
+            raise RuntimeError("crash")
+        await asyncio.sleep(10)
+
+    sup = PluginSupervisor()
+    sup._RESTART_DELAY = 0.0
+    sup.add_task("f", flaky)
+    await sup.start()
+
+    # Wait for 2 crashes to accumulate.
+    for _ in range(200):
+        await asyncio.sleep(0)
+        if sup._records[0].restart_count >= 2:
+            break
+
+    assert sup._records[0].restart_count == 2
+
+    await sup.reload()
+    # Reload must reset the counter.
+    assert sup._records[0].restart_count == 0
+
+    await sup.stop()
+
+
+# ── IPC command handler ───────────────────────────────────────────────────────
+
+async def test_ipc_handler_ping() -> None:
+    sup = PluginSupervisor()
+    handler = _make_ipc_handler(sup)
+    resp = await handler({"cmd": "ping"})
+    assert resp["ok"] is True
+    assert resp["data"]["pong"] is True
+
+
+async def test_ipc_handler_status_shape() -> None:
+    sup = PluginSupervisor()
+    handler = _make_ipc_handler(sup)
+    resp = await handler({"cmd": "status"})
+    assert resp["ok"] is True
+    assert "uptime" in resp["data"]
+    assert "pid" in resp["data"]
+    assert "tasks" in resp["data"]
+    assert isinstance(resp["data"]["tasks"], list)
+
+
+async def test_ipc_handler_stop_signals_shutdown() -> None:
+    sup = PluginSupervisor()
+    handler = _make_ipc_handler(sup)
+    assert not sup._shutdown.is_set()
+    resp = await handler({"cmd": "stop"})
+    assert resp["ok"] is True
+    assert sup._shutdown.is_set()
+
+
+async def test_ipc_handler_reload_returns_task_count() -> None:
+    sup = PluginSupervisor()
+    handler = _make_ipc_handler(sup)
+    resp = await handler({"cmd": "reload"})
+    assert resp["ok"] is True
+    assert resp["data"]["tasks_reloaded"] == 0
+
+
+async def test_ipc_handler_unknown_command() -> None:
+    sup = PluginSupervisor()
+    handler = _make_ipc_handler(sup)
+    resp = await handler({"cmd": "bogus"})
+    assert resp["ok"] is False
+    assert "error" in resp["data"]
+    assert "bogus" in resp["data"]["error"]

tests/unit/test_email_plugin.py

@@ -0,0 +1,384 @@
+"""Unit tests for the email plugin — pure-logic helpers, no network calls."""
+from __future__ import annotations
+
+import json
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+# Import helpers directly — they depend only on stdlib
+from pyra.bundled_plugins.email.plugin import (
+    EmailMessage,
+    FilterRule,
+    _build_imap_search,
+    _decode_header,
+    _gmail_action_summary,
+    _gmail_criteria_summary,
+    _normalize_to_gmail,
+    _normalize_to_outlook,
+    _outlook_actions_summary,
+    _parse_raw_message,
+    _strip_html,
+)
+
+
+# ── _strip_html ────────────────────────────────────────────────────────────────
+
+def test_strip_html_removes_tags():
+    result = _strip_html("<p>Hello <b>world</b></p>")
+    assert "<" not in result
+    assert "Hello" in result
+    assert "world" in result
+
+
+def test_strip_html_decodes_entities():
+    result = _strip_html("&lt;script&gt; &amp; &quot;test&quot;")
+    assert "<script>" in result
+    assert "&" in result
+
+
+def test_strip_html_removes_style_and_script():
+    html = "<style>body{color:red}</style><script>alert(1)</script><p>Keep this</p>"
+    result = _strip_html(html)
+    assert "color" not in result
+    assert "alert" not in result
+    assert "Keep this" in result
+
+
+def test_strip_html_plain_text_unchanged():
+    result = _strip_html("Hello, world!")
+    assert result == "Hello, world!"
+
+
+# ── _decode_header ─────────────────────────────────────────────────────────────
+
+def test_decode_header_plain():
+    assert _decode_header("Hello") == "Hello"
+
+
+def test_decode_header_encoded():
+    # RFC 2047 base64-encoded UTF-8
+    encoded = "=?utf-8?b?SGVsbG8gV29ybGQ=?="
+    assert _decode_header(encoded) == "Hello World"
+
+
+def test_decode_header_empty():
+    assert _decode_header("") == ""
+
+
+# ── _parse_raw_message ─────────────────────────────────────────────────────────
+
+def _make_raw_email(
+    from_addr: str = "sender@example.com",
+    to_addr: str = "recipient@example.com",
+    subject: str = "Test Subject",
+    body: str = "Hello from test.",
+    message_id: str = "<test123@example.com>",
+) -> bytes:
+    return (
+        f"From: {from_addr}\r\n"
+        f"To: {to_addr}\r\n"
+        f"Subject: {subject}\r\n"
+        f"Date: Mon, 01 Jan 2024 12:00:00 +0000\r\n"
+        f"Message-ID: {message_id}\r\n"
+        f"MIME-Version: 1.0\r\n"
+        f"Content-Type: text/plain; charset=utf-8\r\n"
+        f"\r\n"
+        f"{body}\r\n"
+    ).encode()
+
+
+def test_parse_raw_message_basic_fields():
+    raw = _make_raw_email()
+    msg = _parse_raw_message(raw, uid="42", folder="INBOX", is_read=False)
+    assert msg.uid == "42"
+    assert msg.folder == "INBOX"
+    assert msg.from_addr == "sender@example.com"
+    assert "recipient@example.com" in msg.to_addrs
+    assert msg.subject == "Test Subject"
+    assert msg.body_text == "Hello from test."
+    assert msg.is_read is False
+    assert msg.has_attachments is False
+    assert msg.attachments == []
+    assert msg.message_id == "<test123@example.com>"
+
+
+def test_parse_raw_message_snippet_truncated():
+    long_body = "A" * 500
+    raw = _make_raw_email(body=long_body)
+    msg = _parse_raw_message(raw, uid="1", folder="INBOX", is_read=True)
+    assert len(msg.snippet) <= 200
+
+
+def test_parse_raw_message_body_truncated_at_8000():
+    huge_body = "x" * 10000
+    raw = _make_raw_email(body=huge_body)
+    msg = _parse_raw_message(raw, uid="1", folder="INBOX", is_read=False)
+    assert len(msg.body_text) <= 8030  # 8000 + "[...truncated]"
+    assert "truncated" in msg.body_text
+
+
+def test_parse_raw_message_html_stripped():
+    raw = _make_raw_email(body="<html><body><p>Plain text content</p></body></html>")
+    # Create HTML part manually
+    html_raw = (
+        "From: a@b.com\r\nTo: c@d.com\r\nSubject: Test\r\n"
+        "MIME-Version: 1.0\r\nContent-Type: text/html; charset=utf-8\r\n\r\n"
+        "<html><body><p>Plain text content</p></body></html>\r\n"
+    ).encode()
+    msg = _parse_raw_message(html_raw, uid="1", folder="INBOX", is_read=False)
+    assert "<" not in msg.body_text
+    assert "Plain text content" in msg.body_text
+
+
+# ── _build_imap_search ─────────────────────────────────────────────────────────
+
+def test_build_imap_search_unread():
+    from imap_tools import AND
+    criteria = _build_imap_search("unread invoices")
+    # Should produce an AND with seen=False
+    assert criteria is not None
+
+
+def test_build_imap_search_from():
+    criteria = _build_imap_search("from:boss@company.com")
+    assert criteria is not None
+
+
+def test_build_imap_search_subject():
+    criteria = _build_imap_search("subject: meeting notes")
+    assert criteria is not None
+
+
+def test_build_imap_search_fallback():
+    criteria = _build_imap_search("random search terms")
+    assert criteria is not None
+
+
+# ── Gmail rule normalisation ───────────────────────────────────────────────────
+
+def test_normalize_to_gmail_from_condition():
+    criteria, action = _normalize_to_gmail({"from": "boss@company.com"}, {"mark_read": True})
+    assert criteria.get("from") == "boss@company.com"
+    assert "UNREAD" in action.get("removeLabelIds", [])
+
+
+def test_normalize_to_gmail_move_to():
+    criteria, action = _normalize_to_gmail({"subject": "invoice"}, {"move_to": "Bills"})
+    assert criteria.get("subject") == "invoice"
+    assert "Bills" in action.get("addLabelIds", [])
+    assert "INBOX" in action.get("removeLabelIds", [])
+
+
+def test_normalize_to_gmail_mark_important():
+    _, action = _normalize_to_gmail({}, {"mark_important": True})
+    assert "IMPORTANT" in action.get("addLabelIds", [])
+
+
+def test_normalize_to_gmail_forward():
+    _, action = _normalize_to_gmail({}, {"forward_to": "archive@example.com"})
+    assert action.get("forward") == "archive@example.com"
+
+
+def test_gmail_criteria_summary_empty():
+    assert _gmail_criteria_summary({}) == "(any)"
+
+
+def test_gmail_criteria_summary_from():
+    assert "from=boss" in _gmail_criteria_summary({"from": "boss@company.com"})
+
+
+def test_gmail_action_summary_empty():
+    assert _gmail_action_summary({}) == "(no action)"
+
+
+# ── Outlook rule normalisation ─────────────────────────────────────────────────
+
+def test_normalize_to_outlook_from():
+    body = _normalize_to_outlook({"from": "a@b.com"}, {"move_to": "Work"})
+    from_addrs = body["conditions"].get("fromAddresses", [])
+    assert any("a@b.com" in str(a) for a in from_addrs)
+    assert body["actions"].get("moveToFolder") == "Work"
+
+
+def test_normalize_to_outlook_subject_contains():
+    body = _normalize_to_outlook({"subject": "invoice"}, {"mark_read": True})
+    assert "invoice" in body["conditions"].get("subjectContains", [])
+    assert body["actions"].get("markAsRead") is True
+
+
+def test_normalize_to_outlook_mark_important():
+    body = _normalize_to_outlook({}, {"mark_important": True})
+    assert body["actions"].get("markImportance") == "high"
+
+
+def test_normalize_to_outlook_delete():
+    body = _normalize_to_outlook({}, {"delete": True})
+    assert body["actions"].get("delete") is True
+
+
+# ── email_move folder-not-found path ──────────────────────────────────────────
+
+def test_email_move_returns_error_when_folder_missing(tmp_pyra_home):
+    from pyra.bundled_plugins.email.plugin import EmailPlugin
+
+    plugin = EmailPlugin()
+
+    # Inject a mock provider with known folders
+    mock_provider = MagicMock()
+    mock_provider.list_folders.return_value = ["INBOX", "Sent", "Trash"]
+    plugin._provider_instance = mock_provider
+
+    result = plugin._tool_move("uid123", "NonExistent", "INBOX")
+
+    assert "does not exist" in result.lower()
+    assert "email_create_folder" in result
+    mock_provider.move_message.assert_not_called()
+
+
+def test_email_move_succeeds_when_folder_exists(tmp_pyra_home):
+    from pyra.bundled_plugins.email.plugin import EmailPlugin
+
+    plugin = EmailPlugin()
+
+    mock_provider = MagicMock()
+    mock_provider.list_folders.return_value = ["INBOX", "Work", "Newsletters"]
+    plugin._provider_instance = mock_provider
+
+    result = plugin._tool_move("uid456", "Work", "INBOX")
+
+    assert "moved" in result.lower()
+    mock_provider.move_message.assert_called_once_with("uid456", "INBOX", "Work")
+
+
+# ── email_list_rules not-supported path ───────────────────────────────────────
+
+def test_email_list_rules_not_supported(tmp_pyra_home):
+    from pyra.bundled_plugins.email.plugin import EmailPlugin
+
+    plugin = EmailPlugin()
+    mock_provider = MagicMock()
+    mock_provider.list_rules.side_effect = NotImplementedError
+    plugin._provider_instance = mock_provider
+
+    result = plugin._tool_list_rules()
+    assert "not supported" in result.lower()
+
+
+# ── daemon/events integration ─────────────────────────────────────────────────
+
+@pytest.mark.asyncio
+async def test_events_publish_and_subscribe():
+    from pyra.daemon import events
+    events.reset()
+
+    await events.publish({"type": "new_email", "subject": "Test"})
+
+    received = []
+    async for event in events.subscribe_forever():
+        received.append(event)
+        break  # only need one
+
+    assert received[0]["type"] == "new_email"
+    assert received[0]["subject"] == "Test"
+    events.reset()
+
+
+@pytest.mark.asyncio
+async def test_events_queue_full_drops_silently():
+    from pyra.daemon import events
+    events.reset()
+
+    # Fill the queue
+    for i in range(200):
+        await events.publish({"n": i})
+
+    # This should not raise even though queue is full
+    await events.publish({"n": 999})
+
+    events.reset()
+
+
+# ── ProtonMail Bridge connectivity check (mocked) ─────────────────────────────
+
+def test_protonmail_setup_aborts_when_bridge_unreachable(tmp_pyra_home):
+    """_setup_protonmail should abort gracefully when Bridge is not running."""
+    import socket
+    from unittest.mock import patch, MagicMock
+    from pyra.bundled_plugins.email.plugin import EmailPlugin
+
+    plugin = EmailPlugin()
+    console = MagicMock()
+    vault_writer = MagicMock()
+
+    with patch("socket.create_connection", side_effect=ConnectionRefusedError):
+        plugin._setup_protonmail(console, vault_writer, "user@proton.me")
+
+    # Should not store any vault key if Bridge is unreachable
+    vault_writer.assert_not_called()
+
+
+# ── messaging bot recommendation ──────────────────────────────────────────────
+
+def test_check_messaging_bot_warns_when_no_bot(tmp_pyra_home):
+    from pyra.bundled_plugins.email.plugin import EmailPlugin
+    from unittest.mock import MagicMock, patch
+    from pyra.config.schema import PyraConfig, ProviderConfig, PluginConfig
+
+    plugin = EmailPlugin()
+    console = MagicMock()
+
+    cfg = PyraConfig(ai=ProviderConfig(provider_id="lmstudio", model="test"))
+    cfg.plugins = PluginConfig(enabled=[])  # no bots
+
+    with patch("pyra.bundled_plugins.email.plugin.EmailPlugin._load_settings", return_value={}), \
+         patch("pyra.config.manager.load_config", return_value=cfg):
+        plugin._check_messaging_bot(console)
+
+    # Should have printed something (Panel) recommending a bot
+    console.print.assert_called()
+
+
+# ── Tool list completeness ─────────────────────────────────────────────────────
+
+def test_plugin_exposes_16_tools():
+    from pyra.bundled_plugins.email.plugin import EmailPlugin
+    plugin = EmailPlugin()
+    # on_load with no-op vault reader
+    plugin.on_load(lambda _: None)
+    tools = plugin.tools()
+    tool_names = [t.name for t in tools]
+    assert len(tools) == 16
+
+    expected = {
+        "email_list_folder", "email_read", "email_send", "email_reply",
+        "email_forward", "email_move", "email_delete", "email_mark_read",
+        "email_search", "email_list_folders", "email_create_folder",
+        "email_inbox_summary", "email_list_rules", "email_create_rule",
+        "email_delete_rule", "email_bulk_action",
+    }
+    assert set(tool_names) == expected
+
+
+def test_write_tools_require_approval():
+    from pyra.bundled_plugins.email.plugin import EmailPlugin
+    plugin = EmailPlugin()
+    plugin.on_load(lambda _: None)
+    tools = {t.name: t for t in plugin.tools()}
+
+    for name in ["email_send", "email_reply", "email_forward", "email_move",
+                 "email_delete", "email_create_folder", "email_create_rule",
+                 "email_delete_rule", "email_bulk_action"]:
+        assert tools[name].requires_approval, f"{name} should require approval"
+
+
+def test_read_tools_no_approval():
+    from pyra.bundled_plugins.email.plugin import EmailPlugin
+    plugin = EmailPlugin()
+    plugin.on_load(lambda _: None)
+    tools = {t.name: t for t in plugin.tools()}
+
+    for name in ["email_list_folder", "email_read", "email_mark_read",
+                 "email_search", "email_list_folders", "email_inbox_summary",
+                 "email_list_rules"]:
+        assert not tools[name].requires_approval, f"{name} should NOT require approval"