fix(cookbook): record real Windows pid for local serve so Stop kills the model (#5912)

* fix(cookbook): record real Windows pid for local serve so Stop kills the model

The Windows-local serve runner recorded Git Bash's `$$`, which is the
MSYS/Cygwin pid, not the Windows pid. Win32 tooling (taskkill,
Get-CimInstance ParentProcessId, Stop-Process) can't match an MSYS pid, so
the frontend Stop-Tree walk found nothing and the llama-server child survived
after Stop, leaving the model loaded and the GPU pinned.

Record the serving shell's true Win32 pid via `/proc/$$/winpid`, falling back
to the outer proc.pid already written from Python when the map is unavailable.

The existing pid-tracking test asserted the buggy `$$` literal at the source
level, so it passed while the feature was broken; update it to the winpid
behavior and add a focused regression test.

* fix(cookbook): make Windows serve pid handoff deterministic

---------

Co-authored-by: Alexandre Teixeira <111787685+alteixeira20@users.noreply.github.com>
Co-authored-by: Alexandre Teixeira <alexandremagteixeira@gmail.com>

.github/workflows/ci.yml

@@ -2,7 +2,7 @@ name: CI
 
 on:
   push:
-    branches: [main]
+    branches: [main, dev]
   pull_request:
 
 # Least privilege: none of the jobs write to the repo.
@@ -103,10 +103,7 @@ jobs:
   python-tests:
     name: Python tests (pytest)
     runs-on: ubuntu-latest
-    # Informational for now: the suite has known flaky / environment-dependent
+    # Make Python test validation authoritative for the configured scope.
-    # failures (test isolation + embedding-model assertions). Tracked under the
-    # ROADMAP "fresh install smoke tests" item; make this required once green.
-    continue-on-error: true
     steps:
       - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0  # v7.0.0
         with:

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():
+def _migrate_email_account_default_invariant():
-    """If email_accounts is empty and settings.json has legacy flat imap_host/smtp_host
+    """Normalize legacy duplicates and install durable at-most-one enforcement.
-    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."""
+    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:
+        with engine.begin() as conn:
-            tables = [r[0] for r in conn.execute(text(
+            if not inspect(conn).has_table(EmailAccount.__tablename__):
-                "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:
                 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
+            for account_id in duplicate_ids:
-        import uuid as _uuid
+                conn.execute(
-        from pathlib import Path
+                    text("UPDATE email_accounts SET is_default = :value WHERE id = :id"),
-        settings_file = Path(SETTINGS_FILE)
+                    {"value": False, "id": account_id},
-        if not settings_file.exists():
+                )
-            return
+            conn.execute(text(index_ddl))
-        try:
-            s = _json.loads(settings_file.read_text(encoding="utf-8"))
-        except Exception:
-            return
 
-        imap_host = (s.get("imap_host") or "").strip()
+        if duplicate_ids:
-        smtp_host = (s.get("smtp_host") or "").strip()
+            logger.warning(
-        if not imap_host and not smtp_host:
+                "Normalized %d duplicate default email account(s) before "
-            return  # nothing to migrate
+                "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:
+        db.execute(text("""
-            conn.execute(text("""
+            INSERT INTO email_accounts
-                INSERT INTO email_accounts
+              (id, owner, name, is_default, enabled,
-                  (id, owner, name, is_default, enabled,
+               imap_host, imap_port, imap_user, imap_password, imap_starttls,
-                   imap_host, imap_port, imap_user, imap_password, imap_starttls,
+               smtp_host, smtp_port, smtp_user, smtp_password,
-                   smtp_host, smtp_port, smtp_user, smtp_password,
+               from_address, created_at, updated_at)
-                   from_address, created_at, updated_at)
+            VALUES
-                VALUES
+              (:id, :owner, :name, :is_default, :enabled,
-                  (:id, :owner, :name, :is_default, :enabled,
+               :imap_host, :imap_port, :imap_user, :imap_password, :imap_starttls,
-                   :imap_host, :imap_port, :imap_user, :imap_password, :imap_starttls,
+               :smtp_host, :smtp_port, :smtp_user, :smtp_password,
-                   :smtp_host, :smtp_port, :smtp_user, :smtp_password,
+               :from_address, :created_at, :updated_at)
-                   :from_address, :created_at, :updated_at)
+        """), {
-            """), {
+            "id": _uuid.uuid4().hex,
-                "id": _uuid.uuid4().hex,
+            "owner": None,
-                "owner": None,
+            "name": "Default",
-                "name": "Default",
+            "is_default": True,
-                "is_default": True,
+            "enabled": True,
-                "enabled": True,
+            "imap_host": imap_host,
-                "imap_host": imap_host,
+            "imap_port": int(s.get("imap_port") or 993),
-                "imap_port": int(s.get("imap_port") or 993),
+            "imap_user": s.get("imap_user") or "",
-                "imap_user": s.get("imap_user") or "",
+            "imap_password": s.get("imap_password") or "",
-                "imap_password": s.get("imap_password") or "",
+            "imap_starttls": bool(s.get("imap_starttls", True)),
-                "imap_starttls": bool(s.get("imap_starttls", True)),
+            "smtp_host": smtp_host,
-                "smtp_host": smtp_host,
+            "smtp_port": int(s.get("smtp_port") or 465),
-                "smtp_port": int(s.get("smtp_port") or 465),
+            "smtp_user": s.get("smtp_user") or "",
-                "smtp_user": s.get("smtp_user") or "",
+            "smtp_password": s.get("smtp_password") or "",
-                "smtp_password": s.get("smtp_password") or "",
+            "from_address": s.get("email_from") or "",
-                "from_address": s.get("email_from") or "",
+            "created_at": now,
-                "created_at": now,
+            "updated_at": now,
-                "updated_at": now,
+        })
-            })
+        db.commit()
-            logging.getLogger(__name__).info("Seeded email_accounts 'Default' from settings.json")
+        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()

docs/setup.md

@@ -309,6 +309,32 @@ container. Cookbook **Serve** is a separate workflow for serving downloaded
 models through Odysseus/llama.cpp, so Windows users with an existing Ollama
 install usually only need to add the endpoint in Settings.
 
+**Tool calls not firing on a manually-added Ollama `/v1` endpoint.** By
+design, a local Ollama `/v1` endpoint defaults to the conservative
+text-based (fenced-block) tool-calling path rather than native structured
+tool calls, since some locally-served models mishandle native schemas (see
+#1567). This is correct for most local setups, but if you know your specific
+model reliably supports native tool calling (check `ollama show <model>` for
+`tools` under Capabilities), you can opt that endpoint in explicitly. There
+is currently no UI control for this on manually-added endpoints (see #5192);
+the flag can still be set directly against the existing API, from a browser
+console on an authenticated admin session:
+
+```js
+fetch('/api/model-endpoints/<endpoint-id>', {
+  method: 'PATCH',
+  credentials: 'same-origin',
+  headers: {'Content-Type': 'application/json'},
+  body: JSON.stringify({supports_tools: true})
+}).then(r => r.json()).then(console.log)
+```
+
+Find `<endpoint-id>` by inspecting the `/api/model-endpoints` response (or
+your browser's network tab while Settings loads the endpoint list). Send
+`supports_tools: false` to disable native structured tool calls and force the
+conservative fenced/text path, or `supports_tools: null` to return the endpoint
+to the Auto heuristic.
+
 **Useful checks.**
 
 ```bash

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/calendar_routes.py

@@ -10,6 +10,7 @@ from typing import Optional, List
 from fastapi import APIRouter, HTTPException, Request, UploadFile, File
 from pydantic import BaseModel
 from sqlalchemy import or_, and_
+from sqlalchemy.exc import IntegrityError
 from dateutil.rrule import rrulestr
 
 from core.database import SessionLocal, CalendarCal, CalendarDeletedEvent, CalendarEvent
@@ -221,22 +222,125 @@ class EventUpdate(BaseModel):
 
 # ── Helpers ──
 
+_DEFAULT_CALENDAR_NAMESPACE = uuid.UUID("4840613a-9847-4a3b-bd75-19e6bc5fc3ce")
+
+
+def _default_calendar_id(owner: str, collision_index: int = 0) -> str:
+    """Return one stable primary-key candidate for an owner's lazy default.
+
+    Slot zero preserves the original owner-derived identifier.  Later slots
+    let a username be reused after its prior calendar was migrated to another
+    owner during a rename, without making concurrent first use choose random
+    and therefore divergent identifiers.
+    """
+    if collision_index == 0:
+        candidate_name = owner
+    else:
+        candidate_name = json.dumps(
+            [owner, collision_index],
+            ensure_ascii=False,
+            separators=(",", ":"),
+        )
+    return str(uuid.uuid5(_DEFAULT_CALENDAR_NAMESPACE, candidate_name))
+
+
+def _begin_sqlite_default_write(db) -> None:
+    """Serialize an absent-default check with other SQLite writers.
+
+    SQLite's default deferred transactions allow two workers to both read an
+    empty calendar set before either writes.  ``BEGIN IMMEDIATE`` acquires the
+    writer reservation before the second, authoritative lookup.  We issue it
+    only when the driver has not already opened a write transaction; a caller
+    with a pending write already owns the required reservation.
+    """
+    connection = db.connection()
+    dbapi_connection = connection.connection
+    driver_connection = getattr(
+        dbapi_connection,
+        "driver_connection",
+        dbapi_connection,
+    )
+    if not getattr(driver_connection, "in_transaction", False):
+        connection.exec_driver_sql("BEGIN IMMEDIATE")
+
+
 def _ensure_default_calendar(db, owner: str = None) -> CalendarCal:
-    """Create default calendar if none exist for this owner."""
+    """Return the owner's calendar, staging a default in the caller's transaction.
+
+    A stable owner-derived primary key makes concurrent first-use inserts
+    converge on one row on every SQL backend.  SQLite additionally serializes
+    the absent-row check because its deferred transactions otherwise permit
+    both workers to read the gap before either writes.  Other backends recover
+    a lost insert race inside a savepoint so the caller's event transaction
+    remains usable and atomic.
+    """
     owner = owner or FALLBACK_OWNER
     cal = db.query(CalendarCal).filter(CalendarCal.owner == owner).first()
-    if not cal:
+    if cal:
+        return cal
+
+    dialect = db.get_bind().dialect.name
+    if dialect == "sqlite":
+        _begin_sqlite_default_write(db)
+        # Another worker may have committed while BEGIN IMMEDIATE waited.
+        cal = db.query(CalendarCal).filter(CalendarCal.owner == owner).first()
+        if cal:
+            return cal
+
+    collision_index = 0
+    while True:
+        default_id = _default_calendar_id(owner, collision_index)
+
+        if dialect == "sqlite":
+            # BEGIN IMMEDIATE above makes this occupancy check authoritative:
+            # another SQLite writer cannot rename, delete, or claim this slot
+            # until the caller commits or rolls back.
+            occupant = db.query(CalendarCal).filter(
+                CalendarCal.id == default_id,
+            ).first()
+            if occupant is not None:
+                if occupant.owner == owner:
+                    return occupant
+                collision_index += 1
+                continue
+
         cal = CalendarCal(
-            id=str(uuid.uuid4()),
+            id=default_id,
             owner=owner,
             name="Personal",
             color="#5b8abf",
             source="local",
         )
-        db.add(cal)
+
-        db.commit()
+        if dialect == "sqlite":
-        db.refresh(cal)
+            db.add(cal)
-    return cal
+            db.flush()
+            return cal
+
+        try:
+            # A uniqueness failure rolls back only this savepoint, not an event
+            # or reminder already staged by the caller's outer transaction.
+            with db.begin_nested():
+                db.add(cal)
+                db.flush()
+            return cal
+        except IntegrityError:
+            # Use a locking/current read so repeatable-read backends can observe
+            # the row that won after our transaction's original empty snapshot.
+            occupant = db.query(CalendarCal).filter(
+                CalendarCal.id == default_id,
+            ).with_for_update().first()
+            if occupant is None:
+                # Do not misclassify an unrelated integrity failure as an ID
+                # collision and loop forever. A concurrently deleted winner is
+                # safe for the caller to retry as a fresh transaction.
+                raise
+            if occupant.owner == owner:
+                return occupant
+            # A renamed calendar owns this deterministic slot. Advance to the
+            # next stable slot; concurrent callers for this owner will still
+            # converge there.
+            collision_index += 1
 
 
 # Per-request user time context. chat_routes sets this from browser timezone
@@ -1015,6 +1119,9 @@ def setup_calendar_routes(upload_handler=None) -> APIRouter:
         db = SessionLocal()
         try:
             _ensure_default_calendar(db, owner)
+            # Listing calendars intentionally lazily creates a durable default.
+            # Other callers commit it with the event they are creating.
+            db.commit()
             cals = db.query(CalendarCal).filter(CalendarCal.owner == owner).all()
             return {"calendars": [
                 {"name": c.name, "href": c.id, "color": c.color, "source": c.source}
@@ -1023,6 +1130,7 @@ def setup_calendar_routes(upload_handler=None) -> APIRouter:
         except HTTPException:
             raise
         except Exception as e:
+            db.rollback()
             logger.error("Failed to list calendars: %s", e)
             raise HTTPException(500, "Failed to list calendars")
         finally:

routes/cookbook_routes.py

@@ -73,6 +73,30 @@ _HF_TOKEN_STATUS_SNIPPET = (
 )
 
 
+def _windows_local_pid_record_line(pid_path: Path, ready_path: Path) -> str:
+    """Build the Git Bash prelude that records a Win32-stoppable PID.
+
+    Python publishes the detached outer process's Win32 PID first, then touches
+    ``ready_path``. The inner Git Bash runner waits for that publication before
+    replacing the fallback with its own Win32 PID from /proc/<msys-pid>/winpid.
+
+    Missing, malformed, or late mappings leave the valid outer PID untouched.
+    """
+    pp = shlex.quote(pid_path.as_posix())
+    rp = shlex.quote(ready_path.as_posix())
+    return (
+        "i=0; "
+        f"while [ ! -e {rp} ] && [ \"$i\" -lt 500 ]; do "
+        "i=$((i+1)); sleep 0.01; done; "
+        f"if [ -e {rp} ]; then "
+        "winpid=\"$(cat /proc/$$/winpid 2>/dev/null || true)\"; "
+        "case \"$winpid\" in ''|*[!0-9]*) ;; "
+        f"*) printf '%s\\n' \"$winpid\" > {pp} ;; esac; "
+        "fi; "
+        f"rm -f {rp}"
+    )
+
+
 def _append_mlx_image_server_script(runner_lines: list[str]) -> None:
     """Write the MLX image API helper next to the tmux runner on remote hosts."""
     script_path = Path(__file__).resolve().parents[1] / "scripts" / "mlx_image_server.py"
@@ -978,15 +1002,18 @@ def setup_cookbook_routes() -> APIRouter:
         directly (simple commands only). Returns the launched job record."""
         log_path = TMUX_LOG_DIR / f"{session_id}.log"
         pid_path = TMUX_LOG_DIR / f"{session_id}.pid"
+        pid_ready_path: Path | None = None
         bash = find_bash()
         if bash:
             # Run the existing bash wrapper verbatim through Git Bash, redirecting
             # all output to the log the poller reads. Paths handed to bash use
             # POSIX form + shell-quoting so drive paths / spaces survive.
             inner = TMUX_LOG_DIR / f"{session_id}_run.sh"
-            pp = shlex.quote(pid_path.as_posix())
+            pid_ready_path = TMUX_LOG_DIR / f"{session_id}.pid.ready"
+            pid_ready_path.unlink(missing_ok=True)
             inner.write_text(
-                f"printf '%s\\n' \"$$\" > {pp}\n" + "\n".join(bash_lines) + "\n",
+                _windows_local_pid_record_line(pid_path, pid_ready_path) + "\n"
+                + "\n".join(bash_lines) + "\n",
                 encoding="utf-8",
             )
             lp = shlex.quote(log_path.as_posix())
@@ -1020,7 +1047,18 @@ def setup_cookbook_routes() -> APIRouter:
             env=env,
             **detached_popen_kwargs(),
         )
+        # Publish a valid Win32 ancestor first. The Git Bash runner may then
+        # replace it with its own Win32 pid, but never before this fallback exists.
         pid_path.write_text(str(proc.pid), encoding="utf-8")
+        if pid_ready_path is not None:
+            try:
+                pid_ready_path.touch()
+            except OSError as e:
+                logger.warning(
+                    "Could not publish Windows local PID handoff for %s: %s",
+                    session_id,
+                    e,
+                )
         return {"pid": proc.pid, "log_path": str(log_path)}
 
     @router.post("/api/model/download")

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,
@@ -195,6 +196,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:
@@ -5487,9 +5546,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 = _email_account_owner_scope(q, owner)
-                q = q.filter(EmailAccount.owner == owner)
             row = q.first()
             if row is None:
                 row = EmailAccount(id=_uuid.uuid4().hex, owner=owner, name="Default", is_default=True, enabled=True)
@@ -5515,8 +5574,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 = _email_account_owner_scope(clear_q, owner)
-                clear_q = clear_q.filter(EmailAccount.owner == owner)
             clear_q.update({EmailAccount.is_default: False})
             db.commit()
         finally:
@@ -5611,6 +5669,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,
@@ -5637,9 +5696,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)
+            scope_q = _email_account_owner_scope(db.query(EmailAccount), owner)
-            if owner:
-                scope_q = scope_q.filter(EmailAccount.owner == owner)
             existing_count = scope_q.count()
             if row.is_default or existing_count == 0:
                 scope_q.update({EmailAccount.is_default: False})
@@ -5690,28 +5747,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)
+            row = _lock_and_reload_email_account(
-            if not row:
+                db, account_id, owner, initial_scope
-                return {"ok": False, "error": "Account not found"}
+            )
+            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
+                promote_q = db.query(EmailAccount).filter(
-                if owner:
+                    EmailAccount.id != account_id,
-                    promote_q = promote_q.filter(EmailAccount.owner == owner)
+                    EmailAccount.enabled == True,  # noqa: E712
-                promote = promote_q.order_by(EmailAccount.created_at.asc()).first()
+                )
+                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()
@@ -5924,18 +5992,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)
+            row = _lock_and_reload_email_account(
-            if not row:
+                db, account_id, owner, initial_scope
-                return {"ok": False, "error": "Account not found"}
+            )
-            # SECURITY: scope the "clear other defaults" sweep to this user's
+            # Scope the sweep to the target row's normalized owner partition;
-            # accounts so we don't unset another user's default flag.
+            # this also handles visible legacy NULL/empty-owner accounts.
-            clear_q = db.query(EmailAccount)
+            clear_q = _email_account_owner_scope(
-            if owner:
+                db.query(EmailAccount), row.owner or ""
-                clear_q = clear_q.filter(EmailAccount.owner == owner)
+            )
             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:
+def _owner_scope(query, owner: str):
-    Base.metadata.create_all(bind=engine)
+    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(
+        return {
-            EmailAccount.name == NAME, EmailAccount.imap_user == IMAP_USER
+            row.owner or ""
-        ).first()
+            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(
+        rows = _lock_and_load_demo_rows(db, scopes)
-            EmailAccount.name == NAME, EmailAccount.imap_user == IMAP_USER
+        deleted_ids = [row.id for row in rows]
-        ).all()
+        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

src/builtin_actions.py

@@ -20,6 +20,395 @@ from src.interactive_gate import wait_for_interactive_quiet
 logger = logging.getLogger(__name__)
 
 
+def _read_email_urgency_state(state_path):
+    """Read one atomic urgency checkpoint, tolerating the legacy shape."""
+    from pathlib import Path
+
+    state_path = Path(state_path)
+    try:
+        state = (
+            json.loads(state_path.read_text(encoding="utf-8"))
+            if state_path.exists()
+            else {}
+        )
+    except Exception:
+        return {}
+    return state if isinstance(state, dict) else {}
+
+
+def _email_urgency_account_generations(state):
+    """Return normalized per-account checkpoint/complete generations.
+
+    Checkpoint generations fence every accepted state mutation. Complete
+    generations advance only for a non-stale complete scan. Missing metadata
+    is the legacy generation zero.
+    """
+    raw = state.get("account_generations", {}) if isinstance(state, dict) else {}
+    if not isinstance(raw, dict):
+        return {}
+
+    generations = {}
+    for account_id, value in raw.items():
+        if isinstance(value, dict):
+            checkpoint = value.get("checkpoint", 0)
+            complete = value.get("complete", 0)
+        else:
+            # Tolerate an intermediate scalar representation as one completed
+            # checkpoint generation instead of discarding its fence.
+            checkpoint = value
+            complete = value
+        try:
+            checkpoint = max(0, int(checkpoint))
+        except (TypeError, ValueError):
+            checkpoint = 0
+        try:
+            complete = max(0, int(complete))
+        except (TypeError, ValueError):
+            complete = 0
+        generations[str(account_id)] = {
+            "checkpoint": checkpoint,
+            "complete": complete,
+        }
+    return generations
+
+
+def _email_urgency_string_set(value):
+    if not isinstance(value, (list, tuple, set, frozenset)):
+        return set()
+    return {str(item) for item in value if isinstance(item, (str, int))}
+
+
+def _acquire_email_urgency_state_lock(
+    state_path,
+    lock_db_path,
+    cancel_event,
+    timeout_seconds=120,
+):
+    """Acquire the cross-process urgency lock without blocking the app loop."""
+    import sqlite3
+    import time
+    from pathlib import Path
+
+    state_path = Path(state_path)
+    state_path.parent.mkdir(parents=True, exist_ok=True)
+    deadline = time.monotonic() + timeout_seconds
+
+    while not cancel_event.is_set():
+        remaining = deadline - time.monotonic()
+        if remaining <= 0:
+            raise sqlite3.OperationalError("timed out waiting for urgency state lock")
+        conn = sqlite3.connect(
+            str(lock_db_path),
+            timeout=min(0.25, max(0.01, remaining)),
+            check_same_thread=False,
+        )
+        try:
+            conn.execute("BEGIN IMMEDIATE")
+        except sqlite3.OperationalError as exc:
+            conn.close()
+            if "locked" not in str(exc).lower():
+                raise
+            cancel_event.wait(min(0.05, max(0.0, remaining)))
+            continue
+        except BaseException:
+            conn.close()
+            raise
+
+        if cancel_event.is_set():
+            conn.rollback()
+            conn.close()
+            return None, None
+        return conn, _read_email_urgency_state(state_path)
+
+    return None, None
+
+
+def _close_email_urgency_state_lock(conn):
+    if conn is None:
+        return
+    try:
+        try:
+            conn.rollback()
+        except Exception:
+            pass
+    finally:
+        conn.close()
+
+
+def _commit_email_urgency_state(conn, state_path, next_state):
+    """Atomically publish JSON before releasing the SQLite write lock."""
+    import uuid
+    from pathlib import Path
+
+    state_path = Path(state_path)
+    temp_path = state_path.with_name(
+        f".{state_path.name}.{uuid.uuid4().hex}.tmp"
+    )
+    try:
+        temp_path.write_text(json.dumps(next_state), encoding="utf-8")
+        temp_path.replace(state_path)
+        conn.commit()
+    except BaseException:
+        conn.rollback()
+        raise
+    finally:
+        temp_path.unlink(missing_ok=True)
+        conn.close()
+
+
+async def _run_email_urgency_state_transaction(
+    state_path,
+    lock_db_path,
+    operation,
+):
+    """Serialize one urgency decision while keeping async work on this loop.
+
+    Only lock acquisition waits in a worker thread. ``operation`` is awaited
+    on the caller's long-lived event loop, where shared async clients, locks,
+    and the browser-notification queue belong. Cancellation rolls back the
+    SQLite transaction and never publishes a checkpoint.
+    """
+    import asyncio
+    import threading
+
+    loop = asyncio.get_running_loop()
+    cancel_event = threading.Event()
+    acquire_future = loop.run_in_executor(
+        None,
+        _acquire_email_urgency_state_lock,
+        state_path,
+        lock_db_path,
+        cancel_event,
+    )
+    try:
+        conn, prior = await asyncio.shield(acquire_future)
+    except asyncio.CancelledError as cancelled:
+        cancel_event.set()
+        # The acquisition worker owns any connection until it returns. Wait
+        # for its short busy-poll to observe cancellation, then close a lock it
+        # may have won concurrently with the cancellation request.
+        while True:
+            try:
+                conn, _prior = await asyncio.shield(acquire_future)
+                break
+            except asyncio.CancelledError:
+                continue
+            except Exception:
+                conn = None
+                break
+        _close_email_urgency_state_lock(conn)
+        raise cancelled
+
+    if conn is None:
+        raise asyncio.CancelledError
+
+    try:
+        result, next_state = await operation(prior)
+        # Keep this small atomic publish synchronous. There is no await between
+        # the successful operation and commit, so cancellation cannot be
+        # observed and then followed by a checkpoint.
+        try:
+            _commit_email_urgency_state(conn, state_path, next_state)
+        finally:
+            conn = None
+        return result
+    except BaseException:
+        _close_email_urgency_state_lock(conn)
+        raise
+
+
+def _email_urgency_account_key(message_key):
+    return str(message_key).split(":", 1)[0]
+
+
+def _email_urgency_payload_account_ids(state):
+    """Return account IDs that still own user-visible urgency payload."""
+    if not isinstance(state, dict):
+        return set()
+
+    per_uid = state.get("per_uid", {})
+    per_uid_keys = per_uid if isinstance(per_uid, dict) else {}
+    return {
+        _email_urgency_account_key(key) for key in per_uid_keys
+    } | {
+        _email_urgency_account_key(key)
+        for key in _email_urgency_string_set(state.get("notified_uids", []))
+    }
+
+
+def _email_urgency_known_account_ids(state):
+    """Return payload owners plus generation-only active/retired markers."""
+    return _email_urgency_payload_account_ids(state) | set(
+        _email_urgency_account_generations(state)
+    )
+
+
+def _email_urgency_stale_accounts(
+    prior,
+    base_account_generations,
+    account_ids,
+):
+    prior_generations = _email_urgency_account_generations(prior)
+    base_generations = _email_urgency_account_generations(
+        {"account_generations": base_account_generations}
+    )
+    return {
+        str(account_id)
+        for account_id in account_ids
+        if prior_generations.get(str(account_id), {}).get("checkpoint", 0)
+        != base_generations.get(str(account_id), {}).get("checkpoint", 0)
+    }
+
+
+def _merge_email_urgency_state(
+    prior,
+    *,
+    owner,
+    per_uid_scores,
+    notified_uids,
+    all_unread_keys,
+    fully_scanned_account_ids,
+    base_account_generations,
+    timestamp,
+    retired_account_ids=(),
+    base_payload_account_ids=(),
+    known_account_ids=(),
+):
+    """Merge a scan without letting an older snapshot erase newer facts."""
+    prior_per_uid = prior.get("per_uid", {})
+    if not isinstance(prior_per_uid, dict):
+        prior_per_uid = {}
+    complete = {str(account_id) for account_id in fully_scanned_account_ids}
+    prior_generations = _email_urgency_account_generations(prior)
+    retire_requested = {str(account_id) for account_id in retired_account_ids}
+    observed_accounts = {
+        _email_urgency_account_key(key) for key in per_uid_scores
+    } | complete | retire_requested
+    stale_accounts = _email_urgency_stale_accounts(
+        prior,
+        base_account_generations,
+        observed_accounts,
+    )
+    prior_payload_accounts = _email_urgency_payload_account_ids(prior)
+    base_payload_accounts = {
+        str(account_id) for account_id in base_payload_account_ids
+    }
+    # A selected account can be absent from the base snapshot. If another
+    # worker creates its first payload before this transaction wins the lock,
+    # membership itself is a fence even when both snapshots normalize to the
+    # legacy generation zero.
+    retired_accounts = {
+        account_id
+        for account_id in retire_requested - stale_accounts
+        if not (
+            account_id in prior_payload_accounts
+            and account_id not in base_payload_accounts
+        )
+    }
+    fresh_complete = complete - stale_accounts - retired_accounts
+    changed_accounts = set(fresh_complete)
+
+    merged_per_uid = {
+        key: value
+        for key, value in prior_per_uid.items()
+        if _email_urgency_account_key(key) not in retired_accounts
+    }
+    for key in list(merged_per_uid):
+        account_id = _email_urgency_account_key(key)
+        if account_id in fresh_complete:
+            merged_per_uid.pop(key, None)
+            changed_accounts.add(account_id)
+    # Partial scans may add or refresh facts, but absence from a partial scan
+    # is not evidence that another checkpoint or UI row is stale. When another
+    # worker committed after this scan captured its base generation, discard
+    # this account's whole stale snapshot. A key absent from the newer state
+    # may have been removed/read, so even a stale-only key is not safely
+    # additive without another fresh scan.
+    for key, value in per_uid_scores.items():
+        account_id = _email_urgency_account_key(key)
+        if account_id in stale_accounts or account_id in retired_accounts:
+            continue
+        if merged_per_uid.get(key) != value:
+            changed_accounts.add(account_id)
+        merged_per_uid[key] = value
+
+    prior_notified = _email_urgency_string_set(prior.get("notified_uids", []))
+    merged_notified = {
+        key
+        for key in prior_notified
+        if _email_urgency_account_key(key) not in retired_accounts
+    }
+    for key in _email_urgency_string_set(notified_uids) - prior_notified:
+        account_id = _email_urgency_account_key(key)
+        if account_id in stale_accounts or account_id in retired_accounts:
+            continue
+        merged_notified.add(key)
+        changed_accounts.add(account_id)
+    for key in list(merged_notified):
+        if (
+            _email_urgency_account_key(key) in fresh_complete
+            and key not in all_unread_keys
+        ):
+            merged_notified.discard(key)
+            changed_accounts.add(_email_urgency_account_key(key))
+
+    next_generations = {
+        account_id: dict(value)
+        for account_id, value in prior_generations.items()
+    }
+    for account_id in changed_accounts:
+        generation = next_generations.setdefault(
+            account_id,
+            {"checkpoint": 0, "complete": 0},
+        )
+        generation["checkpoint"] += 1
+        if account_id in fresh_complete:
+            generation["complete"] += 1
+    for account_id in {str(value) for value in known_account_ids}:
+        next_generations.setdefault(
+            account_id,
+            {"checkpoint": 0, "complete": 0},
+        )
+    for account_id in retired_accounts:
+        # Every authoritative absence advances its generation, even when the
+        # prior state is already a payload-empty tombstone. A re-enabled scan
+        # may have captured that previous tombstone immediately before the
+        # account was disabled/deleted again; monotonic advancement is what
+        # makes that in-flight scan stale.
+        generation = next_generations.setdefault(
+            account_id,
+            {"checkpoint": 0, "complete": 0},
+        )
+        generation["checkpoint"] += 1
+
+    total_unread = 0
+    total_urgent = 0
+    max_score = 0
+    for value in merged_per_uid.values():
+        if not isinstance(value, dict):
+            continue
+        try:
+            score = max(0, min(3, int(value.get("score", 0))))
+        except (TypeError, ValueError):
+            score = 0
+        max_score = max(max_score, score)
+        if value.get("unread"):
+            total_unread += 1
+            if score >= 2:
+                total_urgent += 1
+
+    return {
+        "ts": timestamp,
+        "owner": owner or "",
+        "total_unread": total_unread,
+        "total_urgent": total_urgent,
+        "max_score": max_score,
+        "per_uid": merged_per_uid,
+        "notified_uids": sorted(merged_notified),
+        "account_generations": next_generations,
+    }
+
+
 class TaskNoop(BaseException):
     """Raised by an action when it determined there's nothing to do.
 
@@ -1878,6 +2267,7 @@ async def action_check_email_urgency(owner: str, **kwargs) -> Tuple[str, bool]:
         # filename for single-user installs (matches prior behaviour).
         _owner_slug = "".join(c if (c.isalnum() or c in "-_.@") else "_" for c in (owner or "default"))
         STATE_PATH = _P(DATA_DIR) / f"email_urgency_state_{_owner_slug}.json"
+        STATE_LOCK_DB = STATE_PATH.with_suffix(".lock.sqlite3")
         CACHE_DIR = _P(EMAIL_URGENCY_CACHE_DIR)
         CACHE_DIR.mkdir(parents=True, exist_ok=True)
         STATE_PATH.parent.mkdir(parents=True, exist_ok=True)
@@ -1892,35 +2282,144 @@ async def action_check_email_urgency(owner: str, **kwargs) -> Tuple[str, bool]:
             "shopping", "social", "work", "personal", "legal", "support", "promo",
         }
 
-        # ── 1. Resolve LLM candidates (utility primary + utility fallbacks; fall
+        # Resolve with the task owner as before, but defer the availability
-        # through to default chat as a last resort).
+        # gate until after authoritative account cleanup. State retirement must
+        # still run when no model is configured.
         from src.task_endpoint import resolve_task_candidates
         candidates = resolve_task_candidates(owner=owner)
-        if not candidates:
-            return "No LLM endpoint available", False
-
         target_account_id = _email_task_account_id(kwargs)
 
-        # ── 2. Enumerate enabled accounts. Match this task's owner AND fall
+        # ── 1. Enumerate enabled accounts. Match this task's owner AND fall
         # back to the legacy "unowned account whose imap_user / from_address
         # == this owner" pattern — same rule `_get_email_config` uses, so a
         # pre-multi-user account row still gets picked up for the seeded task.
-        db = _SL()
+        def _enumerate_enabled_accounts():
-        try:
+            db = _SL()
-            from sqlalchemy import and_ as _and, or_ as _or
+            try:
-            q = db.query(_EA).filter(_EA.enabled == True)  # noqa: E712
+                from sqlalchemy import and_ as _and, or_ as _or
-            if owner:
+                q = db.query(_EA).filter(_EA.enabled == True)  # noqa: E712
-                unowned = _or(_EA.owner == None, _EA.owner == "")  # noqa: E711
+                if owner:
-                same_mailbox = _or(_EA.imap_user == owner, _EA.from_address == owner)
+                    unowned = _or(_EA.owner == None, _EA.owner == "")  # noqa: E711
-                q = q.filter(_or(_EA.owner == owner, _and(unowned, same_mailbox)))
+                    same_mailbox = _or(
-            if target_account_id:
+                        _EA.imap_user == owner,
-                q = q.filter(_EA.id == target_account_id)
+                        _EA.from_address == owner,
-            accounts = q.all()
+                    )
-        finally:
+                    q = q.filter(
-            db.close()
+                        _or(_EA.owner == owner, _and(unowned, same_mailbox))
+                    )
+                if target_account_id:
+                    q = q.filter(_EA.id == target_account_id)
+                return q.all()
+            finally:
+                db.close()
+
+        initial_accounts = _enumerate_enabled_accounts()
+        initial_account_ids = {
+            str(account.id) for account in initial_accounts
+        }
+
+        # Register every account before IMAP work, including its first-ever
+        # scan. A concurrent zero-account cleanup can then advance this marker
+        # and fence delivery even before the scan has produced payload.
+        registered_state = None
+        if initial_account_ids:
+            async def _register_accounts(prior):
+                next_state = _merge_email_urgency_state(
+                    prior,
+                    owner=owner,
+                    per_uid_scores={},
+                    notified_uids=prior.get("notified_uids", []),
+                    all_unread_keys=set(),
+                    fully_scanned_account_ids=set(),
+                    base_account_generations=(
+                        _email_urgency_account_generations(prior)
+                    ),
+                    timestamp=_time.time(),
+                    known_account_ids=initial_account_ids,
+                )
+                # Return the exact state committed by registration. This is
+                # the scan's generation token: adopting a later checkpoint
+                # after account cleanup would let the stale scan appear fresh.
+                return next_state, next_state
+
+            registered_state = await _run_email_urgency_state_transaction(
+                STATE_PATH,
+                STATE_LOCK_DB,
+                _register_accounts,
+            )
+
+        # Revalidate after registration. If deletion/disable and its cleanup
+        # completed before the marker was published, this second enumeration
+        # observes the absence and this action retires its own marker instead
+        # of starting IMAP. Accounts newly appearing between the two reads are
+        # left for the next pass rather than scanned without prior registration.
+        verified_accounts = _enumerate_enabled_accounts()
+        enabled_account_ids = {
+            str(account.id) for account in verified_accounts
+        }
+        accounts = [
+            account
+            for account in verified_accounts
+            if str(account.id) in initial_account_ids
+        ]
+
+        # Capture the checkpoint basis before cleanup or IMAP. A full
+        # owner-wide enumeration authoritatively retires all known state IDs
+        # absent from the current enabled/visible set. A scoped task may retire
+        # only its selected missing/disabled account. Existing accounts remain
+        # present even if their later network scan fails, so transient IMAP
+        # failure never erases their last known state.
+        base_state = (
+            registered_state
+            if registered_state is not None
+            else _read_email_urgency_state(STATE_PATH)
+        )
+        base_account_generations = _email_urgency_account_generations(
+            base_state
+        )
+        base_payload_account_ids = _email_urgency_payload_account_ids(base_state)
+        known_state_account_ids = _email_urgency_known_account_ids(base_state)
+        if target_account_id:
+            retired_account_ids = (
+                {str(target_account_id)}
+                if str(target_account_id) not in enabled_account_ids
+                else set()
+            )
+        else:
+            retired_account_ids = (
+                known_state_account_ids - enabled_account_ids
+            )
+
+        if retired_account_ids:
+            async def _retire_accounts(prior):
+                next_state = _merge_email_urgency_state(
+                    prior,
+                    owner=owner,
+                    per_uid_scores={},
+                    notified_uids=prior.get("notified_uids", []),
+                    all_unread_keys=set(),
+                    fully_scanned_account_ids=set(),
+                    base_account_generations=base_account_generations,
+                    timestamp=_time.time(),
+                    retired_account_ids=retired_account_ids,
+                    base_payload_account_ids=base_payload_account_ids,
+                )
+                return None, next_state
+
+            await _run_email_urgency_state_transaction(
+                STATE_PATH,
+                STATE_LOCK_DB,
+                _retire_accounts,
+            )
         if not accounts:
             raise TaskNoop("no email accounts configured")
 
+        # ── 2. Account retirement above is state maintenance and does not
+        # depend on model availability. Scanning still requires the utility
+        # primary/fallback candidates resolved for this task owner.
+        if not candidates:
+            return "No LLM endpoint available", False
+
         urgency_prompt = settings.get("urgent_email_prompt", "")
         per_uid_scores = {}   # key = "<acc_id>:<uid>" → {"score": 0-3, "reason": "..."}
         all_unread_keys = set()
@@ -1929,6 +2428,7 @@ async def action_check_email_urgency(owner: str, **kwargs) -> Tuple[str, bool]:
         failed_classifications = []
         tag_write_details = []
         scanned = 0
+        fully_scanned_account_ids = set()
 
         def _heuristic_email_verdict(item: dict) -> dict:
             blob = (
@@ -2024,16 +2524,27 @@ async def action_check_email_urgency(owner: str, **kwargs) -> Tuple[str, bool]:
             def _scan_one(account=acc, cache_uids=cache.get("uids", {})):
                 """Sync IMAP work runs in a thread."""
                 results = []
+                scan_complete = True
                 conn = _imap_connect(account.id)
                 try:
-                    conn.select("INBOX", readonly=True)
+                    select_status, _select_data = conn.select("INBOX", readonly=True)
+                    if select_status != "OK":
+                        return results, False
                     # Tag recent inbox mail, not only unread mail. Urgency
                     # reminders below still only notify for unread messages.
                     since_str = AGE_CUTOFF.strftime("%d-%b-%Y")
                     status, data = conn.uid("SEARCH", None, f'(SINCE {since_str})')
-                    if status != "OK" or not data or not data[0]:
+                    if status != "OK":
-                        return results
+                        return results, False
-                    uids = data[0].split()[-30:]
+                    if not data or not data[0]:
+                        return results, True
+                    matching_uids = data[0].split()
+                    if len(matching_uids) > 30:
+                        # The scale guard deliberately processes only the most
+                        # recent 30. That is a partial account snapshot, so it
+                        # cannot justify pruning older checkpoint facts.
+                        scan_complete = False
+                    uids = matching_uids[-30:]
                     for uid_b in uids:
                         uid = uid_b.decode() if isinstance(uid_b, bytes) else str(uid_b)
                         key = f"{account.id}:{uid}"
@@ -2041,12 +2552,41 @@ async def action_check_email_urgency(owner: str, **kwargs) -> Tuple[str, bool]:
                         cached_ok = isinstance(cached, dict) and cached.get("triage_version") == TRIAGE_VERSION
                         results.append({"key": key, "uid": uid, "cached": cached if cached_ok else None})
                         if cached_ok:
-                            # Already classified — skip the fetch.
+                            # Cached verdicts still need a lightweight FLAGS
+                            # refresh. Without it a cached unread message looks
+                            # read and its successful notification checkpoint
+                            # is pruned on the next pass.
+                            try:
+                                st, flag_data = conn.uid("FETCH", uid_b, "(UID FLAGS)")
+                                if st != "OK" or not flag_data:
+                                    scan_complete = False
+                                    results.pop()
+                                    continue
+                                flag_parts = []
+                                for part in flag_data:
+                                    if isinstance(part, (bytes, bytearray)):
+                                        flag_parts.append(bytes(part))
+                                    elif (
+                                        isinstance(part, tuple)
+                                        and part
+                                        and isinstance(part[0], (bytes, bytearray))
+                                    ):
+                                        flag_parts.append(bytes(part[0]))
+                                flags_blob = b" ".join(flag_parts)
+                                results[-1]["unread"] = b"\\Seen" not in flags_blob
+                            except Exception as _fe:
+                                scan_complete = False
+                                results.pop()
+                                logger.debug(
+                                    f"urgency: flag fetch for uid {uid} failed: {_fe}"
+                                )
                             continue
                         # Pull headers + first ~800 chars of plaintext body.
                         try:
                             st, msg_data = conn.uid("FETCH", uid_b, "(UID FLAGS RFC822.HEADER BODY.PEEK[TEXT]<0.800>)")
                             if st != "OK" or not msg_data:
+                                scan_complete = False
+                                results.pop()
                                 continue
                             flags_blob = b" ".join(
                                 part[0] for part in msg_data
@@ -2060,6 +2600,8 @@ async def action_check_email_urgency(owner: str, **kwargs) -> Tuple[str, bool]:
                                 if isinstance(part, tuple) and part[1]:
                                     raw += part[1] + b"\n\n"
                             if not raw:
+                                scan_complete = False
+                                results.pop()
                                 continue
                             msg = _email_mod.message_from_bytes(raw)
                             # Skip Odysseus-generated reminders so the scanner
@@ -2115,17 +2657,21 @@ async def action_check_email_urgency(owner: str, **kwargs) -> Tuple[str, bool]:
                                 "unread": is_unread,
                             })
                         except Exception as _fe:
+                            scan_complete = False
+                            results.pop()
                             logger.debug(f"urgency: header fetch for uid {uid} failed: {_fe}")
                 finally:
                     try: conn.logout()
                     except Exception: pass
-                return results
+                return results, scan_complete
 
             try:
-                items = await _aio.to_thread(_scan_one)
+                items, scan_complete = await _aio.to_thread(_scan_one)
             except Exception as e:
                 logger.warning(f"urgency: IMAP scan failed for account {acc.id}: {e}")
                 continue
+            if scan_complete:
+                fully_scanned_account_ids.add(str(acc.id))
 
             for item in items:
                 scanned += 1
@@ -2262,13 +2808,13 @@ async def action_check_email_urgency(owner: str, **kwargs) -> Tuple[str, bool]:
                     logger.debug(f"urgency: LLM classify failed for {key}: {e}")
                     continue
 
-            # ── Prune cache entries for UIDs that are no longer in the recent
+            if scan_complete:
-            # scan window. Read messages remain cached because tags are useful
+                # Only a complete account scan proves a cached UID left the
-            # on read mail too; unread state is refreshed per scan above.
+                # recent window. Partial/failing scans preserve prior facts.
-            seen_uids = {it["uid"] for it in items}
+                seen_uids = {it["uid"] for it in items}
-            cache_uids = cache.get("uids", {})
+                cache_uids = cache.get("uids", {})
-            for stale in [u for u in cache_uids if u not in seen_uids]:
+                for stale in [u for u in cache_uids if u not in seen_uids]:
-                cache_uids.pop(stale, None)
+                    cache_uids.pop(stale, None)
 
             try:
                 cache_file.write_text(_json.dumps(cache), encoding="utf-8")
@@ -2372,40 +2918,34 @@ async def action_check_email_urgency(owner: str, **kwargs) -> Tuple[str, bool]:
 
         # ── 4. Aggregate state. urgent = score ≥ 2.
         urgent_keys = [k for k, v in per_uid_scores.items() if v.get("score", 0) >= 2 and v.get("unread")]
-        max_score = max((v.get("score", 0) for v in per_uid_scores.values()), default=0)
-        total_urgent = len(urgent_keys)
 
-        # Load prior state to know which urgent UIDs we've already notified.
+        # ── 5. Fire a reminder only when a previously-unnotified UID scores
-        try:
+        # urgent. The read, decision, delivery, and checkpoint are serialized
-            prior = _json.loads(STATE_PATH.read_text(encoding="utf-8")) if STATE_PATH.exists() else {}
+        # below so two scheduler workers cannot both act on the same stale
-        except Exception:
+        # state or overwrite each other's successful checkpoint.
-            prior = {}
-        notified_uids = set(prior.get("notified_uids", []))
-
-        # ── 5. Fire reminder ONLY when a previously-unnotified UID scores urgent.
-        new_urgent = [k for k in urgent_keys if k not in notified_uids]
         newly_notified = set()
         notify_failed = set()
-        if new_urgent:
+
-            title = "Urgent email" if total_urgent == 1 else f"{total_urgent} urgent emails"
+        def _urgency_reminder_payload(reminder_keys):
-            # Build a real listing — subject · sender · reason for each urgent
+            total = len(reminder_keys)
-            # one — so the reminder email tells you which messages to act on,
+            title = "Urgent email" if total == 1 else f"{total} urgent emails"
-            # not just "4 needing reply". Optional deep-link when the user has
-            # `app_public_url` configured in Settings (so the email row links
-            # straight into the Odysseus Email tab).
-            # Sort: highest-scored UIDs first; cap at 10 to keep the email tidy.
             sorted_urgent = sorted(
-                ((k, per_uid_scores[k]) for k in urgent_keys),
+                ((key, per_uid_scores[key]) for key in reminder_keys),
-                key=lambda kv: kv[1].get("score", 0), reverse=True,
+                key=lambda item: item[1].get("score", 0),
+                reverse=True,
             )[:10]
             _pub = (settings.get("app_public_url") or "").strip().rstrip("/")
             from urllib.parse import quote as _quote
-            lines = [f"{total_urgent} email" + ("" if total_urgent == 1 else "s") + " need an urgent reply:", ""]
+            lines = [
-            for i, (k, v) in enumerate(sorted_urgent, 1):
+                f"{total} email" + ("" if total == 1 else "s")
-                subj = (v.get("subject") or "(no subject)")[:160]
+                + " need an urgent reply:",
-                frm = v.get("from") or ""
+                "",
-                why = v.get("reason") or ""
+            ]
-                uid_for_link = str(k).split(":", 1)[-1]
+            for i, (key, value) in enumerate(sorted_urgent, 1):
+                subj = (value.get("subject") or "(no subject)")[:160]
+                frm = value.get("from") or ""
+                why = value.get("reason") or ""
+                uid_for_link = str(key).split(":", 1)[-1]
                 hash_link = f"#email={_quote('INBOX', safe='')}:{uid_for_link}"
                 open_link = f"{_pub}/{hash_link}" if _pub else hash_link
                 line = f"{i}. {subj}"
@@ -2415,57 +2955,94 @@ async def action_check_email_urgency(owner: str, **kwargs) -> Tuple[str, bool]:
                     line += f"  ·  {why}"
                 lines.append(line)
                 lines.append(f"   Open email: {open_link}")
-            if total_urgent > len(sorted_urgent):
+            if total > len(sorted_urgent):
                 lines.append("")
-                lines.append(f"…and {total_urgent - len(sorted_urgent)} more.")
+                lines.append(f"…and {total - len(sorted_urgent)} more.")
-            body = "\n".join(lines)
+            return title, "\n".join(lines)
-            try:
+
-                # Call dispatch_reminder DIRECTLY (no HTTP/auth roundtrip — the
+        async def _dispatch_urgency_reminder(reminder_keys):
-                # endpoint version 401's the background scheduler because it
+            # Call dispatch_reminder directly: a scheduler has no browser
-                # has no session cookie).
+            # session cookie with which to call the HTTP endpoint.
-                from routes.note_routes import dispatch_reminder
+            from routes.note_routes import dispatch_reminder
-                dispatch_result = await dispatch_reminder(
+            title, body = _urgency_reminder_payload(reminder_keys)
-                    title=title, note_body=body, note_id="urgent-email",
+            return await dispatch_reminder(
-                    owner=owner or "",
+                title=title,
-                )
+                note_body=body,
-                channel = (settings.get("reminder_channel") or "browser").strip().lower()
+                note_id="urgent-email",
-                delivered = bool(dispatch_result.get("browser_sent"))
+                owner=owner or "",
-                if channel == "email":
+            )
-                    delivered = bool(dispatch_result.get("email_sent"))
+
-                elif channel == "ntfy":
+        async def _dispatch_and_checkpoint(prior):
-                    delivered = bool(dispatch_result.get("ntfy_sent"))
+            notified_uids = _email_urgency_string_set(
-                elif channel == "webhook":
+                prior.get("notified_uids", [])
-                    delivered = bool(dispatch_result.get("webhook_sent"))
+            )
-                if delivered:
+            observed_accounts = {
-                    newly_notified.update(new_urgent)
+                _email_urgency_account_key(key) for key in per_uid_scores
-                else:
+            } | fully_scanned_account_ids
+            stale_accounts = _email_urgency_stale_accounts(
+                prior,
+                base_account_generations,
+                observed_accounts,
+            )
+            # Generation fencing must happen before delivery, not only during
+            # merge. A stale-only unread UID may have been removed, read, or
+            # downgraded by the newer completed scan.
+            deliverable_urgent = [
+                key
+                for key in urgent_keys
+                if _email_urgency_account_key(key) not in stale_accounts
+            ]
+            new_urgent = [
+                key
+                for key in deliverable_urgent
+                if key not in notified_uids
+            ]
+            if new_urgent:
+                try:
+                    dispatch_result = await _dispatch_urgency_reminder(
+                        deliverable_urgent
+                    )
+                    channel = (settings.get("reminder_channel") or "browser").strip().lower()
+                    delivered = bool(dispatch_result.get("browser_sent"))
+                    if channel == "email":
+                        delivered = bool(dispatch_result.get("email_sent"))
+                    elif channel == "ntfy":
+                        delivered = bool(dispatch_result.get("ntfy_sent"))
+                    elif channel == "webhook":
+                        delivered = bool(dispatch_result.get("webhook_sent"))
+                    if delivered:
+                        newly_notified.update(new_urgent)
+                        notified_uids.update(new_urgent)
+                    else:
+                        notify_failed.update(new_urgent)
+                        logger.warning(
+                            "urgency: reminder dispatch returned no successful "
+                            f"delivery path: {dispatch_result}"
+                        )
+                except Exception as e:
+                    logger.warning(f"urgency: reminder dispatch failed: {e}")
                     notify_failed.update(new_urgent)
-                    logger.warning(f"urgency: reminder dispatch returned no successful delivery path: {dispatch_result}")
-            except Exception as e:
-                logger.warning(f"urgency: reminder dispatch failed: {e}")
-                notify_failed.update(new_urgent)
-            # Mark only successfully delivered UIDs as notified so a transient
-            # SMTP/ntfy/browser failure retries instead of lying forever.
-            notified_uids.update(newly_notified)
 
-        # Prune notified_uids that aren't unread anymore (so a future re-urgent
+            next_state = _merge_email_urgency_state(
-        # message with the same UID — rare but possible after archive→unarchive
+                prior,
-        # — can re-notify). Keep only UIDs still in `all_unread_keys`.
+                owner=owner,
-        notified_uids = {u for u in notified_uids if u in all_unread_keys}
+                per_uid_scores=per_uid_scores,
+                notified_uids=notified_uids,
+                all_unread_keys=all_unread_keys,
+                fully_scanned_account_ids=fully_scanned_account_ids,
+                base_account_generations=base_account_generations,
+                timestamp=_time.time(),
+            )
+            return notified_uids, next_state
 
-        state = {
-            "ts": _time.time(),
-            "owner": owner or "",
-            "total_unread": len(all_unread_keys),
-            "total_urgent": total_urgent,
-            "max_score": max_score,
-            "per_uid": per_uid_scores,
-            "notified_uids": sorted(notified_uids),
-        }
         try:
-            STATE_PATH.write_text(_json.dumps(state), encoding="utf-8")
+            await _run_email_urgency_state_transaction(
+                STATE_PATH,
+                STATE_LOCK_DB,
+                _dispatch_and_checkpoint,
+            )
         except Exception as e:
-            logger.warning(f"urgency: state write failed: {e}")
+            logger.warning(f"urgency: state transaction failed: {e}")
 
         # ── 6. Activity-log summary — counts line on top, then per-tier
         # bulleted breakdown so the user can see WHICH emails ranked where

src/llm_core.py

@@ -1319,8 +1319,8 @@ _MISTRAL_REASONING_EFFORT = os.getenv("ODYSSEUS_MISTRAL_REASONING_EFFORT", "high
 
 # Models that support structured thinking — may output </think> without opening tag
 _THINKING_MODEL_PATTERNS = (
-    "qwen3", "qwq", "deepseek-r1", "deepseek-reasoner", "minimax",
+    "qwen3", "qwq", "deepseek-r1", "deepseek-reasoner", "deepseek-v4",
-    "m2-reap", "gemma", "stepfun", "step-3", "step3",
+    "minimax", "m2-reap", "gemma", "stepfun", "step-3", "step3",
     "magistral", "mistral-small", "mistral-medium",
 )
 

src/teacher_escalation.py

@@ -233,7 +233,8 @@ async def _call_teacher(teacher_model_spec: str, prompt: str,
                         owner: Optional[str] = None) -> Optional[str]:
     """Call the configured teacher endpoint with the escalation prompt."""
     from src.llm_core import llm_call_async
-    from src.ai_interaction import _resolve_model, _TEACHER_SYSTEM_PROMPT
+    from src.ai_interaction import _resolve_model
+    from src.agent_tools.model_interaction_tools import _TEACHER_SYSTEM_PROMPT
     try:
         url, model, headers = await asyncio.to_thread(_resolve_model, teacher_model_spec, owner=owner)
     except Exception as e:

src/tools/calendar.py

@@ -196,6 +196,9 @@ async def do_manage_calendar(content: str, owner: Optional[str] = None) -> Dict:
     try:
         if action == "list_calendars":
             _ensure_default_calendar(db, owner)
+            # This read path intentionally persists the lazily-created default;
+            # event creation commits it in the event's transaction instead.
+            db.commit()
             cals = _calendar_query().all()
             result = [{"name": c.name, "href": c.id} for c in cals]
             if result:

src/upload_handler.py

@@ -35,6 +35,16 @@ import logging
 
 logger = logging.getLogger(__name__)
 
+UploadIndexFileSignature = tuple[
+    str,
+    Optional[int],
+    Optional[int],
+    Optional[int],
+    Optional[int],
+    Optional[int],
+]
+UploadIndexSignature = tuple[UploadIndexFileSignature, ...]
+
 
 class UploadCleanupSafetyError(RuntimeError):
     """Raised when cleanup cannot prove that destructive work is safe."""
@@ -242,7 +252,7 @@ class UploadHandler:
 
         # In-memory index cache to avoid O(N) disk I/O on every request
         self._index_cache: Optional[Dict[str, Any]] = None
-        self._index_mtime: float = 0.0
+        self._index_signature: Optional[UploadIndexSignature] = None
     
     def inside_base_dir(self, path: str) -> bool:
         """Check if path is inside base directory"""
@@ -727,62 +737,119 @@ class UploadHandler:
         # Update cache if this is the main index
         if path.endswith("uploads.json"):
             self._index_cache = data
+            self._index_signature = self._upload_index_signature(
+                (path, path + ".bak")
+            )
+
+    @staticmethod
+    def _upload_index_signature(
+        paths: tuple[str, ...],
+    ) -> Optional[UploadIndexSignature]:
+        """Return file identities strong enough to validate the index cache.
+
+        Modification time alone is insufficient: a torn write can change a
+        file without receiving a strictly newer timestamp on some filesystems.
+        Size, inode, and nanosecond change times make those mutations visible
+        while preserving the cache fast path for unchanged files.
+        """
+        signature: list[UploadIndexFileSignature] = []
+        for candidate in paths:
             try:
-                self._index_mtime = os.path.getmtime(path)
+                stat_result = os.stat(candidate)
+            except FileNotFoundError:
+                signature.append((candidate, None, None, None, None, None))
+                continue
             except OSError:
-                self._index_mtime = time.time()
+                return None
+            signature.append(
+                (
+                    candidate,
+                    stat_result.st_dev,
+                    stat_result.st_ino,
+                    stat_result.st_size,
+                    stat_result.st_mtime_ns,
+                    stat_result.st_ctime_ns,
+                )
+            )
+        return tuple(signature)
 
     def _load_upload_index(self, *, fail_on_error: bool = False) -> Dict[str, Any]:
-        """Load the upload index from disk/cache. Uses mtime-based validation
+        """Load the upload index from disk/cache. Uses file-identity validation
-        to avoid redundant parsing on hot paths. When ``fail_on_error`` is
+        to avoid redundant parsing on hot paths without missing same-timestamp
-        true, a missing, malformed, or unreadable live index raises so
+        mutations. When ``fail_on_error`` is true, a missing, malformed, or
-        destructive callers cannot mistake corruption for an empty store.
+        unreadable live index raises so destructive callers cannot mistake
+        corruption for an empty store.
         """
         uploads_db_path = os.path.join(self.upload_dir, "uploads.json")
         candidates = (uploads_db_path, uploads_db_path + ".bak")
-        if fail_on_error:
+        for _attempt in range(3):
-            # A backup is intentionally the previous snapshot. It is useful for
+            signature = self._upload_index_signature(candidates)
-            # non-destructive reads, but cannot authorize deletion when the live
+            if fail_on_error:
-            # index is missing or corrupt.
+                # A backup is intentionally the previous snapshot. It is useful for
-            if not os.path.exists(uploads_db_path):
+                # non-destructive reads, but cannot authorize deletion when the live
-                raise ValueError("live uploads database is missing")
+                # index is missing or corrupt.
-            existing_candidates = [uploads_db_path]
+                if not os.path.exists(uploads_db_path):
-        else:
+                    raise ValueError("live uploads database is missing")
-            existing_candidates = [path for path in candidates if os.path.exists(path)]
+                existing_candidates = [uploads_db_path]
-        if not existing_candidates:
+            else:
-            self._index_cache = {}
+                existing_candidates = [
-            self._index_mtime = 0.0
+                    path for path in candidates if os.path.exists(path)
-            return {}
+                ]
+            if not existing_candidates:
+                self._index_cache = {}
+                self._index_signature = signature
+                return {}
 
-        # Check cache validity
+            # Check cache validity
-        try:
-            mtime = max(os.path.getmtime(path) for path in existing_candidates)
             if (
                 not fail_on_error
+                and signature is not None
                 and self._index_cache is not None
-                and mtime <= self._index_mtime
+                and signature == self._index_signature
             ):
                 return self._index_cache
-        except OSError:
-            mtime = 0.0
 
-        # Try the live file first, fall back to the .bak sibling if the
+            # Try the live file first, fall back to the .bak sibling if the
-        # live file is truncated/corrupted.
+            # live file is truncated/corrupted. A candidate parsed from an old
-        for candidate in existing_candidates:
+            # inode is accepted only when the whole index signature stays
-            try:
+            # stable through the read; otherwise retry so the cache cannot pair
-                with open(candidate, "r", encoding="utf-8") as f:
+            # stale data with a fresh replacement signature.
-                    data = json.load(f)
+            index_changed_during_read = False
-                if isinstance(data, dict):
+            for candidate in existing_candidates:
-                    self._index_cache = data
+                try:
-                    self._index_mtime = mtime
+                    with open(candidate, "r", encoding="utf-8") as f:
-                    return data
+                        data = json.load(f)
-            except Exception as e:
+                    verified_signature = self._upload_index_signature(candidates)
-                logger.warning(f"Failed to read uploads database ({candidate}): {e}")
+                    if (
+                        signature is not None
+                        and verified_signature is not None
+                        and verified_signature != signature
+                    ):
+                        index_changed_during_read = True
+                        break
+                    if isinstance(data, dict):
+                        self._index_cache = data
+                        self._index_signature = verified_signature
+                        return data
+                except Exception as e:
+                    logger.warning(f"Failed to read uploads database ({candidate}): {e}")
+                    verified_signature = self._upload_index_signature(candidates)
+                    if (
+                        signature is not None
+                        and verified_signature is not None
+                        and verified_signature != signature
+                    ):
+                        index_changed_during_read = True
+                        break
+                    continue
+            if index_changed_during_read:
                 continue
+            break
 
         if fail_on_error:
             raise ValueError("live uploads database is unreadable")
         self._index_cache = {}
+        self._index_signature = self._upload_index_signature(candidates)
         return {}
 
     def get_upload_info(self, upload_id: str) -> Optional[Dict[str, Any]]:

tests/test_calendar_default_transaction.py

@@ -0,0 +1,538 @@
+"""Default calendar creation belongs to the caller's transaction.
+
+Before this regression, ``_ensure_default_calendar`` committed independently.
+If event persistence then failed, the event rolled back but a new ``Personal``
+calendar remained (``calendar_count=1``, ``event_count=0``).
+"""
+
+import json
+import threading
+from contextlib import contextmanager
+from datetime import datetime, timedelta
+from types import SimpleNamespace
+
+import pytest
+from fastapi import HTTPException
+from sqlalchemy import create_engine, event
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.orm import Session, sessionmaker
+from sqlalchemy.pool import NullPool
+
+from tests.helpers.import_state import clear_fake_database_modules
+
+clear_fake_database_modules()
+
+import core.database as cdb  # noqa: E402
+import routes.calendar_routes as calendar_routes  # noqa: E402
+from core.database import CalendarCal, CalendarEvent  # noqa: E402
+from routes.calendar_routes import EventCreate  # noqa: E402
+from routes.calendar_routes import (  # noqa: E402
+    _default_calendar_id,
+    _ensure_default_calendar,
+)
+
+
+class _RejectEventCommit(Session):
+    """Reproduce an event commit failure after default-calendar creation."""
+
+    def commit(self):
+        if any(isinstance(row, CalendarEvent) for row in self.new):
+            raise RuntimeError("commit guard rejected event commit")
+        return super().commit()
+
+
+@pytest.fixture
+def session_factory(tmp_path, monkeypatch):
+    engine = create_engine(
+        f"sqlite:///{tmp_path / 'calendar.db'}",
+        connect_args={"check_same_thread": False},
+        poolclass=NullPool,
+    )
+    cdb.Base.metadata.create_all(engine)
+    factory = sessionmaker(
+        bind=engine,
+        autoflush=False,
+        autocommit=False,
+        class_=_RejectEventCommit,
+    )
+    monkeypatch.setattr(cdb, "SessionLocal", factory)
+    monkeypatch.setattr(calendar_routes, "SessionLocal", factory)
+    try:
+        yield factory
+    finally:
+        engine.dispose()
+
+
+def _request():
+    return SimpleNamespace(state=SimpleNamespace(current_user="alice"))
+
+
+def _endpoint(method, suffix):
+    router = calendar_routes.setup_calendar_routes()
+    for route in router.routes:
+        if route.path.endswith(suffix) and method in route.methods:
+            return route.endpoint
+    raise RuntimeError(f"{method} *{suffix} not found")
+
+
+def _counts(factory):
+    db = factory()
+    try:
+        return db.query(CalendarCal).count(), db.query(CalendarEvent).count()
+    finally:
+        db.close()
+
+
+async def test_route_event_failure_rolls_back_new_default_calendar(session_factory):
+    create_event = _endpoint("POST", "/events")
+
+    with pytest.raises(HTTPException) as caught:
+        await create_event(
+            _request(),
+            EventCreate(summary="Planning", dtstart="2126-07-20T09:00:00Z"),
+        )
+
+    assert caught.value.status_code == 500
+    assert _counts(session_factory) == (0, 0)
+
+
+async def test_route_event_validation_failure_rolls_back_new_default_calendar(
+    session_factory,
+):
+    create_event = _endpoint("POST", "/events")
+
+    with pytest.raises(HTTPException) as caught:
+        await create_event(
+            _request(),
+            EventCreate(summary="Planning", dtstart="not-a-datetime"),
+        )
+
+    assert caught.value.status_code == 500
+    assert _counts(session_factory) == (0, 0)
+
+
+async def test_tool_event_failure_rolls_back_new_default_calendar(session_factory):
+    from src.tools.calendar import do_manage_calendar
+
+    result = await do_manage_calendar(
+        json.dumps({
+            "action": "create_event",
+            "summary": "Planning",
+            "dtstart": "2126-07-20T09:00:00Z",
+        }),
+        owner="alice",
+    )
+
+    assert result["exit_code"] == 1
+    assert "commit guard rejected event commit" in result["error"]
+    assert _counts(session_factory) == (0, 0)
+
+
+async def test_tool_event_validation_failure_rolls_back_new_default_calendar(
+    session_factory,
+):
+    from src.tools.calendar import do_manage_calendar
+
+    result = await do_manage_calendar(
+        json.dumps({
+            "action": "create_event",
+            "summary": "Planning",
+            "dtstart": "not-a-datetime",
+        }),
+        owner="alice",
+    )
+
+    assert result["exit_code"] == 1
+    assert "Could not parse dtstart" in result["error"]
+    assert _counts(session_factory) == (0, 0)
+
+
+async def test_route_list_calendars_persists_lazy_default(session_factory):
+    list_calendars = _endpoint("GET", "/calendars")
+
+    result = await list_calendars(_request())
+
+    assert [calendar["name"] for calendar in result["calendars"]] == ["Personal"]
+    assert _counts(session_factory) == (1, 0)
+
+
+async def test_tool_list_calendars_persists_lazy_default(session_factory):
+    from src.tools.calendar import do_manage_calendar
+
+    result = await do_manage_calendar(
+        json.dumps({"action": "list_calendars"}),
+        owner="alice",
+    )
+
+    assert result["exit_code"] == 0
+    assert [calendar["name"] for calendar in result["calendars"]] == ["Personal"]
+    assert _counts(session_factory) == (1, 0)
+
+
+def test_repeated_rename_and_reuse_uses_stable_fallback_ids(tmp_path):
+    engine = create_engine(
+        f"sqlite:///{tmp_path / 'renamed-calendar.db'}",
+        connect_args={"check_same_thread": False},
+        poolclass=NullPool,
+    )
+    cdb.Base.metadata.create_all(engine)
+    factory = sessionmaker(bind=engine, autoflush=False, autocommit=False)
+    db = factory()
+    try:
+        first = _ensure_default_calendar(db, "alice")
+        assert first.id == _default_calendar_id("alice")
+        db.commit()
+
+        # The supported user-rename migration changes owner columns while
+        # deliberately preserving durable row identifiers.
+        first.owner = "bob"
+        db.commit()
+
+        second = _ensure_default_calendar(db, "alice")
+        assert second.id == _default_calendar_id("alice", 1)
+        db.commit()
+
+        # Repeating the same lifecycle must advance deterministically instead
+        # of failing or choosing a random identifier.
+        second.owner = "carol"
+        db.commit()
+
+        third = _ensure_default_calendar(db, "alice")
+        assert third.id == _default_calendar_id("alice", 2)
+        db.commit()
+
+        rows = db.query(CalendarCal).order_by(CalendarCal.owner).all()
+        assert [(row.owner, row.id) for row in rows] == [
+            ("alice", _default_calendar_id("alice", 2)),
+            ("bob", _default_calendar_id("alice")),
+            ("carol", _default_calendar_id("alice", 1)),
+        ]
+    finally:
+        db.close()
+        engine.dispose()
+
+
+def _assert_concurrent_first_use(tmp_path, occupied_owner=None):
+    engine = create_engine(
+        f"sqlite:///{tmp_path / 'concurrent-calendar.db'}",
+        connect_args={"check_same_thread": False, "timeout": 10},
+        poolclass=NullPool,
+    )
+    cdb.Base.metadata.create_all(engine)
+    factory = sessionmaker(bind=engine, autoflush=False, autocommit=False)
+    expected_collision_index = 0
+    if occupied_owner is not None:
+        seed = factory()
+        try:
+            seed.add(CalendarCal(
+                id=_default_calendar_id("alice"),
+                owner=occupied_owner,
+                name="Personal",
+                source="local",
+            ))
+            seed.commit()
+            expected_collision_index = 1
+        finally:
+            seed.close()
+    first_staged = threading.Event()
+    second_selected = threading.Event()
+    errors = []
+
+    @event.listens_for(engine, "after_cursor_execute")
+    def observe_second_gap(conn, cursor, statement, parameters, context, executemany):
+        if (
+            threading.current_thread().name == "calendar-worker-second"
+            and statement.lstrip().upper().startswith("SELECT")
+            and "FROM calendars" in statement
+        ):
+            second_selected.set()
+
+    def create_default(worker, hold=False):
+        db = factory()
+        try:
+            if not hold:
+                assert first_staged.wait(5)
+            cal = _ensure_default_calendar(db, "alice")
+            start = datetime(2126, 7, 20, 9 if hold else 10)
+            db.add(CalendarEvent(
+                uid=worker,
+                calendar_id=cal.id,
+                summary=f"Event {worker}",
+                dtstart=start,
+                dtend=start + timedelta(hours=1),
+            ))
+            if hold:
+                first_staged.set()
+                # The second session has observed the uncommitted gap before
+                # this transaction releases its writer reservation.
+                assert second_selected.wait(5)
+            db.commit()
+            assert cal.id == _default_calendar_id("alice", expected_collision_index)
+        except BaseException as exc:  # pragma: no cover - asserted below
+            errors.append((worker, exc))
+            db.rollback()
+        finally:
+            db.close()
+
+    first = threading.Thread(
+        target=create_default,
+        args=("first", True),
+        name="calendar-worker-first",
+    )
+    second = threading.Thread(
+        target=create_default,
+        args=("second",),
+        name="calendar-worker-second",
+    )
+    first.start()
+    second.start()
+    first.join(10)
+    second.join(10)
+
+    try:
+        assert not first.is_alive() and not second.is_alive()
+        assert errors == []
+        db = factory()
+        try:
+            rows = db.query(CalendarCal).filter(CalendarCal.owner == "alice").all()
+            assert [(row.id, row.name) for row in rows] == [
+                (_default_calendar_id("alice", expected_collision_index), "Personal")
+            ]
+            assert db.query(CalendarEvent).count() == 2
+            if occupied_owner is not None:
+                occupied = db.query(CalendarCal).filter(
+                    CalendarCal.id == _default_calendar_id("alice"),
+                ).one()
+                assert occupied.owner == occupied_owner
+        finally:
+            db.close()
+    finally:
+        engine.dispose()
+
+
+def test_concurrent_first_use_creates_one_sqlite_default(tmp_path):
+    _assert_concurrent_first_use(tmp_path)
+
+
+def test_concurrent_first_use_after_rename_creates_one_fallback_default(tmp_path):
+    _assert_concurrent_first_use(tmp_path, occupied_owner="bob")
+
+
+def test_sqlite_default_stays_in_callers_transaction(session_factory):
+    db = session_factory()
+    try:
+        cal = _ensure_default_calendar(db, "rollback-owner")
+        assert cal.id == _default_calendar_id("rollback-owner")
+        db.rollback()
+    finally:
+        db.close()
+
+    verify = session_factory()
+    try:
+        assert (
+            verify.query(CalendarCal)
+            .filter(CalendarCal.owner == "rollback-owner")
+            .count()
+            == 0
+        )
+    finally:
+        verify.close()
+
+
+def test_sqlite_fallback_default_stays_in_callers_transaction(session_factory):
+    seed = session_factory()
+    try:
+        seed.add(CalendarCal(
+            id=_default_calendar_id("alice"),
+            owner="bob",
+            name="Personal",
+            source="local",
+        ))
+        seed.commit()
+    finally:
+        seed.close()
+
+    db = session_factory()
+    try:
+        cal = _ensure_default_calendar(db, "alice")
+        assert cal.id == _default_calendar_id("alice", 1)
+        db.rollback()
+    finally:
+        db.close()
+
+    verify = session_factory()
+    try:
+        assert verify.query(CalendarCal).filter(CalendarCal.owner == "alice").count() == 0
+        assert verify.query(CalendarCal).filter(CalendarCal.owner == "bob").count() == 1
+    finally:
+        verify.close()
+
+
+class _FakeDialect:
+    name = "postgresql"
+
+
+class _FakeBind:
+    dialect = _FakeDialect()
+
+
+class _FakeQuery:
+    def __init__(self, session):
+        self.session = session
+
+    def filter(self, *conditions):
+        return self
+
+    def with_for_update(self):
+        self.session.locking_read = True
+        return self
+
+    def first(self):
+        self.session.query_count += 1
+        if self.session.query_count == 1:
+            return None
+        return self.session.winner
+
+
+class _GenericRaceSession:
+    """Minimal non-SQLite session that loses the deterministic-ID race."""
+
+    def __init__(self):
+        self.query_count = 0
+        self.nested_entries = 0
+        self.locking_read = False
+        self.candidate = None
+        self.winner = CalendarCal(
+            id=_default_calendar_id("alice"),
+            owner="alice",
+            name="Personal",
+            source="local",
+        )
+
+    def get_bind(self):
+        return _FakeBind()
+
+    def query(self, model):
+        assert model is CalendarCal
+        return _FakeQuery(self)
+
+    @contextmanager
+    def begin_nested(self):
+        self.nested_entries += 1
+        yield
+
+    def add(self, row):
+        self.candidate = row
+
+    def flush(self):
+        raise IntegrityError("insert", {}, RuntimeError("duplicate primary key"))
+
+
+def test_generic_backend_lost_race_recovers_inside_savepoint():
+    db = _GenericRaceSession()
+
+    winner = _ensure_default_calendar(db, "alice")
+
+    assert winner is db.winner
+    assert db.nested_entries == 1
+    assert db.locking_read is True
+    assert db.candidate.id == db.winner.id
+
+
+def test_generic_backend_unattributed_integrity_error_is_not_retried():
+    db = _GenericRaceSession()
+    db.winner = None
+
+    with pytest.raises(IntegrityError):
+        _ensure_default_calendar(db, "alice")
+
+    assert db.nested_entries == 1
+
+
+class _GenericRenamedSlotSession(_GenericRaceSession):
+    """A different owner occupies slot zero; slot one remains available."""
+
+    def __init__(self):
+        super().__init__()
+        self.candidates = []
+        self.winner = CalendarCal(
+            id=_default_calendar_id("alice"),
+            owner="bob",
+            name="Personal",
+            source="local",
+        )
+
+    def add(self, row):
+        self.candidate = row
+        self.candidates.append(row)
+
+    def flush(self):
+        if len(self.candidates) == 1:
+            raise IntegrityError("insert", {}, RuntimeError("duplicate primary key"))
+
+
+def test_generic_backend_renamed_slot_advances_inside_savepoint():
+    db = _GenericRenamedSlotSession()
+
+    fallback = _ensure_default_calendar(db, "alice")
+
+    assert fallback is db.candidates[-1]
+    assert fallback.id == _default_calendar_id("alice", 1)
+    assert fallback.owner == "alice"
+    assert db.nested_entries == 2
+    assert db.locking_read is True
+    assert db.winner.owner == "bob"
+
+
+def test_generic_backend_fallback_keeps_outer_transaction_usable(tmp_path):
+    engine = create_engine(
+        f"sqlite:///{tmp_path / 'generic-savepoint-calendar.db'}",
+        poolclass=NullPool,
+    )
+    cdb.Base.metadata.create_all(engine)
+    # SQLite supplies a lightweight local SQL executor here; changing only the
+    # dispatch name exercises the real Session/savepoint branch used by
+    # PostgreSQL-style backends without pretending to validate their dialect.
+    engine.dialect.name = "postgresql"
+    factory = sessionmaker(bind=engine, autoflush=False, autocommit=False)
+
+    seed = factory()
+    try:
+        seed.add(CalendarCal(
+            id=_default_calendar_id("alice"),
+            owner="bob",
+            name="Personal",
+            source="local",
+        ))
+        seed.commit()
+    finally:
+        seed.close()
+
+    db = factory()
+    try:
+        cal = _ensure_default_calendar(db, "alice")
+        start = datetime(2126, 7, 20, 9)
+        db.add(CalendarEvent(
+            uid="after-fallback",
+            calendar_id=cal.id,
+            summary="Atomic",
+            dtstart=start,
+            dtend=start + timedelta(hours=1),
+        ))
+        db.commit()
+    finally:
+        db.close()
+
+    verify = factory()
+    try:
+        assert [
+            (row.owner, row.id)
+            for row in verify.query(CalendarCal).order_by(CalendarCal.owner).all()
+        ] == [
+            ("alice", _default_calendar_id("alice", 1)),
+            ("bob", _default_calendar_id("alice")),
+        ]
+        assert verify.query(CalendarEvent).count() == 1
+    finally:
+        verify.close()
+        engine.dispose()

tests/test_ci_authoritative_validation.py

@@ -0,0 +1,35 @@
+"""Regression coverage for authoritative Python CI validation."""
+
+import re
+from pathlib import Path
+
+
+_WORKFLOW = (
+    Path(__file__).resolve().parent.parent / ".github" / "workflows" / "ci.yml"
+)
+
+
+def _indented_block(text: str, heading: str, indent: int) -> str:
+    pattern = re.compile(
+        rf"(?ms)^{' ' * indent}{re.escape(heading)}:\n"
+        rf"(?P<body>(?:(?:{' ' * (indent + 2)}.*|\s*)\n)*)"
+    )
+    match = pattern.search(text)
+    assert match is not None, f"missing {heading!r} block"
+    return match.group(0)
+
+
+def test_ci_runs_on_integrated_dev_pushes():
+    workflow = _WORKFLOW.read_text()
+    push = _indented_block(workflow, "push", 2)
+
+    assert re.search(r"(?m)^    branches:\s*\[main,\s*dev\]\s*$", push)
+    assert "paths-ignore:" not in push
+
+
+def test_python_tests_are_authoritative():
+    workflow = _WORKFLOW.read_text()
+    python_tests = _indented_block(workflow, "python-tests", 2)
+
+    assert "python -m pytest -q" in python_tests
+    assert "continue-on-error:" not in python_tests

tests/test_cookbook_helpers.py

@@ -723,7 +723,12 @@ def test_local_windows_download_pid_tracks_inner_bash_and_stop_kills_tree():
     routes_src = (Path(__file__).resolve().parents[1] / "routes" / "cookbook_routes.py").read_text(encoding="utf-8")
     running_src = (Path(__file__).resolve().parents[1] / "static" / "js" / "cookbookRunning.js").read_text(encoding="utf-8")
 
-    assert 'printf \'%s\\\\n\' \\"$$\\" > {pp}' in routes_src
+    # The Windows-local runner publishes Python's valid Win32 fallback before
+    # allowing Git Bash to replace it with /proc/$$/winpid.
+    assert "_windows_local_pid_record_line(pid_path, pid_ready_path)" in routes_src
+    assert "/proc/$$/winpid" in routes_src
+    assert "pid_ready_path.touch()" in routes_src
+    assert '\\"$$\\" > {pp}' not in routes_src
     assert "function Stop-Tree([int]$Id)" in running_src
     assert "('ParentProcessId = ' + $Id)" in running_src
     assert "Stop-Tree ([int]$p)" in running_src

tests/test_cookbook_local_serve_pid_winpid.py

@@ -0,0 +1,180 @@
+"""Behavioral regression coverage for Windows-local Cookbook PID recording."""
+
+import os
+import subprocess
+import time
+from pathlib import Path
+
+from routes.cookbook_routes import _windows_local_pid_record_line
+
+
+ROOT = Path(__file__).resolve().parents[1]
+COOKBOOK_ROUTES = ROOT / "routes" / "cookbook_routes.py"
+
+
+def _fake_cat(tmp_path: Path, body: str) -> Path:
+    fake_bin = tmp_path / "bin"
+    fake_bin.mkdir()
+    cat = fake_bin / "cat"
+    cat.write_text("#!/bin/sh\n" + body + "\n", encoding="utf-8")
+    cat.chmod(0o755)
+    return fake_bin
+
+
+def _env_for(fake_bin: Path, **extra: str) -> dict[str, str]:
+    env = dict(os.environ)
+    env["PATH"] = str(fake_bin) + os.pathsep + env.get("PATH", "")
+    env.update(extra)
+    return env
+
+
+def _run_pid_line(
+    pid_path: Path,
+    ready_path: Path,
+    fake_bin: Path,
+    **extra_env: str,
+) -> subprocess.CompletedProcess:
+    return subprocess.run(
+        ["bash", "-c", _windows_local_pid_record_line(pid_path, ready_path)],
+        capture_output=True,
+        text=True,
+        env=_env_for(fake_bin, **extra_env),
+        timeout=10,
+    )
+
+
+def test_windows_local_pid_line_records_numeric_winpid_after_fallback(tmp_path):
+    pid_path = tmp_path / "serve.pid"
+    ready_path = tmp_path / "serve.pid.ready"
+
+    pid_path.write_text("11111", encoding="utf-8")
+    ready_path.touch()
+
+    cat_arg = tmp_path / "cat-arg.txt"
+    fake_bin = _fake_cat(
+        tmp_path,
+        'printf "%s\\n" "$1" > "$FAKE_CAT_ARG"\n'
+        'printf "%s\\n" "$FAKE_WINPID"',
+    )
+
+    result = _run_pid_line(
+        pid_path,
+        ready_path,
+        fake_bin,
+        FAKE_CAT_ARG=str(cat_arg),
+        FAKE_WINPID="42324",
+    )
+
+    assert result.returncode == 0, result.stderr
+    assert pid_path.read_text(encoding="utf-8").strip() == "42324"
+    assert not ready_path.exists()
+
+    proc_path = cat_arg.read_text(encoding="utf-8").strip()
+    parts = proc_path.strip("/").split("/")
+    assert len(parts) == 3
+    assert parts[0] == "proc"
+    assert parts[1].isdigit()
+    assert parts[2] == "winpid"
+
+
+def test_windows_local_pid_line_waits_for_python_fallback_before_replacing(tmp_path):
+    pid_path = tmp_path / "serve.pid"
+    ready_path = tmp_path / "serve.pid.ready"
+
+    fake_bin = _fake_cat(
+        tmp_path,
+        'printf "%s\\n" "$FAKE_WINPID"',
+    )
+
+    proc = subprocess.Popen(
+        [
+            "bash",
+            "-c",
+            _windows_local_pid_record_line(pid_path, ready_path),
+        ],
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+        text=True,
+        env=_env_for(fake_bin, FAKE_WINPID="42324"),
+    )
+
+    # The inner shell has started, but Python has not published its fallback yet.
+    time.sleep(0.05)
+    assert proc.poll() is None
+    assert not pid_path.exists()
+
+    # Simulate the post-Popen Python publication order.
+    pid_path.write_text("31100", encoding="utf-8")
+    ready_path.touch()
+
+    stdout, stderr = proc.communicate(timeout=10)
+
+    assert proc.returncode == 0, stderr or stdout
+    assert pid_path.read_text(encoding="utf-8").strip() == "42324"
+    assert not ready_path.exists()
+
+
+def test_windows_local_pid_line_preserves_outer_pid_when_mapping_missing(tmp_path):
+    pid_path = tmp_path / "serve.pid"
+    ready_path = tmp_path / "serve.pid.ready"
+
+    pid_path.write_text("31100", encoding="utf-8")
+    ready_path.touch()
+
+    fake_bin = _fake_cat(tmp_path, "exit 1")
+
+    result = _run_pid_line(
+        pid_path,
+        ready_path,
+        fake_bin,
+    )
+
+    assert result.returncode == 0, result.stderr
+    assert pid_path.read_text(encoding="utf-8").strip() == "31100"
+    assert not ready_path.exists()
+
+
+def test_windows_local_pid_line_rejects_malformed_mapping(tmp_path):
+    pid_path = tmp_path / "serve.pid"
+    ready_path = tmp_path / "serve.pid.ready"
+
+    pid_path.write_text("31100", encoding="utf-8")
+    ready_path.touch()
+
+    fake_bin = _fake_cat(
+        tmp_path,
+        'printf "not-a-win32-pid\\n"',
+    )
+
+    result = _run_pid_line(
+        pid_path,
+        ready_path,
+        fake_bin,
+    )
+
+    assert result.returncode == 0, result.stderr
+    assert pid_path.read_text(encoding="utf-8").strip() == "31100"
+    assert not ready_path.exists()
+
+
+def test_local_windows_launcher_publishes_fallback_before_releasing_inner_runner():
+    source = COOKBOOK_ROUTES.read_text(encoding="utf-8")
+    start = source.index("    def _launch_local_detached(")
+    end = source.index(
+        '    @router.post("/api/model/download")',
+        start,
+    )
+    launcher = source[start:end]
+
+    assert "_windows_local_pid_record_line(pid_path, pid_ready_path)" in launcher
+    assert "pid_ready_path.unlink(missing_ok=True)" in launcher
+
+    fallback = launcher.index(
+        'pid_path.write_text(str(proc.pid), encoding="utf-8")'
+    )
+    release = launcher.index("pid_ready_path.touch()")
+
+    assert fallback < release
+
+    # Never write Git Bash's bare MSYS $$ to the session pid file.
+    assert '\\"$$\\" > {pp}' not in launcher

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_llm_core_thinking_models.py

@@ -0,0 +1,27 @@
+"""Regression coverage for structured-thinking model detection."""
+
+import os
+
+os.environ.setdefault("DATABASE_URL", "sqlite:///:memory:")
+
+import pytest
+
+from src.llm_core import _supports_thinking
+
+
+@pytest.mark.parametrize(
+    "model",
+    [
+        "deepseek-v4",
+        "deepseek-v4-flash",
+        "DeepSeek-V4-Flash",
+        "deepseek/deepseek-v4-flash",
+    ],
+)
+def test_deepseek_v4_models_support_thinking(model):
+    assert _supports_thinking(model) is True
+
+
+@pytest.mark.parametrize("model", ["deepseek-v3", "deepseek-chat"])
+def test_other_deepseek_models_are_not_promoted_to_thinking(model):
+    assert _supports_thinking(model) is False

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
 

tests/test_teacher_audit_owner_scope.py

@@ -21,10 +21,17 @@ def test_call_teacher_scopes_model_resolution_to_owner(monkeypatch):
         return ("http://endpoint.local/v1", "teacher-model", {})
 
     async def fake_llm_call_async(url, model, messages, **kwargs):
+        seen["messages"] = messages
         return "teacher reply"
 
+    from src.agent_tools import model_interaction_tools
+
     monkeypatch.setattr("src.ai_interaction._resolve_model", fake_resolve_model)
-    monkeypatch.setattr("src.ai_interaction._TEACHER_SYSTEM_PROMPT", "sys", raising=False)
+    monkeypatch.setattr(
+        model_interaction_tools,
+        "_TEACHER_SYSTEM_PROMPT",
+        "sys",
+    )
     monkeypatch.setattr("src.llm_core.llm_call_async", fake_llm_call_async)
 
     result = asyncio.run(
@@ -34,6 +41,7 @@ def test_call_teacher_scopes_model_resolution_to_owner(monkeypatch):
     assert result == "teacher reply"
     assert seen["owner"] == "alice"
     assert seen["spec"] == "teacher-model"
+    assert seen["messages"][0] == {"role": "system", "content": "sys"}
 
 
 def test_audit_teacher_resolution_scoped_to_owner(monkeypatch):

tests/test_upload_handler_atomicity.py

@@ -15,6 +15,7 @@ These tests exercise:
 * Smoke tests: normal upload, duplicate detection, info lookup after
   a backup-recovery scenario.
 """
+import builtins
 import concurrent.futures
 import io
 import json
@@ -59,6 +60,16 @@ def _db_path(handler: UploadHandler) -> str:
     return os.path.join(handler.upload_dir, "uploads.json")
 
 
+def _truncate_without_newer_mtime(path: str) -> None:
+    """Model a filesystem where a torn write shares the cached timestamp."""
+    before = os.stat(path)
+    with open(path, "rb") as f:
+        full = f.read()
+    with open(path, "wb") as f:
+        f.write(full[: max(1, len(full) // 2)])
+    os.utime(path, ns=(before.st_atime_ns, before.st_mtime_ns))
+
+
 def _seed_entry(owner: str, file_hash: str, file_id: str) -> dict:
     return {
         "id": file_id,
@@ -246,10 +257,7 @@ def test_partial_write_recovery_via_bak(tmp_path):
         "Production _atomic_write_json must create a .bak sibling on subsequent writes."
     )
 
-    full = open(db_path, "rb").read()
+    _truncate_without_newer_mtime(db_path)
-    truncated_len = max(1, len(full) // 2)
-    with open(db_path, "wb") as f:
-        f.write(full[:truncated_len])
 
     recovered = handler._load_upload_index()
     missing = [k for k in original if k not in recovered]
@@ -259,6 +267,69 @@ def test_partial_write_recovery_via_bak(tmp_path):
     )
 
 
+def test_partial_write_recovery_via_bak_after_restart(tmp_path):
+    """A fresh handler must recover the previous snapshot from ``.bak``."""
+    handler = _make_handler(tmp_path)
+    db_path = _db_path(handler)
+    original = {
+        f"owner:hash_{i}": _seed_entry("owner", f"hash_{i}", f"id_{i}")
+        for i in range(3)
+    }
+    handler._atomic_write_json(db_path, original)
+    handler._atomic_write_json(db_path, {"latest": True})
+    _truncate_without_newer_mtime(db_path)
+
+    restarted_handler = UploadHandler(
+        base_dir=handler.base_dir,
+        upload_dir=handler.upload_dir,
+    )
+
+    assert restarted_handler._load_upload_index() == original
+
+
+def test_unchanged_upload_index_uses_cache(tmp_path, monkeypatch):
+    """The stronger file signature must preserve the unchanged-index fast path."""
+    handler = _make_handler(tmp_path)
+    original = {"owner:hash": _seed_entry("owner", "hash", "id")}
+    handler._atomic_write_json(_db_path(handler), original)
+
+    def fail_if_parsed(_file):
+        raise AssertionError("unchanged upload index should be served from cache")
+
+    monkeypatch.setattr(json, "load", fail_if_parsed)
+
+    assert handler._load_upload_index() == original
+
+
+def test_upload_index_retries_when_replaced_during_read(tmp_path, monkeypatch):
+    """Do not cache old JSON under the signature of a newer atomic replace."""
+    handler = _make_handler(tmp_path)
+    db_path = _db_path(handler)
+    old_index = {"owner:old": _seed_entry("owner", "old", "old_id")}
+    new_index = {"owner:new": _seed_entry("owner", "new", "new_id")}
+    handler._atomic_write_json(db_path, old_index)
+    handler._index_cache = None
+    handler._index_signature = None
+
+    real_open = builtins.open
+    replaced = False
+
+    def racing_open(file, mode="r", *args, **kwargs):
+        nonlocal replaced
+        handle = real_open(file, mode, *args, **kwargs)
+        if os.fspath(file) == db_path and "r" in mode and not replaced:
+            replaced = True
+            replacement = db_path + ".replacement"
+            with real_open(replacement, "w", encoding="utf-8") as out:
+                json.dump(new_index, out)
+            os.replace(replacement, db_path)
+        return handle
+
+    monkeypatch.setattr(builtins, "open", racing_open)
+
+    assert handler._load_upload_index() == new_index
+
+
 # ---------------------------------------------------------------------------
 # Atomicity primitive audit on the production module.
 # ---------------------------------------------------------------------------
@@ -390,10 +461,8 @@ def test_smoke_info_lookup_after_bak_recovery(tmp_path):
     handler._atomic_write_json(db_path, {"sentinel": True})
     assert os.path.exists(db_path + ".bak")
 
-    # Truncate the live file.
+    # Truncate the live file without assuming the filesystem advances mtime.
-    full = open(db_path, "rb").read()
+    _truncate_without_newer_mtime(db_path)
-    with open(db_path, "wb") as f:
-        f.write(full[: max(1, len(full) // 2)])
 
     info = handler.get_upload_info(first["id"])
     assert info is not None, "Info lookup must succeed after .bak recovery."