Merge pull request #4983 from michaelxer/fix-setup-link-4926-20260628

fix(docs): correct broken backup-restore link in setup.md

docs/setup.md

@@ -472,4 +472,4 @@ All user data lives in `data/` (gitignored): `app.db` (sessions, messages, docum
 `memory.json`, `presets.json`, `uploads/`, `personal_docs/`, `chroma/`, `settings.json`.
 
 To back up or restore everything in `data/`, see the
-[Backup & Restore guide](docs/backup-restore.md).
+[Backup & Restore guide](backup-restore.md).

mcp_servers/email_server.py

@@ -58,6 +58,11 @@ def _uid_fetch_rows(data) -> list:
 _ACCOUNT_CACHE: dict = {}  # key = normalized account selector -> config dict
 _MCP_OWNER_ARG = "_odysseus_owner"
 _CURRENT_OWNER: ContextVar[str | None] = ContextVar("email_mcp_owner", default=None)
+_OWNER_ENV_KEYS = ("ODYSSEUS_MCP_EMAIL_OWNER", "ODYSSEUS_EMAIL_OWNER")
+_OWNER_SCOPE_ERROR = (
+    "Error: email MCP requires an authenticated owner or ODYSSEUS_MCP_EMAIL_OWNER "
+    "when owner-scoped email accounts are configured."
+)
 
 
 def _clean_header_value(value) -> str:
@@ -71,13 +76,29 @@ def _db_path() -> Path:
     return Path(APP_DB)
 
 
+def _configured_owner() -> str | None:
+    for key in _OWNER_ENV_KEYS:
+        owner = os.environ.get(key, "").strip()
+        if owner:
+            return owner
+    return None
+
+
 def _current_owner() -> str:
     owner = _CURRENT_OWNER.get()
-    return str(owner or "").strip()
+    return str(owner or _configured_owner() or "").strip()
+
+
+def _account_owner(row: dict) -> str:
+    return str(row.get("owner") or "").strip()
+
+
+def _has_owner_scoped_accounts(rows: list[dict]) -> bool:
+    return any(_account_owner(r) for r in rows)
 
 
 def _account_visible_to_owner(row: dict, owner: str) -> bool:
-    row_owner = str(row.get("owner") or "").strip()
+    row_owner = _account_owner(row)
     if row_owner == owner:
         return True
     if row_owner:
@@ -96,8 +117,7 @@ def _filter_accounts_for_owner(rows: list[dict]) -> list[dict]:
     if owner:
         return [r for r in rows if _account_visible_to_owner(r, owner)]
 
-    owners = {str(r.get("owner") or "").strip() for r in rows if str(r.get("owner") or "").strip()}
-    if len(owners) > 1:
+    if _has_owner_scoped_accounts(rows):
         return []
     return rows
 
@@ -106,8 +126,7 @@ def _mcp_owner_required(rows: list[dict] | None = None) -> bool:
     if _current_owner():
         return False
     rows = rows if rows is not None else _read_accounts_from_db()
-    owners = {str(r.get("owner") or "").strip() for r in rows if str(r.get("owner") or "").strip()}
-    return len(owners) > 1
+    return _has_owner_scoped_accounts(rows)
 
 
 def _load_email_writing_style() -> str:
@@ -274,6 +293,8 @@ def _load_config(account: str | None = None) -> dict:
     }
 
     raw_rows = _read_accounts_from_db()
+    if _mcp_owner_required(raw_rows):
+        raise ValueError(_OWNER_SCOPE_ERROR)
     rows = _filter_accounts_for_owner(raw_rows)
     row = _resolve_account_from_rows(rows, account)
     if _current_owner() and raw_rows and not rows:
@@ -1193,10 +1214,14 @@ def _send_email(to, subject, body, in_reply_to=None, references=None, cc=None, b
     UI. This closes the auto-send hole that let earlier models invent
     signatures and ship them to real recipients without confirmation."""
     if _read_agent_email_confirm_setting():
+        # Even confirmation-first sends must resolve the selected account now.
+        # Otherwise a caller could stage a pending draft against another
+        # owner's account selector before browser approval handles it.
+        cfg = _load_config(account)
         return _stash_agent_draft(
             to=to, subject=subject, body=body,
             in_reply_to=in_reply_to, references=references,
-            cc=cc, bcc=bcc, account=account,
+            cc=cc, bcc=bcc, account=cfg.get("account_id") or account,
         )
     send_account, cfg = _resolve_send_config(account)
     msg = EmailMessage()
@@ -2142,10 +2167,7 @@ async def call_tool(name: str, arguments: dict) -> list[TextContent]:
     try:
         all_db_accounts = _read_accounts_from_db()
         if _mcp_owner_required(all_db_accounts):
-            return [TextContent(
-                type="text",
-                text="Error: email MCP requires an authenticated owner when multiple email account owners are configured.",
-            )]
+            return [TextContent(type="text", text=_OWNER_SCOPE_ERROR)]
 
         if name == "list_email_accounts":
             rows = _filter_accounts_for_owner(all_db_accounts)

routes/email_helpers.py

@@ -349,7 +349,7 @@ def _assert_owns_account(account_id: str, owner: str) -> None:
             row = db.query(_EA).filter(_EA.id == account_id).first()
             if row is None:
                 raise HTTPException(404, "Account not found")
-            if row.owner and row.owner != owner:
+            if not _account_visible_to_owner(row, owner):
                 # Treat as 404 (not 403) so we don't leak existence.
                 raise HTTPException(404, "Account not found")
         finally:
@@ -362,6 +362,26 @@ def _assert_owns_account(account_id: str, owner: str) -> None:
         logger.error(f"Account-owner check failed: {e}")
         raise HTTPException(503, "Account check failed")
 
+
+def _account_visible_to_owner(row, owner: str) -> bool:
+    """Whether an authenticated `owner` may act on this EmailAccount row.
+
+    Mirrors the SQL predicate in `_get_email_config`'s
+    `_owner_or_matching_legacy_account`: a caller sees an account they own, or a
+    legacy owner-less account (owner NULL/"") only when its own mailbox
+    (`imap_user` / `from_address`) is the caller's. `email_accounts` is the one
+    owner-scoped table deliberately left out of the legacy-owner migration
+    backfill, so ownerless rows persist on multi-user deploys — making this the
+    gate that keeps one tenant off another's imported mailbox and its decrypted
+    IMAP/SMTP credentials."""
+    row_owner = getattr(row, "owner", None) or ""
+    if row_owner:
+        return row_owner == owner
+    return owner in {
+        getattr(row, "imap_user", None) or "",
+        getattr(row, "from_address", None) or "",
+    }
+
 def _q(name: str) -> str:
     """Quote an IMAP mailbox name. Defensive: escapes `\\` and `"` and wraps
     in double quotes so user-supplied folder names with spaces or quotes can't
@@ -903,12 +923,13 @@ def _get_email_config(account_id: str | None = None, owner: str = "") -> dict:
         try:
             if account_id:
                 row = db.query(_EA).filter(_EA.id == account_id, _EA.enabled == True).first()  # noqa: E712
-                # If the resolved row belongs to a different owner, treat as
+                # If the resolved row isn't visible to this owner, treat as
                 # not-found rather than silently serving it. This is a defense
                 # in depth — `require_owner` already calls `_assert_owns_account`
                 # for query-param account_ids, but other callers (cookbook
-                # rules, scheduled poller) may not.
-                if row is not None and owner and row.owner and row.owner != owner:
+                # rules, scheduled poller) may not. Ownerless legacy rows are
+                # only visible on a mailbox match, same as the fallback below.
+                if row is not None and owner and not _account_visible_to_owner(row, owner):
                     row = None
             # Fallback path — restrict to this owner's accounts so we don't
             # leak another user's default mailbox to an unconfigured user.

routes/task_routes.py

@@ -11,10 +11,14 @@ from fastapi import APIRouter, HTTPException, Request
 from pydantic import BaseModel
 
 from core.database import SessionLocal, ScheduledTask, TaskRun
-from core.middleware import INTERNAL_TOOL_USER
 from core.constants import internal_api_base
 from src.auth_helpers import get_current_user
 from src.constants import DATA_DIR, EMAIL_URGENCY_CACHE_DIR
+from src.task_action_policy import (
+    ADMIN_ONLY_TASK_ACTIONS,
+    is_admin_only_task_action,
+    owner_has_admin_task_privileges,
+)
 from src.task_scheduler import compute_next_run, HOUSEKEEPING_DEFAULTS
 from routes.prefs_routes import _load_for_user, _save_for_user
 
@@ -417,28 +421,18 @@ def setup_task_routes(task_scheduler) -> APIRouter:
                 db.close()
         return {"ok": True, "opened": True, "enabled": bool(prefs.get("tasks_enabled")), "resumed": resumed}
 
-    # Actions that execute shell/SSH commands — restricted to admins.
+    # Actions that execute shell/SSH commands or cross into admin-only
+    # Cookbook serving surfaces — restricted to admins.
     # Non-admin users cannot create tasks with these action types via the
     # API. See review CRIT-C.
-    _ADMIN_ONLY_ACTIONS = {"run_local", "run_script", "ssh_command"}
+    _ADMIN_ONLY_ACTIONS = ADMIN_ONLY_TASK_ACTIONS
 
     def _is_admin(user: str | None) -> bool:
-        if not user:
-            return False
-        # In-process tool-loopback marker — AuthMiddleware validated
-        # the internal token + loopback client before stamping this,
-        # so treat as admin-equivalent.
-        if user == INTERNAL_TOOL_USER:
-            return True
-        try:
-            from core.auth import AuthManager
-            auth = AuthManager()
-            if not auth.is_configured:
-                # Unconfigured single-user deploy: trust the local owner.
-                return True
-            return bool(auth.is_admin(user))
-        except Exception:
-            return False
+        return owner_has_admin_task_privileges(user)
+
+    def _require_admin_for_task_action(user: str | None, task_type: str | None, action: str | None) -> None:
+        if is_admin_only_task_action(task_type, action) and not _is_admin(user):
+            raise HTTPException(403, f"Action '{action}' requires admin privileges")
 
     def _validate_then_task_id(db, then_task_id: Optional[str], user: Optional[str], current_task_id: Optional[str] = None) -> Optional[str]:
         target_id = (then_task_id or "").strip()
@@ -466,8 +460,7 @@ def setup_task_routes(task_scheduler) -> APIRouter:
         # Block shell-executing action types for non-admins. action_run_local
         # uses subprocess.run(shell=True) and ssh_command / run_script run
         # arbitrary commands.
-        if req.task_type == "action" and req.action in _ADMIN_ONLY_ACTIONS and not _is_admin(user):
-            raise HTTPException(403, f"Action '{req.action}' requires admin privileges")
+        _require_admin_for_task_action(user, req.task_type, req.action)
         if req.trigger_type == "schedule" and not req.schedule:
             raise HTTPException(400, "Schedule is required for schedule-triggered tasks")
         if req.trigger_type == "schedule" and req.schedule == "cron" and not req.cron_expression:
@@ -681,6 +674,10 @@ def setup_task_routes(task_scheduler) -> APIRouter:
             if user and task.owner != user:
                 raise HTTPException(403, "Access denied")
 
+            next_task_type = req.task_type if req.task_type is not None else task.task_type
+            next_action = req.action if req.action is not None else task.action
+            _require_admin_for_task_action(user, next_task_type, next_action)
+
             if req.name is not None:
                 task.name = req.name
             if req.prompt is not None:
@@ -688,9 +685,6 @@ def setup_task_routes(task_scheduler) -> APIRouter:
             if req.task_type is not None:
                 task.task_type = req.task_type
             if req.action is not None:
-                # Same admin-only gate as create — see CRIT-C.
-                if req.action in _ADMIN_ONLY_ACTIONS and not _is_admin(user):
-                    raise HTTPException(403, f"Action '{req.action}' requires admin privileges")
                 task.action = req.action
             if req.output_target is not None:
                 task.output_target = req.output_target
@@ -807,6 +801,7 @@ def setup_task_routes(task_scheduler) -> APIRouter:
                 raise HTTPException(404, "Task not found")
             if user and task.owner != user:
                 raise HTTPException(403, "Access denied")
+            _require_admin_for_task_action(user, task.task_type, task.action)
             task.status = "active"
             if (task.trigger_type or "schedule") == "schedule":
                 task.next_run = compute_next_run(
@@ -869,6 +864,7 @@ def setup_task_routes(task_scheduler) -> APIRouter:
                 raise HTTPException(404, "Task not found")
             if user and task.owner != user:
                 raise HTTPException(403, "Access denied")
+            _require_admin_for_task_action(user, task.task_type, task.action)
         finally:
             db.close()
         started = await task_scheduler.run_task_now(task_id, force=force)
@@ -1058,6 +1054,14 @@ def setup_task_routes(task_scheduler) -> APIRouter:
             ).first()
             if not task:
                 raise HTTPException(404, "Not found")
+            if (
+                is_admin_only_task_action(task.task_type, task.action)
+                and not owner_has_admin_task_privileges(task.owner)
+            ):
+                task.status = "paused"
+                task.next_run = None
+                db.commit()
+                raise HTTPException(403, f"Action '{task.action}' requires admin privileges")
         finally:
             db.close()
         started = await task_scheduler.run_task_now(task_id)

src/task_action_policy.py

@@ -0,0 +1,47 @@
+"""Shared privilege policy for scheduled task actions."""
+
+from __future__ import annotations
+
+ADMIN_ONLY_TASK_ACTIONS = frozenset({
+    "run_local",
+    "run_script",
+    "ssh_command",
+    "cookbook_serve",
+})
+
+
+def is_admin_only_task_action(task_type: str | None, action: str | None) -> bool:
+    return (task_type or "llm") == "action" and (action or "") in ADMIN_ONLY_TASK_ACTIONS
+
+
+def owner_has_admin_task_privileges(owner: str | None) -> bool:
+    try:
+        from src.auth_helpers import _auth_disabled
+        if _auth_disabled():
+            return True
+    except Exception:
+        pass
+
+    if owner:
+        try:
+            from core.middleware import INTERNAL_TOOL_USER
+            if owner == INTERNAL_TOOL_USER:
+                return True
+        except Exception:
+            pass
+
+    try:
+        from core.auth import AuthManager
+        auth = AuthManager()
+        if not auth.is_configured:
+            return True
+        if not owner:
+            return False
+        return bool(auth.is_admin(owner))
+    except Exception:
+        pass
+
+    if not owner:
+        return False
+
+    return False

src/task_scheduler.py

@@ -10,6 +10,10 @@ from datetime import datetime, timedelta, timezone
 from typing import Any, Awaitable, Callable, Dict, Tuple
 
 from core.auth import RESERVED_USERNAMES
+from src.task_action_policy import (
+    is_admin_only_task_action,
+    owner_has_admin_task_privileges,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -821,6 +825,28 @@ class TaskScheduler:
                     db.commit()
                 return
 
+            if (
+                is_admin_only_task_action(task.task_type, task.action)
+                and not owner_has_admin_task_privileges(task.owner)
+            ):
+                msg = f"Action '{task.action}' requires admin privileges"
+                blocked = db.query(TaskRun).filter(TaskRun.id == run_id).first()
+                if blocked:
+                    blocked.status = "error"
+                    blocked.result = msg
+                    blocked.error = msg
+                    blocked.finished_at = _utcnow()
+                task.status = "paused"
+                task.next_run = None
+                task.last_run = _utcnow()
+                logger.warning(
+                    "Paused admin-only task %s for non-admin owner %r",
+                    task_id,
+                    task.owner,
+                )
+                db.commit()
+                return
+
             if gate_foreground:
                 waiting = db.query(TaskRun).filter(TaskRun.id == run_id).first()
                 if waiting and waiting.status == "queued":

tests/test_auth_regressions.py

@@ -334,16 +334,11 @@ def test_pop_notifications_owner_filtered():
 def test_admin_only_actions_set_contains_shell_runners():
     """The constant defining shell-executing action types must include
     the three risky entries. Catches accidental removal."""
-    from routes import task_routes
-    # `_ADMIN_ONLY_ACTIONS` is a closure constant. Easiest pin: re-read
-    # the source and check for the three risky entries + the admin gate
-    # wording.
-    src = open(task_routes.__file__, encoding="utf-8").read()
-    assert '"run_local"' in src
-    assert '"run_script"' in src
-    assert '"ssh_command"' in src
-    # And the gate is wired into both create and update paths.
-    assert "Action '" in src and "requires admin privileges" in src
+    from src.task_action_policy import ADMIN_ONLY_TASK_ACTIONS
+
+    assert "run_local" in ADMIN_ONLY_TASK_ACTIONS
+    assert "run_script" in ADMIN_ONLY_TASK_ACTIONS
+    assert "ssh_command" in ADMIN_ONLY_TASK_ACTIONS
 
 
 def test_task_create_notification_default_allows_action_specific_defaults():

tests/test_email_ownerless_account_owner_scope.py

@@ -0,0 +1,117 @@
+"""Cross-tenant access control for legacy owner-less email accounts.
+
+`email_accounts` is the one owner-scoped table left out of the legacy-owner
+migration backfill (core/database.py), so rows with owner NULL/"" persist on a
+multi-user deploy — e.g. an account configured while auth was disabled, or an
+imported legacy row. The HTTP route guards (`_assert_owns_account` and the
+explicit-account_id path in `_get_email_config`) must scope such rows to a
+mailbox match, exactly like the `_owner_or_matching_legacy_account` fallback and
+the MCP `_account_visible_to_owner` gate. Otherwise any authenticated user can
+read/send/update-credentials/delete another tenant's imported mailbox.
+"""
+
+from unittest import mock
+
+import pytest
+from fastapi import HTTPException
+
+
+def _make_db():
+    from sqlalchemy import create_engine
+    from sqlalchemy.orm import sessionmaker
+    from core.database import Base
+    engine = create_engine("sqlite:///:memory:", connect_args={"check_same_thread": False})
+    Base.metadata.create_all(engine)
+    Factory = sessionmaker(bind=engine)
+    return Factory
+
+
+def _make_account(Factory, account_id, owner, imap_user, from_address="", is_default=False):
+    from core.database import EmailAccount
+    db = Factory()
+    row = EmailAccount(
+        id=account_id,
+        owner=owner,
+        name="Test",
+        enabled=True,
+        is_default=is_default,
+        imap_host="imap.example.com",
+        imap_port=993,
+        imap_user=imap_user,
+        smtp_host="smtp.example.com",
+        smtp_port=587,
+        smtp_user=imap_user,
+        from_address=from_address or imap_user,
+    )
+    db.add(row)
+    db.commit()
+    db.close()
+
+
+def test_assert_owns_account_rejects_ownerless_account_for_other_tenant():
+    """The core regression: a legacy owner-less mailbox is NOT accessible to an
+    authenticated caller whose own mailbox does not match it."""
+    from routes.email_helpers import _assert_owns_account
+    Factory = _make_db()
+    # owner="" (created while auth was disabled); mailbox belongs to victim.
+    _make_account(Factory, "acct-legacy", owner="", imap_user="victim@corp.com")
+
+    with mock.patch("core.database.SessionLocal", Factory):
+        with pytest.raises(HTTPException) as exc:
+            _assert_owns_account("acct-legacy", "attacker")
+    assert exc.value.status_code == 404
+
+
+def test_assert_owns_account_allows_owned_account():
+    from routes.email_helpers import _assert_owns_account
+    Factory = _make_db()
+    _make_account(Factory, "acct-bob", owner="bob", imap_user="bob@corp.com")
+    with mock.patch("core.database.SessionLocal", Factory):
+        _assert_owns_account("acct-bob", "bob")  # no raise
+
+
+def test_assert_owns_account_allows_ownerless_account_on_mailbox_match():
+    """Legacy-claim path stays intact: the user whose mailbox matches an
+    owner-less account may still act on it (imap_user or from_address)."""
+    from routes.email_helpers import _assert_owns_account
+    Factory = _make_db()
+    _make_account(Factory, "acct-legacy", owner="", imap_user="alice@corp.com")
+    with mock.patch("core.database.SessionLocal", Factory):
+        _assert_owns_account("acct-legacy", "alice@corp.com")  # no raise
+
+
+def test_assert_owns_account_noop_for_single_user_mode():
+    """owner == "" (unconfigured / single-user) accepts any account, unchanged."""
+    from routes.email_helpers import _assert_owns_account
+    Factory = _make_db()
+    _make_account(Factory, "acct-legacy", owner="", imap_user="whoever@corp.com")
+    with mock.patch("core.database.SessionLocal", Factory):
+        _assert_owns_account("acct-legacy", "")  # no raise
+
+
+def test_get_email_config_does_not_resolve_ownerless_account_for_other_tenant(monkeypatch):
+    """`_get_email_config(account_id=..., owner=...)` must not serve an
+    owner-less account (and its decrypted creds) to a non-matching tenant."""
+    import routes.email_helpers as eh
+    Factory = _make_db()
+    _make_account(Factory, "acct-legacy", owner="", imap_user="victim@corp.com", is_default=True)
+
+    # Make the settings.json / env fallback empty and deterministic.
+    monkeypatch.setattr(eh, "_load_settings", lambda: {}, raising=False)
+    for var in ("IMAP_HOST", "SMTP_HOST", "IMAP_USER", "SMTP_USER"):
+        monkeypatch.delenv(var, raising=False)
+
+    with mock.patch("core.database.SessionLocal", Factory):
+        cfg = eh._get_email_config(account_id="acct-legacy", owner="attacker")
+
+    assert cfg.get("account_id") != "acct-legacy"
+
+
+def test_get_email_config_resolves_ownerless_account_on_mailbox_match():
+    """The mailbox owner still resolves their claimable legacy account by id."""
+    import routes.email_helpers as eh
+    Factory = _make_db()
+    _make_account(Factory, "acct-legacy", owner="", imap_user="alice@corp.com")
+    with mock.patch("core.database.SessionLocal", Factory):
+        cfg = eh._get_email_config(account_id="acct-legacy", owner="alice@corp.com")
+    assert cfg.get("account_id") == "acct-legacy"

tests/test_mcp_email_decode_header_spaces.py

@@ -16,7 +16,20 @@ pytest.importorskip("mcp")
 import mcp_servers.email_server as es
 
 
-def _init_accounts_db(path):
+@pytest.fixture(autouse=True)
+def _clear_mcp_email_owner_env(monkeypatch):
+    for key in es._OWNER_ENV_KEYS:
+        monkeypatch.delenv(key, raising=False)
+    es._ACCOUNT_CACHE.clear()
+    yield
+    es._ACCOUNT_CACHE.clear()
+
+
+def _init_accounts_db(path, rows=None):
+    rows = rows or [
+        ("acct-alice", "alice", "Alice Mail", 1, "alice@example.com", "alice@example.com", "alice@example.com", "2026-01-01"),
+        ("acct-bob", "bob", "Bob Mail", 1, "bob@example.com", "bob@example.com", "bob@example.com", "2026-01-02"),
+    ]
     conn = sqlite3.connect(path)
     conn.execute(
         """
@@ -50,10 +63,7 @@ def _init_accounts_db(path):
         VALUES (?, ?, ?, ?, 1, 'imap.example.com', 993, ?, '', 1,
                 'smtp.example.com', 465, 'ssl', ?, '', ?, ?)
         """,
-        [
-            ("acct-alice", "alice", "Alice Mail", 1, "alice@example.com", "alice@example.com", "alice@example.com", "2026-01-01"),
-            ("acct-bob", "bob", "Bob Mail", 1, "bob@example.com", "bob@example.com", "bob@example.com", "2026-01-02"),
-        ],
+        rows,
     )
     conn.commit()
     conn.close()
@@ -106,6 +116,37 @@ async def test_mcp_email_requires_owner_when_multiple_account_owners_exist(tmp_p
     assert "requires an authenticated owner" in out[0].text
 
 
+@pytest.mark.asyncio
+async def test_mcp_email_requires_owner_when_any_account_owner_exists(tmp_path, monkeypatch):
+    db_path = tmp_path / "app.db"
+    _init_accounts_db(
+        db_path,
+        rows=[
+            ("acct-alice", "alice", "Alice Mail", 1, "alice@example.com", "alice@example.com", "alice@example.com", "2026-01-01"),
+        ],
+    )
+    monkeypatch.setattr(es, "APP_DB", str(db_path))
+
+    out = await es.call_tool("list_email_accounts", {})
+
+    assert "requires an authenticated owner" in out[0].text
+    assert "Alice Mail" not in out[0].text
+
+
+@pytest.mark.asyncio
+async def test_mcp_email_configured_owner_filters_accounts(tmp_path, monkeypatch):
+    db_path = tmp_path / "app.db"
+    _init_accounts_db(db_path)
+    monkeypatch.setattr(es, "APP_DB", str(db_path))
+    monkeypatch.setenv("ODYSSEUS_MCP_EMAIL_OWNER", "alice")
+
+    out = await es.call_tool("list_email_accounts", {})
+    text = out[0].text
+
+    assert "Alice Mail" in text
+    assert "Bob Mail" not in text
+
+
 def test_mcp_email_scoped_owner_without_visible_account_skips_legacy_fallback(tmp_path, monkeypatch):
     db_path = tmp_path / "app.db"
     settings_path = tmp_path / "settings.json"
@@ -137,11 +178,81 @@ def test_mcp_email_scoped_owner_without_visible_account_skips_legacy_fallback(tm
         es._ACCOUNT_CACHE.clear()
 
 
+@pytest.mark.asyncio
+async def test_mcp_email_owner_cannot_use_other_owner_account_for_list_read_send_or_draft(tmp_path, monkeypatch):
+    import src.constants as constants
+
+    db_path = tmp_path / "app.db"
+    scheduled_path = tmp_path / "scheduled_emails.db"
+    _init_accounts_db(db_path)
+    monkeypatch.setattr(es, "APP_DB", str(db_path))
+    monkeypatch.setattr(constants, "SCHEDULED_EMAILS_DB", str(scheduled_path))
+    monkeypatch.setattr(es, "_read_agent_email_confirm_setting", lambda: True)
+
+    calls = [
+        ("list_emails", {"account": "Bob Mail"}),
+        ("read_email", {"uid": "1", "account": "Bob Mail"}),
+        ("send_email", {
+            "to": "recipient@example.com",
+            "subject": "Blocked",
+            "body": "Do not stage.",
+            "account": "Bob Mail",
+        }),
+        ("draft_email", {
+            "to": "recipient@example.com",
+            "subject": "Blocked",
+            "body": "Do not draft.",
+            "account": "Bob Mail",
+        }),
+    ]
+
+    for tool_name, args in calls:
+        out = await es.call_tool(tool_name, {**args, "_odysseus_owner": "alice"})
+        assert "Email account not found for selector" in out[0].text, tool_name
+        assert "Bob Mail" not in out[0].text or "Available accounts" in out[0].text
+
+    assert not scheduled_path.exists()
+
+
+@pytest.mark.asyncio
+async def test_mcp_send_email_stages_with_visible_owner_account_id(tmp_path, monkeypatch):
+    import src.constants as constants
+
+    app_db_path = tmp_path / "app.db"
+    scheduled_path = tmp_path / "scheduled_emails.db"
+    _init_accounts_db(app_db_path)
+    monkeypatch.setattr(es, "APP_DB", str(app_db_path))
+    monkeypatch.setattr(constants, "SCHEDULED_EMAILS_DB", str(scheduled_path))
+    monkeypatch.setattr(es, "_read_agent_email_confirm_setting", lambda: True)
+
+    out = await es.call_tool(
+        "send_email",
+        {
+            "to": "recipient@example.com",
+            "subject": "Review",
+            "body": "Please review.",
+            "account": "Alice Mail",
+            "_odysseus_owner": "alice",
+        },
+    )
+
+    assert "Draft staged for approval" in out[0].text
+    conn = sqlite3.connect(scheduled_path)
+    try:
+        row = conn.execute(
+            "SELECT owner, status, account_id FROM scheduled_emails"
+        ).fetchone()
+    finally:
+        conn.close()
+    assert row == ("alice", "agent_draft", "acct-alice")
+
+
 @pytest.mark.asyncio
 async def test_mcp_send_email_stages_owner_scoped_pending_draft(tmp_path, monkeypatch):
     import src.constants as constants
 
     db_path = tmp_path / "scheduled_emails.db"
+    monkeypatch.setattr(es, "APP_DB", str(tmp_path / "missing-app.db"))
     monkeypatch.setattr(constants, "SCHEDULED_EMAILS_DB", str(db_path))
     monkeypatch.setattr(es, "_read_agent_email_confirm_setting", lambda: True)
 

tests/test_task_cookbook_admin_gate.py

@@ -0,0 +1,350 @@
+"""Task CRUD must not let non-admins schedule Cookbook serve actions."""
+
+import sys
+from datetime import datetime, timedelta, timezone
+from types import SimpleNamespace
+from unittest.mock import MagicMock
+
+import pytest
+from fastapi import HTTPException
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+from sqlalchemy.pool import NullPool
+
+from tests.helpers.import_state import clear_fake_database_modules
+
+clear_fake_database_modules()
+
+import core.auth as core_auth
+import core.database as cdb
+import routes.task_routes as task_routes
+from core.database import ScheduledTask
+from core.database import TaskRun
+from src.task_scheduler import TaskScheduler
+
+_REAL_DATABASE_ATTRS = {
+    "Base": cdb.Base,
+    "SessionLocal": cdb.SessionLocal,
+    "ScheduledTask": ScheduledTask,
+    "TaskRun": TaskRun,
+}
+if hasattr(cdb, "engine"):
+    _REAL_DATABASE_ATTRS["engine"] = cdb.engine
+
+
+def _restore_module_binding(monkeypatch, name, module):
+    monkeypatch.setitem(sys.modules, name, module)
+    parent_name, _, attr = name.rpartition(".")
+    parent = sys.modules.get(parent_name)
+    if parent is not None:
+        monkeypatch.setattr(parent, attr, module, raising=False)
+
+
+@pytest.fixture()
+def task_db(monkeypatch, tmp_path):
+    _restore_module_binding(monkeypatch, "core.database", cdb)
+    for attr, value in _REAL_DATABASE_ATTRS.items():
+        monkeypatch.setattr(cdb, attr, value, raising=False)
+    engine = create_engine(
+        f"sqlite:///{tmp_path / 'tasks.db'}",
+        connect_args={"check_same_thread": False},
+        poolclass=NullPool,
+    )
+    cdb.Base.metadata.create_all(engine)
+    testing_session = sessionmaker(bind=engine, autoflush=False, autocommit=False)
+    monkeypatch.setattr(task_routes, "SessionLocal", testing_session)
+    monkeypatch.setattr(cdb, "SessionLocal", testing_session)
+    return testing_session
+
+
+@pytest.fixture()
+def configured_auth(monkeypatch):
+    _restore_module_binding(monkeypatch, "core.auth", core_auth)
+    monkeypatch.setenv("AUTH_ENABLED", "true")
+
+    class FakeAuthManager:
+        is_configured = True
+
+        def is_admin(self, user):
+            return user == "admin"
+
+    monkeypatch.setattr(core_auth, "AuthManager", FakeAuthManager)
+
+
+@pytest.fixture()
+def builtin_action_info(monkeypatch):
+    mod = sys.modules.get("src.builtin_actions")
+    if mod is None:
+        import src.builtin_actions as mod
+    monkeypatch.setattr(
+        mod,
+        "BUILTIN_ACTION_INFO",
+        {
+            "summarize_emails": "Summarize emails",
+            "cookbook_serve": "Serve Cookbook model",
+        },
+        raising=False,
+    )
+
+
+def _req(user):
+    return SimpleNamespace(state=SimpleNamespace(current_user=user))
+
+
+def _endpoint(method, path):
+    router = task_routes.setup_task_routes(MagicMock())
+    for route in router.routes:
+        if getattr(route, "path", None) == path and method in getattr(route, "methods", set()):
+            return route.endpoint
+    raise RuntimeError(f"{method} {path} not found")
+
+
+def _cookbook_create_req():
+    return task_routes.TaskCreate(
+        name="Serve test model",
+        prompt="{}",
+        task_type="action",
+        action="cookbook_serve",
+        trigger_type="webhook",
+    )
+
+
+def _seed_action_task(
+    session_factory,
+    task_id,
+    owner,
+    action="summarize_emails",
+    *,
+    task_type="action",
+    webhook_token=None,
+    next_run=None,
+):
+    db = session_factory()
+    try:
+        task = ScheduledTask(
+            id=task_id,
+            owner=owner,
+            name=task_id,
+            prompt="{}",
+            task_type=task_type,
+            action=action,
+            trigger_type="webhook",
+            status="active",
+            output_target="session",
+            webhook_token=webhook_token,
+            next_run=next_run,
+        )
+        db.add(task)
+        db.commit()
+    finally:
+        db.close()
+
+
+@pytest.mark.asyncio
+async def test_non_admin_cannot_create_cookbook_serve_task(task_db, configured_auth):
+    create_task = _endpoint("POST", "/api/tasks")
+
+    with pytest.raises(HTTPException) as exc:
+        await create_task(_req("alice"), _cookbook_create_req())
+
+    assert exc.value.status_code == 403
+    db = task_db()
+    try:
+        assert db.query(ScheduledTask).count() == 0
+    finally:
+        db.close()
+
+
+@pytest.mark.asyncio
+async def test_non_admin_cannot_update_task_to_cookbook_serve(task_db, configured_auth):
+    _seed_action_task(task_db, "alice-task", "alice")
+    update_task = _endpoint("PUT", "/api/tasks/{task_id}")
+
+    with pytest.raises(HTTPException) as exc:
+        await update_task(
+            _req("alice"),
+            "alice-task",
+            task_routes.TaskUpdate(action="cookbook_serve"),
+        )
+
+    assert exc.value.status_code == 403
+    db = task_db()
+    try:
+        task = db.query(ScheduledTask).filter(ScheduledTask.id == "alice-task").first()
+        assert task.action == "summarize_emails"
+    finally:
+        db.close()
+
+
+@pytest.mark.asyncio
+async def test_non_admin_cannot_update_task_type_to_activate_existing_cookbook_serve(
+    task_db, configured_auth
+):
+    _seed_action_task(
+        task_db,
+        "alice-task",
+        "alice",
+        action="cookbook_serve",
+        task_type="llm",
+    )
+    update_task = _endpoint("PUT", "/api/tasks/{task_id}")
+
+    with pytest.raises(HTTPException) as exc:
+        await update_task(
+            _req("alice"),
+            "alice-task",
+            task_routes.TaskUpdate(task_type="action"),
+        )
+
+    assert exc.value.status_code == 403
+    db = task_db()
+    try:
+        task = db.query(ScheduledTask).filter(ScheduledTask.id == "alice-task").first()
+        assert task.task_type == "llm"
+        assert task.action == "cookbook_serve"
+    finally:
+        db.close()
+
+
+@pytest.mark.asyncio
+async def test_non_admin_cannot_manually_run_existing_cookbook_serve_task(
+    task_db, configured_auth
+):
+    _seed_action_task(task_db, "alice-task", "alice", action="cookbook_serve")
+    scheduler = SimpleNamespace(run_task_now=MagicMock())
+    router = task_routes.setup_task_routes(scheduler)
+    for route in router.routes:
+        if getattr(route, "path", None) == "/api/tasks/{task_id}/run":
+            run_task = route.endpoint
+            break
+    else:
+        raise RuntimeError("POST /api/tasks/{task_id}/run not found")
+
+    with pytest.raises(HTTPException) as exc:
+        await run_task(_req("alice"), "alice-task")
+
+    assert exc.value.status_code == 403
+    scheduler.run_task_now.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_webhook_rejects_stale_non_admin_cookbook_serve_task(
+    task_db, configured_auth
+):
+    _seed_action_task(
+        task_db,
+        "alice-task",
+        "alice",
+        action="cookbook_serve",
+        webhook_token="secret",
+    )
+    webhook_trigger = _endpoint("POST", "/api/tasks/{task_id}/webhook/{token}")
+
+    with pytest.raises(HTTPException) as exc:
+        await webhook_trigger("alice-task", "secret")
+
+    assert exc.value.status_code == 403
+    db = task_db()
+    try:
+        task = db.query(ScheduledTask).filter(ScheduledTask.id == "alice-task").first()
+        assert task.status == "paused"
+        assert task.next_run is None
+    finally:
+        db.close()
+
+
+@pytest.mark.asyncio
+async def test_scheduler_pauses_stale_non_admin_cookbook_serve_task(
+    task_db, configured_auth
+):
+    due = datetime.now(timezone.utc).replace(tzinfo=None) - timedelta(minutes=1)
+    _seed_action_task(
+        task_db,
+        "alice-task",
+        "alice",
+        action="cookbook_serve",
+        next_run=due,
+    )
+    db = task_db()
+    try:
+        db.add(TaskRun(id="run-1", task_id="alice-task", status="queued"))
+        db.commit()
+    finally:
+        db.close()
+
+    scheduler = TaskScheduler.__new__(TaskScheduler)
+    scheduler._task_handles = {}
+    await scheduler._execute_task_locked(
+        "alice-task",
+        "run-1",
+        gate_foreground=False,
+        release_executing=False,
+    )
+
+    db = task_db()
+    try:
+        task = db.query(ScheduledTask).filter(ScheduledTask.id == "alice-task").first()
+        run = db.query(TaskRun).filter(TaskRun.id == "run-1").first()
+        assert task.status == "paused"
+        assert task.next_run is None
+        assert run.status == "error"
+        assert run.error == "Action 'cookbook_serve' requires admin privileges"
+    finally:
+        db.close()
+
+
+@pytest.mark.asyncio
+async def test_non_admin_action_metadata_hides_cookbook_serve(
+    configured_auth, builtin_action_info
+):
+    list_actions = _endpoint("GET", "/api/tasks/meta/actions")
+
+    out = await list_actions(_req("alice"))
+
+    action_names = {action["name"] for action in out["actions"]}
+    assert "cookbook_serve" not in action_names
+
+
+@pytest.mark.asyncio
+async def test_admin_can_create_cookbook_serve_task(task_db, configured_auth):
+    create_task = _endpoint("POST", "/api/tasks")
+
+    out = await create_task(_req("admin"), _cookbook_create_req())
+
+    assert out["action"] == "cookbook_serve"
+    db = task_db()
+    try:
+        task = db.query(ScheduledTask).filter(ScheduledTask.id == out["id"]).first()
+        assert task.owner == "admin"
+    finally:
+        db.close()
+
+
+@pytest.mark.asyncio
+async def test_admin_action_metadata_includes_cookbook_serve(
+    configured_auth, builtin_action_info
+):
+    list_actions = _endpoint("GET", "/api/tasks/meta/actions")
+
+    out = await list_actions(_req("admin"))
+
+    action_names = {action["name"] for action in out["actions"]}
+    assert "cookbook_serve" in action_names
+
+
+@pytest.mark.asyncio
+async def test_auth_disabled_single_user_can_create_cookbook_serve_task(
+    monkeypatch, task_db
+):
+    monkeypatch.setenv("AUTH_ENABLED", "false")
+    create_task = _endpoint("POST", "/api/tasks")
+
+    out = await create_task(_req(None), _cookbook_create_req())
+
+    assert out["action"] == "cookbook_serve"
+    db = task_db()
+    try:
+        task = db.query(ScheduledTask).filter(ScheduledTask.id == out["id"]).first()
+        assert task.owner is None
+    finally:
+        db.close()