Merge f64346dace into 20e7fc0164

core/database.py

@@ -5,7 +5,7 @@ from datetime import datetime, timezone
 from pathlib import Path
 from typing import Optional
 from urllib.parse import unquote, urlparse
-from sqlalchemy import event, create_engine, Column, String, Text, Boolean, DateTime, Integer, ForeignKey, JSON, Index, func, text
+from sqlalchemy import DDL, event, create_engine, Column, String, Text, Boolean, DateTime, Integer, ForeignKey, JSON, Index, func, inspect, text
 from sqlalchemy.engine import Engine, make_url
 from sqlalchemy.types import TypeDecorator
 from sqlalchemy.ext.declarative import declarative_base, declared_attr
@@ -430,6 +430,93 @@ class EmailAccount(TimestampMixin, Base):
     )
 
 
+class EmailAccountOwnerLock(Base):
+    """Durable per-owner mutex for email-account default mutations.
+
+    Row-locking databases serialize mutations by locking this row before they
+    inspect or stage EmailAccount changes.  SQLite uses ``BEGIN IMMEDIATE``
+    instead, because it ignores ``SELECT ... FOR UPDATE``; keeping the table in
+    the shared metadata still makes the non-SQLite path available without a
+    separate migration.  The empty key represents the normalized legacy /
+    unconfigured scope shared by ``owner IS NULL`` and ``owner = ''`` rows.
+    """
+    __tablename__ = "email_account_owner_locks"
+
+    owner_key = Column(String, primary_key=True)
+
+
+_EMAIL_ACCOUNT_DEFAULT_INDEX = "ux_email_accounts_one_default_per_owner"
+_EMAIL_ACCOUNT_DEFAULT_INDEX_DDL = {
+    "sqlite": (
+        f"CREATE UNIQUE INDEX IF NOT EXISTS {_EMAIL_ACCOUNT_DEFAULT_INDEX} "
+        "ON email_accounts (COALESCE(owner, '')) WHERE is_default = 1"
+    ),
+    "postgresql": (
+        f"CREATE UNIQUE INDEX IF NOT EXISTS {_EMAIL_ACCOUNT_DEFAULT_INDEX} "
+        "ON email_accounts ((COALESCE(owner, ''))) WHERE is_default IS TRUE"
+    ),
+}
+
+
+# SQLAlchemy cannot express one portable partial, functional index across the
+# two supported database families.  Register dialect-specific DDL so fresh
+# databases get the invariant as part of create_all(); the startup migration
+# below installs the same index on existing databases after normalizing legacy
+# duplicate rows.
+for _dialect_name, _index_ddl in _EMAIL_ACCOUNT_DEFAULT_INDEX_DDL.items():
+    event.listen(
+        EmailAccount.__table__,
+        "after_create",
+        DDL(_index_ddl).execute_if(dialect=_dialect_name),
+    )
+
+
+def lock_email_account_owner_mutations(db, *owners: str) -> None:
+    """Lock normalized email-account owner scopes in canonical order.
+
+    ``NULL`` and the empty string are one legacy/single-user owner partition,
+    matching the unique default-account index.  SQLite has only a database
+    writer reservation, while row-locking databases use durable mutex rows.
+    Sorting all requested owner keys keeps multi-owner operations such as user
+    rename from deadlocking with another mutation that requests the same keys
+    in the opposite order.
+    """
+    from sqlalchemy.exc import IntegrityError
+
+    owner_keys = sorted({owner or "" for owner in owners} or {""})
+    if db.get_bind().dialect.name == "sqlite":
+        db.execute(text("BEGIN IMMEDIATE"))
+        return
+
+    for owner_key in owner_keys:
+        lock_row = db.get(
+            EmailAccountOwnerLock,
+            owner_key,
+            with_for_update=True,
+        )
+        if lock_row is not None:
+            continue
+
+        inserted = False
+        try:
+            with db.begin_nested():
+                db.add(EmailAccountOwnerLock(owner_key=owner_key))
+                db.flush()
+                inserted = True
+        except IntegrityError:
+            # A competing transaction created the mutex row first.  Once its
+            # insert commits, lock that durable row before touching accounts.
+            pass
+
+        if not inserted:
+            (
+                db.query(EmailAccountOwnerLock)
+                .filter(EmailAccountOwnerLock.owner_key == owner_key)
+                .with_for_update()
+                .one()
+            )
+
+
 class ModelEndpoint(TimestampMixin, Base):
     """Admin-configured model endpoints. Models are auto-discovered via /v1/models."""
     __tablename__ = "model_endpoints"
@@ -1812,72 +1899,142 @@ class Integration(TimestampMixin, Base):
 
 
 
-def _migrate_seed_email_account():
-    """If email_accounts is empty and settings.json has legacy flat imap_host/smtp_host
-    keys, create a single default account from them so nothing breaks for users who
-    upgraded. Safe to run repeatedly — it short-circuits once any row exists."""
+def _migrate_email_account_default_invariant():
+    """Normalize legacy duplicates and install durable at-most-one enforcement.
+
+    Older databases only had a non-unique ``(owner, is_default)`` lookup index.
+    Keep the oldest default deterministically in each normalized owner scope,
+    then add the same partial functional unique index used for fresh schemas.
+    """
+    dialect_name = engine.dialect.name
+    index_ddl = _EMAIL_ACCOUNT_DEFAULT_INDEX_DDL.get(dialect_name)
+    if index_ddl is None:
+        logger.warning(
+            "Email-account default uniqueness is not available for database "
+            "dialect %s; mutations remain serialized but are not protected by "
+            "a database constraint",
+            dialect_name,
+        )
+        return
+
     try:
-        with engine.connect() as conn:
-            tables = [r[0] for r in conn.execute(text(
-                "SELECT name FROM sqlite_master WHERE type='table' AND name='email_accounts'"
-            ))]
-            if "email_accounts" not in tables:
-                return
-            existing = conn.execute(text("SELECT COUNT(*) FROM email_accounts")).scalar() or 0
-            if existing > 0:
+        with engine.begin() as conn:
+            if not inspect(conn).has_table(EmailAccount.__tablename__):
                 return
+            default_rows = conn.execute(text("""
+                SELECT id, owner
+                FROM email_accounts
+                WHERE is_default IS TRUE
+                ORDER BY
+                    COALESCE(owner, ''),
+                    CASE WHEN created_at IS NULL THEN 1 ELSE 0 END,
+                    created_at,
+                    id
+            """)).mappings()
+            seen_owner_keys = set()
+            duplicate_ids = []
+            for row in default_rows:
+                owner_key = row["owner"] or ""
+                if owner_key in seen_owner_keys:
+                    duplicate_ids.append(row["id"])
+                else:
+                    seen_owner_keys.add(owner_key)
 
-        import json as _json
-        import uuid as _uuid
-        from pathlib import Path
-        settings_file = Path(SETTINGS_FILE)
-        if not settings_file.exists():
-            return
-        try:
-            s = _json.loads(settings_file.read_text(encoding="utf-8"))
-        except Exception:
-            return
+            for account_id in duplicate_ids:
+                conn.execute(
+                    text("UPDATE email_accounts SET is_default = :value WHERE id = :id"),
+                    {"value": False, "id": account_id},
+                )
+            conn.execute(text(index_ddl))
 
-        imap_host = (s.get("imap_host") or "").strip()
-        smtp_host = (s.get("smtp_host") or "").strip()
-        if not imap_host and not smtp_host:
-            return  # nothing to migrate
+        if duplicate_ids:
+            logger.warning(
+                "Normalized %d duplicate default email account(s) before "
+                "installing %s",
+                len(duplicate_ids),
+                _EMAIL_ACCOUNT_DEFAULT_INDEX,
+            )
+    except Exception:
+        # Starting without the constraint would silently retain the race this
+        # migration is intended to close.  Fail startup so an operator sees and
+        # can repair an incompatible schema instead of accepting unsafe writes.
+        logger.exception("Failed to enforce the email-account default invariant")
+        raise
+
+
+def _migrate_seed_email_account():
+    """Atomically seed one legacy default account when no account exists.
+
+    Reading settings is intentionally done before taking the owner mutex.  The
+    decisive emptiness check and insert share one locked transaction, so two
+    application workers starting together cannot both seed a default row.
+    """
+    import json as _json
+    import uuid as _uuid
+
+    settings_file = Path(SETTINGS_FILE)
+    if not settings_file.exists():
+        return
+    try:
+        s = _json.loads(settings_file.read_text(encoding="utf-8"))
+    except Exception:
+        return
+
+    imap_host = (s.get("imap_host") or "").strip()
+    smtp_host = (s.get("smtp_host") or "").strip()
+    if not imap_host and not smtp_host:
+        return
+
+    db = None
+    try:
+        if not inspect(engine).has_table(EmailAccount.__tablename__):
+            return
+        db = SessionLocal()
+        lock_email_account_owner_mutations(db, "")
+        existing = db.execute(text("SELECT COUNT(*) FROM email_accounts")).scalar() or 0
+        if existing > 0:
+            return
 
         now = utcnow_naive()
-        with engine.begin() as conn:
-            conn.execute(text("""
-                INSERT INTO email_accounts
-                  (id, owner, name, is_default, enabled,
-                   imap_host, imap_port, imap_user, imap_password, imap_starttls,
-                   smtp_host, smtp_port, smtp_user, smtp_password,
-                   from_address, created_at, updated_at)
-                VALUES
-                  (:id, :owner, :name, :is_default, :enabled,
-                   :imap_host, :imap_port, :imap_user, :imap_password, :imap_starttls,
-                   :smtp_host, :smtp_port, :smtp_user, :smtp_password,
-                   :from_address, :created_at, :updated_at)
-            """), {
-                "id": _uuid.uuid4().hex,
-                "owner": None,
-                "name": "Default",
-                "is_default": True,
-                "enabled": True,
-                "imap_host": imap_host,
-                "imap_port": int(s.get("imap_port") or 993),
-                "imap_user": s.get("imap_user") or "",
-                "imap_password": s.get("imap_password") or "",
-                "imap_starttls": bool(s.get("imap_starttls", True)),
-                "smtp_host": smtp_host,
-                "smtp_port": int(s.get("smtp_port") or 465),
-                "smtp_user": s.get("smtp_user") or "",
-                "smtp_password": s.get("smtp_password") or "",
-                "from_address": s.get("email_from") or "",
-                "created_at": now,
-                "updated_at": now,
-            })
-            logging.getLogger(__name__).info("Seeded email_accounts 'Default' from settings.json")
+        db.execute(text("""
+            INSERT INTO email_accounts
+              (id, owner, name, is_default, enabled,
+               imap_host, imap_port, imap_user, imap_password, imap_starttls,
+               smtp_host, smtp_port, smtp_user, smtp_password,
+               from_address, created_at, updated_at)
+            VALUES
+              (:id, :owner, :name, :is_default, :enabled,
+               :imap_host, :imap_port, :imap_user, :imap_password, :imap_starttls,
+               :smtp_host, :smtp_port, :smtp_user, :smtp_password,
+               :from_address, :created_at, :updated_at)
+        """), {
+            "id": _uuid.uuid4().hex,
+            "owner": None,
+            "name": "Default",
+            "is_default": True,
+            "enabled": True,
+            "imap_host": imap_host,
+            "imap_port": int(s.get("imap_port") or 993),
+            "imap_user": s.get("imap_user") or "",
+            "imap_password": s.get("imap_password") or "",
+            "imap_starttls": bool(s.get("imap_starttls", True)),
+            "smtp_host": smtp_host,
+            "smtp_port": int(s.get("smtp_port") or 465),
+            "smtp_user": s.get("smtp_user") or "",
+            "smtp_password": s.get("smtp_password") or "",
+            "from_address": s.get("email_from") or "",
+            "created_at": now,
+            "updated_at": now,
+        })
+        db.commit()
+        logger.info("Seeded email_accounts 'Default' from settings.json")
     except Exception as e:
-        logging.getLogger(__name__).warning(f"seed email account migration: {e}")
+        if db is not None:
+            db.rollback()
+        logger.warning("seed email account migration: %s", e)
+    finally:
+        if db is not None:
+            db.close()
 
 
 # WARNING: Foreign-key enforcement is enabled globally for all SQLite connections.
@@ -1960,6 +2117,7 @@ def init_db():
     _migrate_add_crew_member_id()
     _migrate_add_assistant_columns()
     _migrate_add_email_smtp_security()
+    _migrate_email_account_default_invariant()
     _migrate_seed_email_account()
     _migrate_add_calendar_metadata()
     _migrate_add_calendar_is_utc()

routes/auth_routes.py

@@ -345,9 +345,61 @@ def setup_auth_routes(auth_manager: AuthManager) -> APIRouter:
         # docs, email accounts, tasks, etc.
         try:
             from sqlalchemy import func
-            from core.database import Base, SessionLocal
+            from core.database import (
+                Base,
+                EmailAccount,
+                SessionLocal,
+                lock_email_account_owner_mutations,
+            )
             db = SessionLocal()
             try:
+                # Email-account defaults are protected by per-owner mutex rows.
+                # A rename crosses two owner partitions, so lock both in the
+                # shared helper's canonical order before inspecting either.
+                lock_email_account_owner_mutations(
+                    db, old_username, new_username
+                )
+
+                source_default_ids = [
+                    row[0]
+                    for row in (
+                        db.query(EmailAccount.id)
+                        .filter(
+                            func.lower(EmailAccount.owner) == old_username,
+                            EmailAccount.is_default == True,  # noqa: E712
+                        )
+                        .order_by(EmailAccount.created_at.asc(), EmailAccount.id.asc())
+                        .all()
+                    )
+                ]
+                destination_default_ids = [
+                    row[0]
+                    for row in (
+                        db.query(EmailAccount.id)
+                        .filter(
+                            func.lower(EmailAccount.owner) == new_username,
+                            EmailAccount.is_default == True,  # noqa: E712
+                        )
+                        .order_by(EmailAccount.created_at.asc(), EmailAccount.id.asc())
+                        .all()
+                    )
+                ]
+                if destination_default_ids:
+                    clear_default_ids = (
+                        destination_default_ids[1:] + source_default_ids
+                    )
+                else:
+                    clear_default_ids = source_default_ids[1:]
+                if clear_default_ids:
+                    (
+                        db.query(EmailAccount)
+                        .filter(EmailAccount.id.in_(clear_default_ids))
+                        .update(
+                            {EmailAccount.is_default: False},
+                            synchronize_session=False,
+                        )
+                    )
+
                 for mapper in Base.registry.mappers:
                     model = mapper.class_
                     if not hasattr(model, "owner"):

routes/email_routes.py

@@ -45,6 +45,7 @@ from src.upload_limits import read_upload_limited, EMAIL_COMPOSE_UPLOAD_MAX_BYTE
 
 from routes.email_helpers import (
     _strip_think, _extract_reply, _apply_email_style_mechanics, require_owner, require_user, _assert_owns_account,
+    _account_visible_to_owner,
     _q, _attach_compose_uploads, _cleanup_compose_uploads,
     _load_settings, _save_settings, _get_email_config,
     _send_smtp_message, _smtp_security_mode,
@@ -194,6 +195,64 @@ def _coerce_port(value, default):
         return None, f"Invalid port {value!r}; must be a whole number"
 
 
+def _lock_email_account_owner_mutation(db, *owners: str) -> None:
+    """Delegate account/default serialization to the shared DB primitive."""
+    from core.database import lock_email_account_owner_mutations
+
+    lock_email_account_owner_mutations(db, *owners)
+
+
+def _email_account_owner_scope(query, owner: str):
+    """Restrict a query to one normalized EmailAccount owner partition."""
+    from core.database import EmailAccount
+    from sqlalchemy import or_
+
+    if owner:
+        return query.filter(EmailAccount.owner == owner)
+    return query.filter(or_(EmailAccount.owner == None, EmailAccount.owner == ""))  # noqa: E711
+
+
+def _discover_email_account_mutation_scope(account_id: str, owner: str) -> str:
+    """Read the initial lock key and fail closed before a mutation session."""
+    from core.database import EmailAccount, SessionLocal
+
+    db = SessionLocal()
+    try:
+        row = db.get(EmailAccount, account_id)
+        if row is None or (owner and not _account_visible_to_owner(row, owner)):
+            raise HTTPException(404, "Account not found")
+        return row.owner or ""
+    except HTTPException:
+        raise
+    except Exception as exc:
+        logger.error("Account-owner mutation check failed: %s", exc)
+        raise HTTPException(503, "Account check failed")
+    finally:
+        db.close()
+
+
+def _lock_and_reload_email_account(db, account_id: str, owner: str, scope: str):
+    """Lock, reload, and revalidate an account, retrying if its owner moved."""
+    from core.database import EmailAccount
+
+    owner_scopes = {scope or ""}
+    while True:
+        _lock_email_account_owner_mutation(db, *owner_scopes)
+        row = db.get(EmailAccount, account_id, populate_existing=True)
+        if row is None or (owner and not _account_visible_to_owner(row, owner)):
+            raise HTTPException(404, "Account not found")
+
+        current_scope = row.owner or ""
+        if current_scope in owner_scopes or db.get_bind().dialect.name == "sqlite":
+            return row
+
+        # The account changed owner after discovery but before lock acquisition.
+        # Release the partial lock set and reacquire all observed scopes in the
+        # shared helper's canonical order, then validate from the database again.
+        db.rollback()
+        owner_scopes.add(current_scope)
+
+
 def _email_tag_owner_aliases(account_id: str | None, owner: str = "") -> list[str]:
     aliases = [owner or ""]
     try:
@@ -5428,9 +5487,9 @@ def setup_email_routes():
         import uuid as _uuid
         db = SessionLocal()
         try:
+            _lock_email_account_owner_mutation(db, owner)
             q = db.query(EmailAccount).filter(EmailAccount.is_default == True)  # noqa: E712
-            if owner:
-                q = q.filter(EmailAccount.owner == owner)
+            q = _email_account_owner_scope(q, owner)
             row = q.first()
             if row is None:
                 row = EmailAccount(id=_uuid.uuid4().hex, owner=owner, name="Default", is_default=True, enabled=True)
@@ -5456,8 +5515,7 @@ def setup_email_routes():
             if data.get("smtp_password"):
                 row.smtp_password = _enc(data["smtp_password"])
             clear_q = db.query(EmailAccount).filter(EmailAccount.id != row.id)
-            if owner:
-                clear_q = clear_q.filter(EmailAccount.owner == owner)
+            clear_q = _email_account_owner_scope(clear_q, owner)
             clear_q.update({EmailAccount.is_default: False})
             db.commit()
         finally:
@@ -5552,6 +5610,7 @@ def setup_email_routes():
             return {"ok": False, "error": port_err}
         db = SessionLocal()
         try:
+            _lock_email_account_owner_mutation(db, owner)
             row = EmailAccount(
                 id=_uuid.uuid4().hex,
                 name=name,
@@ -5578,9 +5637,7 @@ def setup_email_routes():
             # the one-default invariant — but scope it to THIS user's accounts,
             # otherwise creating a default would clear every other user's
             # default flag too.
-            scope_q = db.query(EmailAccount)
-            if owner:
-                scope_q = scope_q.filter(EmailAccount.owner == owner)
+            scope_q = _email_account_owner_scope(db.query(EmailAccount), owner)
             existing_count = scope_q.count()
             if row.is_default or existing_count == 0:
                 scope_q.update({EmailAccount.is_default: False})
@@ -5631,28 +5688,39 @@ def setup_email_routes():
 
     @router.delete("/accounts/{account_id}")
     async def delete_email_account(account_id: str, owner: str = Depends(require_user)):
-        _assert_owns_account(account_id, owner)
+        initial_scope = _discover_email_account_mutation_scope(account_id, owner)
         from core.database import SessionLocal, EmailAccount
         db = SessionLocal()
         try:
-            row = db.get(EmailAccount, account_id)
-            if not row:
-                return {"ok": False, "error": "Account not found"}
+            row = _lock_and_reload_email_account(
+                db, account_id, owner, initial_scope
+            )
+            row_scope = row.owner or ""
             was_default = bool(row.is_default)
             db.delete(row)
-            db.commit()
+            # Flush the removal before staging a replacement default.  The
+            # partial unique index is checked statement-by-statement, and the
+            # ORM is otherwise free to UPDATE the promoted row before DELETE.
+            db.flush()
             # If the deleted row was default, promote the next-oldest enabled
             # row owned by THIS user. Without the owner filter we'd promote
             # another user's account and the deleter would silently inherit
             # it as their default.
             if was_default:
-                promote_q = db.query(EmailAccount).filter(EmailAccount.enabled == True)  # noqa: E712
-                if owner:
-                    promote_q = promote_q.filter(EmailAccount.owner == owner)
-                promote = promote_q.order_by(EmailAccount.created_at.asc()).first()
+                promote_q = db.query(EmailAccount).filter(
+                    EmailAccount.id != account_id,
+                    EmailAccount.enabled == True,  # noqa: E712
+                )
+                promote_q = _email_account_owner_scope(promote_q, row_scope)
+                promote = promote_q.order_by(
+                    EmailAccount.created_at.asc(), EmailAccount.id.asc()
+                ).first()
                 if promote:
                     promote.is_default = True
-                    db.commit()
+            # Deletion and any replacement promotion are one durable state
+            # transition, so another worker can never observe or race the old
+            # split-commit gap.
+            db.commit()
             return {"ok": True}
         finally:
             db.close()
@@ -5865,18 +5933,18 @@ def setup_email_routes():
 
     @router.post("/accounts/{account_id}/set-default")
     async def set_default_account(account_id: str, owner: str = Depends(require_user)):
-        _assert_owns_account(account_id, owner)
+        initial_scope = _discover_email_account_mutation_scope(account_id, owner)
         from core.database import SessionLocal, EmailAccount
         db = SessionLocal()
         try:
-            row = db.get(EmailAccount, account_id)
-            if not row:
-                return {"ok": False, "error": "Account not found"}
-            # SECURITY: scope the "clear other defaults" sweep to this user's
-            # accounts so we don't unset another user's default flag.
-            clear_q = db.query(EmailAccount)
-            if owner:
-                clear_q = clear_q.filter(EmailAccount.owner == owner)
+            row = _lock_and_reload_email_account(
+                db, account_id, owner, initial_scope
+            )
+            # Scope the sweep to the target row's normalized owner partition;
+            # this also handles visible legacy NULL/empty-owner accounts.
+            clear_q = _email_account_owner_scope(
+                db.query(EmailAccount), row.owner or ""
+            )
             clear_q.update({EmailAccount.is_default: False})
             row.is_default = True
             db.commit()

scripts/demo_email/demo_account.py

@@ -1,5 +1,5 @@
 #!/usr/bin/env python3
-"""Create/remove the switchable, non-default 'Demo' EmailAccount in Odysseus.
+"""Create/remove the switchable 'Demo' EmailAccount in Odysseus.
 
 Mirrors the existing local-Dovecot account (localhost:31143, STARTTLS) but points
 at the throwaway demo@odysseus.local mailbox. Password is stored Fernet-encrypted
@@ -20,7 +20,14 @@ from pathlib import Path
 ROOT = Path(__file__).resolve().parent.parent.parent
 sys.path.insert(0, str(ROOT))
 
-from core.database import SessionLocal, EmailAccount, Base, engine  # noqa: E402
+from core.database import (  # noqa: E402
+    Base,
+    EmailAccount,
+    SessionLocal,
+    engine,
+    lock_email_account_owner_mutations,
+)
+from sqlalchemy import or_  # noqa: E402
 from src.secret_storage import encrypt  # noqa: E402
 
 NAME = "Demo"
@@ -31,18 +38,98 @@ IMAP_PASSWORD = "demodemo"
 OWNER = ""
 
 
-def setup() -> int:
-    Base.metadata.create_all(bind=engine)
+def _owner_scope(query, owner: str):
+    if owner:
+        return query.filter(EmailAccount.owner == owner)
+    return query.filter(or_(EmailAccount.owner == None, EmailAccount.owner == ""))  # noqa: E711
+
+
+def _discover_demo_scopes() -> set[str]:
     db = SessionLocal()
     try:
-        acct = db.query(EmailAccount).filter(
-            EmailAccount.name == NAME, EmailAccount.imap_user == IMAP_USER
-        ).first()
+        return {
+            row.owner or ""
+            for row in db.query(EmailAccount).filter(
+                EmailAccount.name == NAME,
+                EmailAccount.imap_user == IMAP_USER,
+            ).all()
+        }
+    finally:
+        db.close()
+
+
+def _lock_and_load_demo_rows(db, scopes: set[str]):
+    """Reload Demo rows under every observed owner lock."""
+    scopes = set(scopes) or {OWNER}
+    while True:
+        lock_email_account_owner_mutations(db, *scopes)
+        rows = (
+            db.query(EmailAccount)
+            .filter(
+                EmailAccount.name == NAME,
+                EmailAccount.imap_user == IMAP_USER,
+            )
+            .order_by(EmailAccount.created_at.asc(), EmailAccount.id.asc())
+            .all()
+        )
+        current_scopes = {row.owner or "" for row in rows}
+        if current_scopes.issubset(scopes) or db.get_bind().dialect.name == "sqlite":
+            return rows
+        db.rollback()
+        scopes.update(current_scopes)
+
+
+def _promote_oldest_enabled(db, owner: str, excluded_ids: list[str]) -> None:
+    remaining = _owner_scope(
+        db.query(EmailAccount).filter(
+            EmailAccount.enabled == True,  # noqa: E712
+            ~EmailAccount.id.in_(excluded_ids),
+        ),
+        owner,
+    )
+    if remaining.filter(EmailAccount.is_default == True).first() is not None:  # noqa: E712
+        return
+    promote = remaining.order_by(
+        EmailAccount.created_at.asc(), EmailAccount.id.asc()
+    ).first()
+    if promote is not None:
+        promote.is_default = True
+
+
+def setup() -> int:
+    Base.metadata.create_all(bind=engine)
+    scopes = _discover_demo_scopes() | {OWNER}
+    db = SessionLocal()
+    try:
+        rows = _lock_and_load_demo_rows(db, scopes)
+        acct = rows[0] if rows else None
         if acct is None:
             acct = EmailAccount(id=uuid.uuid4().hex, name=NAME)
             db.add(acct)
+        old_scope = acct.owner or ""
+        was_default = bool(acct.is_default)
+        if old_scope != OWNER:
+            # Move a non-default row first so the unique index cannot see two
+            # defaults transiently while SQLAlchemy flushes the owner move and
+            # old-scope promotion in separate UPDATE statements.
+            acct.is_default = False
+            acct.owner = OWNER
+            db.flush()
+            if was_default:
+                _promote_oldest_enabled(db, old_scope, [acct.id])
+
+        target_default = _owner_scope(
+            db.query(EmailAccount).filter(
+                EmailAccount.id != acct.id,
+                EmailAccount.is_default == True,  # noqa: E712
+            ),
+            OWNER,
+        ).first()
         acct.owner = OWNER
-        acct.is_default = False          # never default — user switches to it
+        # Keep Demo non-default when a real default exists.  If it is the only
+        # enabled account, it must be default to preserve normal create
+        # semantics and avoid leaving the owner partition without one.
+        acct.is_default = target_default is None
         acct.enabled = True
         acct.imap_host = "localhost"
         acct.imap_port = 31143
@@ -57,20 +144,27 @@ def setup() -> int:
         acct.smtp_password = encrypt(IMAP_PASSWORD)
         acct.from_address = IMAP_USER
         db.commit()
-        print(f"'{NAME}' account ready (id={acct.id}, non-default, switchable).")
+        state = "default" if acct.is_default else "non-default"
+        print(f"'{NAME}' account ready (id={acct.id}, {state}, switchable).")
         return 0
     finally:
         db.close()
 
 
 def teardown() -> int:
+    scopes = _discover_demo_scopes()
     db = SessionLocal()
     try:
-        rows = db.query(EmailAccount).filter(
-            EmailAccount.name == NAME, EmailAccount.imap_user == IMAP_USER
-        ).all()
+        rows = _lock_and_load_demo_rows(db, scopes)
+        deleted_ids = [row.id for row in rows]
+        default_scopes = {row.owner or "" for row in rows if row.is_default}
         for r in rows:
             db.delete(r)
+        # Ensure the old default DELETE reaches the database before a
+        # replacement UPDATE; the unique index is enforced per statement.
+        db.flush()
+        for owner in default_scopes:
+            _promote_oldest_enabled(db, owner, deleted_ids)
         db.commit()
         print(f"removed {len(rows)} '{NAME}' account row(s).")
         return 0

tests/test_email_account_default_serialization.py

@@ -0,0 +1,522 @@
+"""Regressions for process-safe email-account default mutations.
+
+The file-backed SQLite fixture uses a fresh connection for every Session.
+That exercises the same database lock boundary used by separate web workers,
+rather than relying on an in-process Python lock.
+"""
+
+import asyncio
+import json
+import sys
+import threading
+import types
+from pathlib import Path
+from types import SimpleNamespace
+from unittest import mock
+
+import pytest
+from fastapi import HTTPException
+from sqlalchemy import create_engine, create_mock_engine, text
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.orm import sessionmaker
+from sqlalchemy.pool import NullPool
+
+
+@pytest.fixture
+def account_db(tmp_path, monkeypatch):
+    from core import database as core_db
+
+    engine = create_engine(
+        f"sqlite:///{tmp_path / 'accounts.db'}",
+        connect_args={"check_same_thread": False, "timeout": 5},
+        poolclass=NullPool,
+    )
+    core_db.Base.metadata.create_all(engine)
+    factory = sessionmaker(
+        bind=engine,
+        autocommit=False,
+        autoflush=False,
+    )
+    monkeypatch.setattr(core_db, "SessionLocal", factory)
+    yield factory
+    engine.dispose()
+
+
+def _endpoint(method, path):
+    from routes import email_routes
+
+    with mock.patch.object(email_routes, "_start_poller"):
+        router = email_routes.setup_email_routes()
+    for route in router.routes:
+        if route.path == path and method in getattr(route, "methods", set()):
+            return route.endpoint
+    raise AssertionError(f"email route not found: {method} {path}")
+
+
+def _named_endpoint(router, name):
+    for route in router.routes:
+        if getattr(getattr(route, "endpoint", None), "__name__", "") == name:
+            return route.endpoint
+    raise AssertionError(f"route not found: {name}")
+
+
+def _seed_account(factory, account_id, owner, *, is_default=False, enabled=True):
+    from core.database import EmailAccount
+
+    db = factory()
+    try:
+        db.add(
+            EmailAccount(
+                id=account_id,
+                owner=owner,
+                name=account_id,
+                is_default=is_default,
+                enabled=enabled,
+            )
+        )
+        db.commit()
+    finally:
+        db.close()
+
+
+def _rows(factory):
+    from core.database import EmailAccount
+
+    db = factory()
+    try:
+        return [
+            (row.id, row.owner, bool(row.is_default))
+            for row in db.query(EmailAccount).order_by(EmailAccount.id).all()
+        ]
+    finally:
+        db.close()
+
+
+def _install_lock_pause(monkeypatch, paused_thread_name):
+    """Pause one worker after acquisition and observe another waiting."""
+    from routes import email_routes
+
+    real_lock = email_routes._lock_email_account_owner_mutation
+    first_acquired = threading.Event()
+    release_first = threading.Event()
+    contender_attempted = threading.Event()
+    contender_acquired = threading.Event()
+
+    def controlled_lock(db, owner):
+        is_first = threading.current_thread().name == paused_thread_name
+        if not is_first:
+            contender_attempted.set()
+        real_lock(db, owner)
+        if is_first:
+            first_acquired.set()
+            assert release_first.wait(5), "timed out releasing first mutation"
+        else:
+            contender_acquired.set()
+
+    monkeypatch.setattr(
+        email_routes,
+        "_lock_email_account_owner_mutation",
+        controlled_lock,
+    )
+    return first_acquired, release_first, contender_attempted, contender_acquired
+
+
+def test_concurrent_first_account_creates_choose_one_default(account_db, monkeypatch):
+    create_account = _endpoint("POST", "/api/email/accounts")
+    first_acquired, release_first, attempted, acquired = _install_lock_pause(
+        monkeypatch, "first-account"
+    )
+    results = {}
+
+    def create(name):
+        results[name] = asyncio.run(
+            create_account({"name": name, "is_default": False}, owner="alice")
+        )
+
+    first = threading.Thread(target=create, args=("First",), name="first-account")
+    second = threading.Thread(target=create, args=("Second",), name="second-account")
+    first.start()
+    assert first_acquired.wait(5)
+    second.start()
+    assert attempted.wait(5)
+    assert not acquired.wait(0.1), "second session bypassed the database mutation lock"
+
+    release_first.set()
+    first.join(5)
+    second.join(5)
+
+    assert not first.is_alive()
+    assert not second.is_alive()
+    assert results["First"]["ok"] is True
+    assert results["Second"]["ok"] is True
+    defaults = [row for row in _rows(account_db) if row[2]]
+    assert [(row[1], row[2]) for row in defaults] == [("alice", True)]
+    assert len(defaults) == 1
+
+
+def test_delete_promotion_and_set_default_are_one_serial_transition(
+    account_db, monkeypatch
+):
+    from sqlalchemy.orm import Session as OrmSession
+
+    _seed_account(account_db, "alice-a", "alice", is_default=True)
+    _seed_account(account_db, "alice-b", "alice")
+    _seed_account(account_db, "alice-c", "alice")
+    _seed_account(account_db, "bob-a", "bob", is_default=True)
+
+    delete_account = _endpoint("DELETE", "/api/email/accounts/{account_id}")
+    set_default = _endpoint("POST", "/api/email/accounts/{account_id}/set-default")
+    first_acquired, release_first, attempted, acquired = _install_lock_pause(
+        monkeypatch, "delete-default"
+    )
+    delete_commit_finished = threading.Event()
+    release_delete_after_commit = threading.Event()
+    real_commit = OrmSession.commit
+    results = {}
+
+    def pause_after_delete_commit(session):
+        real_commit(session)
+        if (
+            threading.current_thread().name == "delete-default"
+            and not delete_commit_finished.is_set()
+        ):
+            delete_commit_finished.set()
+            assert release_delete_after_commit.wait(5), (
+                "timed out releasing delete after its first commit"
+            )
+
+    monkeypatch.setattr(OrmSession, "commit", pause_after_delete_commit)
+
+    def delete_old_default():
+        results["delete"] = asyncio.run(
+            delete_account("alice-a", owner="alice")
+        )
+
+    def select_new_default():
+        results["set"] = asyncio.run(
+            set_default("alice-c", owner="alice")
+        )
+
+    delete_thread = threading.Thread(target=delete_old_default, name="delete-default")
+    set_thread = threading.Thread(target=select_new_default, name="set-default")
+    delete_thread.start()
+    assert first_acquired.wait(5)
+    set_thread.start()
+    assert attempted.wait(5)
+    assert not acquired.wait(0.1), "set-default bypassed the delete transaction"
+
+    release_first.set()
+    assert delete_commit_finished.wait(5)
+    # The deletion transaction has committed.  Let the contender complete
+    # before the deleting handler can continue: if promotion were still a
+    # second commit, it would now run after set-default and recreate two
+    # defaults deterministically.
+    assert acquired.wait(5)
+    set_thread.join(5)
+    release_delete_after_commit.set()
+    delete_thread.join(5)
+
+    assert not delete_thread.is_alive()
+    assert not set_thread.is_alive()
+    assert results == {"delete": {"ok": True}, "set": {"ok": True}}
+    assert _rows(account_db) == [
+        ("alice-b", "alice", False),
+        ("alice-c", "alice", True),
+        ("bob-a", "bob", True),
+    ]
+
+
+def test_upgrade_normalizes_legacy_defaults_and_installs_unique_index(
+    tmp_path, monkeypatch
+):
+    """A pre-index schema upgrades without requiring newer account columns."""
+    from core import database as core_db
+
+    engine = create_engine(
+        f"sqlite:///{tmp_path / 'legacy-accounts.db'}",
+        connect_args={"check_same_thread": False},
+        poolclass=NullPool,
+    )
+    try:
+        with engine.begin() as conn:
+            conn.execute(text("""
+                CREATE TABLE email_accounts (
+                    id VARCHAR PRIMARY KEY,
+                    owner VARCHAR,
+                    name VARCHAR NOT NULL,
+                    is_default BOOLEAN NOT NULL,
+                    enabled BOOLEAN NOT NULL,
+                    created_at DATETIME,
+                    updated_at DATETIME
+                )
+            """))
+            conn.execute(text("""
+                INSERT INTO email_accounts
+                    (id, owner, name, is_default, enabled, created_at, updated_at)
+                VALUES
+                    ('legacy-old', NULL, 'Old', 1, 1, '2024-01-01', '2024-01-01'),
+                    ('legacy-new', '', 'New', 1, 1, '2025-01-01', '2025-01-01')
+            """))
+
+        monkeypatch.setattr(core_db, "engine", engine)
+        core_db._migrate_email_account_default_invariant()
+        core_db._migrate_email_account_default_invariant()  # idempotent replay
+
+        with engine.connect() as conn:
+            defaults = conn.execute(text("""
+                SELECT id FROM email_accounts
+                WHERE is_default IS TRUE
+                ORDER BY id
+            """)).scalars().all()
+            index_names = {
+                row[1] for row in conn.execute(text("PRAGMA index_list(email_accounts)"))
+            }
+        assert defaults == ["legacy-old"]
+        assert core_db._EMAIL_ACCOUNT_DEFAULT_INDEX in index_names
+
+        with pytest.raises(IntegrityError):
+            with engine.begin() as conn:
+                conn.execute(text("""
+                    INSERT INTO email_accounts
+                        (id, owner, name, is_default, enabled, created_at, updated_at)
+                    VALUES
+                        ('legacy-third', NULL, 'Third', 1, 1, '2026-01-01', '2026-01-01')
+                """))
+    finally:
+        engine.dispose()
+
+
+def test_concurrent_legacy_seed_is_one_locked_transaction(
+    tmp_path, monkeypatch, caplog
+):
+    from core import database as core_db
+
+    engine = create_engine(
+        f"sqlite:///{tmp_path / 'seed-accounts.db'}",
+        connect_args={"check_same_thread": False, "timeout": 5},
+        poolclass=NullPool,
+    )
+    core_db.Base.metadata.create_all(engine)
+    factory = sessionmaker(bind=engine, autocommit=False, autoflush=False)
+    settings_file = tmp_path / "settings.json"
+    settings_file.write_text(
+        json.dumps({"imap_host": "imap.example.test", "imap_user": "alice"}),
+        encoding="utf-8",
+    )
+    monkeypatch.setattr(core_db, "engine", engine)
+    monkeypatch.setattr(core_db, "SessionLocal", factory)
+    monkeypatch.setattr(core_db, "SETTINGS_FILE", str(settings_file))
+
+    read_barrier = threading.Barrier(2)
+    real_read_text = Path.read_text
+
+    def synchronized_read(path, *args, **kwargs):
+        value = real_read_text(path, *args, **kwargs)
+        if path == settings_file:
+            read_barrier.wait(5)
+        return value
+
+    monkeypatch.setattr(Path, "read_text", synchronized_read)
+    threads = [
+        threading.Thread(target=core_db._migrate_seed_email_account)
+        for _ in range(2)
+    ]
+    try:
+        for thread in threads:
+            thread.start()
+        for thread in threads:
+            thread.join(5)
+        assert all(not thread.is_alive() for thread in threads)
+
+        with engine.connect() as conn:
+            rows = conn.execute(text("""
+                SELECT owner, is_default FROM email_accounts
+                ORDER BY id
+            """)).all()
+        assert rows == [(None, 1)]
+        assert "seed email account migration:" not in caplog.text
+    finally:
+        engine.dispose()
+
+
+def test_multi_owner_row_locks_are_acquired_in_canonical_order():
+    from core.database import lock_email_account_owner_mutations
+
+    class FakeSession:
+        def __init__(self):
+            self.locked = []
+
+        def get_bind(self):
+            return SimpleNamespace(dialect=SimpleNamespace(name="postgresql"))
+
+        def get(self, _model, owner_key, **kwargs):
+            assert kwargs == {"with_for_update": True}
+            self.locked.append(owner_key)
+            return object()
+
+    db = FakeSession()
+    lock_email_account_owner_mutations(db, "zeta", "", "alpha", "zeta")
+    assert db.locked == ["", "alpha", "zeta"]
+
+
+def test_postgresql_fresh_schema_emits_default_unique_index():
+    from core import database as core_db
+
+    statements = []
+    engine_holder = {}
+
+    def capture(statement, *_args, **_kwargs):
+        statements.append(
+            str(statement.compile(dialect=engine_holder["engine"].dialect))
+        )
+
+    mock_engine = create_mock_engine("postgresql://", capture)
+    engine_holder["engine"] = mock_engine
+    core_db.EmailAccount.__table__.create(mock_engine)
+
+    assert any(
+        core_db._EMAIL_ACCOUNT_DEFAULT_INDEX in statement
+        and "COALESCE(owner, '')" in statement
+        and "WHERE is_default IS TRUE" in statement
+        for statement in statements
+    )
+
+
+def test_rename_serializes_old_and_new_owner_and_stale_set_default_fails_closed(
+    account_db, monkeypatch, tmp_path
+):
+    from core import database as core_db
+    from routes import auth_routes
+
+    _seed_account(account_db, "alice-a", "alice", is_default=True)
+    _seed_account(account_db, "alice-b", "alice")
+    _seed_account(account_db, "bob-a", "bob", is_default=True)
+
+    prefs_module = types.ModuleType("routes.prefs_routes")
+    prefs_module._load = lambda: {}
+    prefs_module._save = lambda _data: None
+    monkeypatch.setitem(sys.modules, "routes.prefs_routes", prefs_module)
+    monkeypatch.setattr(
+        auth_routes, "DEEP_RESEARCH_DIR", str(tmp_path / "deep_research")
+    )
+    monkeypatch.setattr(auth_routes, "MEMORY_FILE", str(tmp_path / "memory.json"))
+    monkeypatch.setattr(auth_routes, "SKILLS_DIR", str(tmp_path / "skills"))
+
+    auth_manager = mock.MagicMock()
+    auth_manager.get_username_for_token.return_value = "admin"
+    auth_manager.is_admin.return_value = True
+    auth_manager.users = {"admin": {}, "alice": {}}
+    auth_manager.rename_user.return_value = True
+    rename_user = _named_endpoint(
+        auth_routes.setup_auth_routes(auth_manager), "rename_user"
+    )
+    set_default = _endpoint("POST", "/api/email/accounts/{account_id}/set-default")
+
+    rename_acquired = threading.Event()
+    release_rename = threading.Event()
+    set_attempted = threading.Event()
+    set_acquired = threading.Event()
+    real_lock = core_db.lock_email_account_owner_mutations
+
+    def controlled_lock(db, *owners):
+        thread_name = threading.current_thread().name
+        if thread_name == "rename-owner":
+            real_lock(db, *owners)
+            rename_acquired.set()
+            assert release_rename.wait(5)
+            return
+        if thread_name == "stale-set-default":
+            set_attempted.set()
+            real_lock(db, *owners)
+            set_acquired.set()
+            return
+        real_lock(db, *owners)
+
+    monkeypatch.setattr(core_db, "lock_email_account_owner_mutations", controlled_lock)
+    request = SimpleNamespace(
+        cookies={"odysseus_session": "admin-token"},
+        app=SimpleNamespace(
+            state=SimpleNamespace(
+                invalidate_token_cache=lambda: None,
+                session_manager=None,
+                research_handler=None,
+                upload_handler=None,
+                personal_docs_manager=None,
+            )
+        ),
+    )
+    results = {}
+
+    def rename_owner():
+        results["rename"] = asyncio.run(
+            rename_user("alice", SimpleNamespace(username="bob"), request)
+        )
+
+    def select_stale_default():
+        try:
+            results["set"] = asyncio.run(
+                set_default("alice-b", owner="alice")
+            )
+        except Exception as exc:  # asserted below with its HTTP status
+            results["set_error"] = exc
+
+    rename_thread = threading.Thread(target=rename_owner, name="rename-owner")
+    set_thread = threading.Thread(
+        target=select_stale_default, name="stale-set-default"
+    )
+    rename_thread.start()
+    assert rename_acquired.wait(5)
+    set_thread.start()
+    assert set_attempted.wait(5)
+    assert not set_acquired.wait(0.1), "set-default bypassed the rename lock"
+
+    release_rename.set()
+    rename_thread.join(5)
+    set_thread.join(5)
+
+    assert not rename_thread.is_alive()
+    assert not set_thread.is_alive()
+    assert results["rename"]["ok"] is True
+    assert isinstance(results["set_error"], HTTPException)
+    assert results["set_error"].status_code == 404
+    assert _rows(account_db) == [
+        ("alice-a", "bob", False),
+        ("alice-b", "bob", False),
+        ("bob-a", "bob", True),
+    ]
+
+
+def test_demo_teardown_promotes_replacement_in_same_transaction(
+    account_db, monkeypatch
+):
+    from core.database import EmailAccount
+    from scripts.demo_email import demo_account
+
+    db = account_db()
+    try:
+        db.add_all([
+            EmailAccount(
+                id="real",
+                owner="",
+                name="Real",
+                is_default=False,
+                enabled=True,
+            ),
+            EmailAccount(
+                id="demo",
+                owner="",
+                name=demo_account.NAME,
+                imap_user=demo_account.IMAP_USER,
+                is_default=True,
+                enabled=True,
+            ),
+        ])
+        db.commit()
+    finally:
+        db.close()
+
+    monkeypatch.setattr(demo_account, "SessionLocal", account_db)
+    monkeypatch.setattr(demo_account, "engine", account_db.kw["bind"])
+
+    assert demo_account.teardown() == 0
+    assert _rows(account_db) == [("real", "", True)]

tests/test_rename_user_owner_sync.py

@@ -114,6 +114,12 @@ def _force_sql_owner_migration_failure(monkeypatch):
         def filter(self, *_args, **_kwargs):
             return self
 
+        def order_by(self, *_args, **_kwargs):
+            return self
+
+        def all(self):
+            return []
+
         def update(self, *_args, **_kwargs):
             raise RuntimeError("forced owner migration failure")
 
@@ -125,6 +131,12 @@ def _force_sql_owner_migration_failure(monkeypatch):
         def query(self, _model):
             return FailingQuery()
 
+        def get_bind(self):
+            return SimpleNamespace(dialect=SimpleNamespace(name="postgresql"))
+
+        def get(self, _model, _key, **_kwargs):
+            return object()
+
         def rollback(self):
             self.rolled_back = True