diff --git a/core/database.py b/core/database.py index a9ad90b8b..6eb529949 100644 --- a/core/database.py +++ b/core/database.py @@ -5,7 +5,7 @@ from datetime import datetime, timezone from pathlib import Path from typing import Optional from urllib.parse import unquote, urlparse -from sqlalchemy import event, create_engine, Column, String, Text, Boolean, DateTime, Integer, ForeignKey, JSON, Index, func, text +from sqlalchemy import DDL, event, create_engine, Column, String, Text, Boolean, DateTime, Integer, ForeignKey, JSON, Index, func, inspect, text from sqlalchemy.engine import Engine, make_url from sqlalchemy.types import TypeDecorator from sqlalchemy.ext.declarative import declarative_base, declared_attr @@ -430,6 +430,93 @@ class EmailAccount(TimestampMixin, Base): ) +class EmailAccountOwnerLock(Base): + """Durable per-owner mutex for email-account default mutations. + + Row-locking databases serialize mutations by locking this row before they + inspect or stage EmailAccount changes. SQLite uses ``BEGIN IMMEDIATE`` + instead, because it ignores ``SELECT ... FOR UPDATE``; keeping the table in + the shared metadata still makes the non-SQLite path available without a + separate migration. The empty key represents the normalized legacy / + unconfigured scope shared by ``owner IS NULL`` and ``owner = ''`` rows. + """ + __tablename__ = "email_account_owner_locks" + + owner_key = Column(String, primary_key=True) + + +_EMAIL_ACCOUNT_DEFAULT_INDEX = "ux_email_accounts_one_default_per_owner" +_EMAIL_ACCOUNT_DEFAULT_INDEX_DDL = { + "sqlite": ( + f"CREATE UNIQUE INDEX IF NOT EXISTS {_EMAIL_ACCOUNT_DEFAULT_INDEX} " + "ON email_accounts (COALESCE(owner, '')) WHERE is_default = 1" + ), + "postgresql": ( + f"CREATE UNIQUE INDEX IF NOT EXISTS {_EMAIL_ACCOUNT_DEFAULT_INDEX} " + "ON email_accounts ((COALESCE(owner, ''))) WHERE is_default IS TRUE" + ), +} + + +# SQLAlchemy cannot express one portable partial, functional index across the +# two supported database families. Register dialect-specific DDL so fresh +# databases get the invariant as part of create_all(); the startup migration +# below installs the same index on existing databases after normalizing legacy +# duplicate rows. +for _dialect_name, _index_ddl in _EMAIL_ACCOUNT_DEFAULT_INDEX_DDL.items(): + event.listen( + EmailAccount.__table__, + "after_create", + DDL(_index_ddl).execute_if(dialect=_dialect_name), + ) + + +def lock_email_account_owner_mutations(db, *owners: str) -> None: + """Lock normalized email-account owner scopes in canonical order. + + ``NULL`` and the empty string are one legacy/single-user owner partition, + matching the unique default-account index. SQLite has only a database + writer reservation, while row-locking databases use durable mutex rows. + Sorting all requested owner keys keeps multi-owner operations such as user + rename from deadlocking with another mutation that requests the same keys + in the opposite order. + """ + from sqlalchemy.exc import IntegrityError + + owner_keys = sorted({owner or "" for owner in owners} or {""}) + if db.get_bind().dialect.name == "sqlite": + db.execute(text("BEGIN IMMEDIATE")) + return + + for owner_key in owner_keys: + lock_row = db.get( + EmailAccountOwnerLock, + owner_key, + with_for_update=True, + ) + if lock_row is not None: + continue + + inserted = False + try: + with db.begin_nested(): + db.add(EmailAccountOwnerLock(owner_key=owner_key)) + db.flush() + inserted = True + except IntegrityError: + # A competing transaction created the mutex row first. Once its + # insert commits, lock that durable row before touching accounts. + pass + + if not inserted: + ( + db.query(EmailAccountOwnerLock) + .filter(EmailAccountOwnerLock.owner_key == owner_key) + .with_for_update() + .one() + ) + + class ModelEndpoint(TimestampMixin, Base): """Admin-configured model endpoints. Models are auto-discovered via /v1/models.""" __tablename__ = "model_endpoints" @@ -1812,72 +1899,142 @@ class Integration(TimestampMixin, Base): -def _migrate_seed_email_account(): - """If email_accounts is empty and settings.json has legacy flat imap_host/smtp_host - keys, create a single default account from them so nothing breaks for users who - upgraded. Safe to run repeatedly — it short-circuits once any row exists.""" +def _migrate_email_account_default_invariant(): + """Normalize legacy duplicates and install durable at-most-one enforcement. + + Older databases only had a non-unique ``(owner, is_default)`` lookup index. + Keep the oldest default deterministically in each normalized owner scope, + then add the same partial functional unique index used for fresh schemas. + """ + dialect_name = engine.dialect.name + index_ddl = _EMAIL_ACCOUNT_DEFAULT_INDEX_DDL.get(dialect_name) + if index_ddl is None: + logger.warning( + "Email-account default uniqueness is not available for database " + "dialect %s; mutations remain serialized but are not protected by " + "a database constraint", + dialect_name, + ) + return + try: - with engine.connect() as conn: - tables = [r[0] for r in conn.execute(text( - "SELECT name FROM sqlite_master WHERE type='table' AND name='email_accounts'" - ))] - if "email_accounts" not in tables: - return - existing = conn.execute(text("SELECT COUNT(*) FROM email_accounts")).scalar() or 0 - if existing > 0: + with engine.begin() as conn: + if not inspect(conn).has_table(EmailAccount.__tablename__): return + default_rows = conn.execute(text(""" + SELECT id, owner + FROM email_accounts + WHERE is_default IS TRUE + ORDER BY + COALESCE(owner, ''), + CASE WHEN created_at IS NULL THEN 1 ELSE 0 END, + created_at, + id + """)).mappings() + seen_owner_keys = set() + duplicate_ids = [] + for row in default_rows: + owner_key = row["owner"] or "" + if owner_key in seen_owner_keys: + duplicate_ids.append(row["id"]) + else: + seen_owner_keys.add(owner_key) - import json as _json - import uuid as _uuid - from pathlib import Path - settings_file = Path(SETTINGS_FILE) - if not settings_file.exists(): - return - try: - s = _json.loads(settings_file.read_text(encoding="utf-8")) - except Exception: - return + for account_id in duplicate_ids: + conn.execute( + text("UPDATE email_accounts SET is_default = :value WHERE id = :id"), + {"value": False, "id": account_id}, + ) + conn.execute(text(index_ddl)) - imap_host = (s.get("imap_host") or "").strip() - smtp_host = (s.get("smtp_host") or "").strip() - if not imap_host and not smtp_host: - return # nothing to migrate + if duplicate_ids: + logger.warning( + "Normalized %d duplicate default email account(s) before " + "installing %s", + len(duplicate_ids), + _EMAIL_ACCOUNT_DEFAULT_INDEX, + ) + except Exception: + # Starting without the constraint would silently retain the race this + # migration is intended to close. Fail startup so an operator sees and + # can repair an incompatible schema instead of accepting unsafe writes. + logger.exception("Failed to enforce the email-account default invariant") + raise + + +def _migrate_seed_email_account(): + """Atomically seed one legacy default account when no account exists. + + Reading settings is intentionally done before taking the owner mutex. The + decisive emptiness check and insert share one locked transaction, so two + application workers starting together cannot both seed a default row. + """ + import json as _json + import uuid as _uuid + + settings_file = Path(SETTINGS_FILE) + if not settings_file.exists(): + return + try: + s = _json.loads(settings_file.read_text(encoding="utf-8")) + except Exception: + return + + imap_host = (s.get("imap_host") or "").strip() + smtp_host = (s.get("smtp_host") or "").strip() + if not imap_host and not smtp_host: + return + + db = None + try: + if not inspect(engine).has_table(EmailAccount.__tablename__): + return + db = SessionLocal() + lock_email_account_owner_mutations(db, "") + existing = db.execute(text("SELECT COUNT(*) FROM email_accounts")).scalar() or 0 + if existing > 0: + return now = utcnow_naive() - with engine.begin() as conn: - conn.execute(text(""" - INSERT INTO email_accounts - (id, owner, name, is_default, enabled, - imap_host, imap_port, imap_user, imap_password, imap_starttls, - smtp_host, smtp_port, smtp_user, smtp_password, - from_address, created_at, updated_at) - VALUES - (:id, :owner, :name, :is_default, :enabled, - :imap_host, :imap_port, :imap_user, :imap_password, :imap_starttls, - :smtp_host, :smtp_port, :smtp_user, :smtp_password, - :from_address, :created_at, :updated_at) - """), { - "id": _uuid.uuid4().hex, - "owner": None, - "name": "Default", - "is_default": True, - "enabled": True, - "imap_host": imap_host, - "imap_port": int(s.get("imap_port") or 993), - "imap_user": s.get("imap_user") or "", - "imap_password": s.get("imap_password") or "", - "imap_starttls": bool(s.get("imap_starttls", True)), - "smtp_host": smtp_host, - "smtp_port": int(s.get("smtp_port") or 465), - "smtp_user": s.get("smtp_user") or "", - "smtp_password": s.get("smtp_password") or "", - "from_address": s.get("email_from") or "", - "created_at": now, - "updated_at": now, - }) - logging.getLogger(__name__).info("Seeded email_accounts 'Default' from settings.json") + db.execute(text(""" + INSERT INTO email_accounts + (id, owner, name, is_default, enabled, + imap_host, imap_port, imap_user, imap_password, imap_starttls, + smtp_host, smtp_port, smtp_user, smtp_password, + from_address, created_at, updated_at) + VALUES + (:id, :owner, :name, :is_default, :enabled, + :imap_host, :imap_port, :imap_user, :imap_password, :imap_starttls, + :smtp_host, :smtp_port, :smtp_user, :smtp_password, + :from_address, :created_at, :updated_at) + """), { + "id": _uuid.uuid4().hex, + "owner": None, + "name": "Default", + "is_default": True, + "enabled": True, + "imap_host": imap_host, + "imap_port": int(s.get("imap_port") or 993), + "imap_user": s.get("imap_user") or "", + "imap_password": s.get("imap_password") or "", + "imap_starttls": bool(s.get("imap_starttls", True)), + "smtp_host": smtp_host, + "smtp_port": int(s.get("smtp_port") or 465), + "smtp_user": s.get("smtp_user") or "", + "smtp_password": s.get("smtp_password") or "", + "from_address": s.get("email_from") or "", + "created_at": now, + "updated_at": now, + }) + db.commit() + logger.info("Seeded email_accounts 'Default' from settings.json") except Exception as e: - logging.getLogger(__name__).warning(f"seed email account migration: {e}") + if db is not None: + db.rollback() + logger.warning("seed email account migration: %s", e) + finally: + if db is not None: + db.close() # WARNING: Foreign-key enforcement is enabled globally for all SQLite connections. @@ -1960,6 +2117,7 @@ def init_db(): _migrate_add_crew_member_id() _migrate_add_assistant_columns() _migrate_add_email_smtp_security() + _migrate_email_account_default_invariant() _migrate_seed_email_account() _migrate_add_calendar_metadata() _migrate_add_calendar_is_utc() diff --git a/docs/setup.md b/docs/setup.md index 53a6fb28c..171a195e7 100644 --- a/docs/setup.md +++ b/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 ` 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/', { + method: 'PATCH', + credentials: 'same-origin', + headers: {'Content-Type': 'application/json'}, + body: JSON.stringify({supports_tools: true}) +}).then(r => r.json()).then(console.log) +``` + +Find `` 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 diff --git a/routes/auth_routes.py b/routes/auth_routes.py index 5c7a4e04a..c0c370561 100644 --- a/routes/auth_routes.py +++ b/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"): diff --git a/routes/calendar_routes.py b/routes/calendar_routes.py index 6e0ee124c..b9c3b0a52 100644 --- a/routes/calendar_routes.py +++ b/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() - db.refresh(cal) - return cal + + if dialect == "sqlite": + db.add(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: diff --git a/routes/cookbook_routes.py b/routes/cookbook_routes.py index 1d79ba809..d51fb9a09 100644 --- a/routes/cookbook_routes.py +++ b/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//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") diff --git a/routes/email_routes.py b/routes/email_routes.py index 03b73ebb5..e10eabd59 100644 --- a/routes/email_routes.py +++ b/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 = q.filter(EmailAccount.owner == owner) + q = _email_account_owner_scope(q, owner) row = q.first() if row is None: row = EmailAccount(id=_uuid.uuid4().hex, owner=owner, name="Default", is_default=True, enabled=True) @@ -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 = clear_q.filter(EmailAccount.owner == owner) + clear_q = _email_account_owner_scope(clear_q, owner) clear_q.update({EmailAccount.is_default: False}) db.commit() finally: @@ -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) - if owner: - scope_q = scope_q.filter(EmailAccount.owner == owner) + scope_q = _email_account_owner_scope(db.query(EmailAccount), owner) existing_count = scope_q.count() if row.is_default or existing_count == 0: scope_q.update({EmailAccount.is_default: False}) @@ -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) - if not row: - return {"ok": False, "error": "Account not found"} + row = _lock_and_reload_email_account( + db, account_id, owner, initial_scope + ) + row_scope = row.owner or "" was_default = bool(row.is_default) db.delete(row) - db.commit() + # Flush the removal before staging a replacement default. The + # partial unique index is checked statement-by-statement, and the + # ORM is otherwise free to UPDATE the promoted row before DELETE. + db.flush() # If the deleted row was default, promote the next-oldest enabled # row owned by THIS user. Without the owner filter we'd promote # another user's account and the deleter would silently inherit # it as their default. if was_default: - promote_q = db.query(EmailAccount).filter(EmailAccount.enabled == True) # noqa: E712 - if owner: - promote_q = promote_q.filter(EmailAccount.owner == owner) - promote = promote_q.order_by(EmailAccount.created_at.asc()).first() + promote_q = db.query(EmailAccount).filter( + EmailAccount.id != account_id, + EmailAccount.enabled == True, # noqa: E712 + ) + promote_q = _email_account_owner_scope(promote_q, row_scope) + promote = promote_q.order_by( + EmailAccount.created_at.asc(), EmailAccount.id.asc() + ).first() if promote: promote.is_default = True - db.commit() + # Deletion and any replacement promotion are one durable state + # transition, so another worker can never observe or race the old + # split-commit gap. + db.commit() return {"ok": True} finally: db.close() @@ -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) - if not row: - return {"ok": False, "error": "Account not found"} - # SECURITY: scope the "clear other defaults" sweep to this user's - # accounts so we don't unset another user's default flag. - clear_q = db.query(EmailAccount) - if owner: - clear_q = clear_q.filter(EmailAccount.owner == owner) + row = _lock_and_reload_email_account( + db, account_id, owner, initial_scope + ) + # Scope the sweep to the target row's normalized owner partition; + # this also handles visible legacy NULL/empty-owner accounts. + clear_q = _email_account_owner_scope( + db.query(EmailAccount), row.owner or "" + ) clear_q.update({EmailAccount.is_default: False}) row.is_default = True db.commit() diff --git a/scripts/demo_email/demo_account.py b/scripts/demo_email/demo_account.py index 9555b6791..8a0f1190a 100755 --- a/scripts/demo_email/demo_account.py +++ b/scripts/demo_email/demo_account.py @@ -1,5 +1,5 @@ #!/usr/bin/env python3 -"""Create/remove the switchable, non-default 'Demo' EmailAccount in Odysseus. +"""Create/remove the switchable 'Demo' EmailAccount in Odysseus. Mirrors the existing local-Dovecot account (localhost:31143, STARTTLS) but points at the throwaway demo@odysseus.local mailbox. Password is stored Fernet-encrypted @@ -20,7 +20,14 @@ from pathlib import Path ROOT = Path(__file__).resolve().parent.parent.parent sys.path.insert(0, str(ROOT)) -from core.database import SessionLocal, EmailAccount, Base, engine # noqa: E402 +from core.database import ( # noqa: E402 + Base, + EmailAccount, + SessionLocal, + engine, + lock_email_account_owner_mutations, +) +from sqlalchemy import or_ # noqa: E402 from src.secret_storage import encrypt # noqa: E402 NAME = "Demo" @@ -31,18 +38,98 @@ IMAP_PASSWORD = "demodemo" OWNER = "" -def setup() -> int: - Base.metadata.create_all(bind=engine) +def _owner_scope(query, owner: str): + if owner: + return query.filter(EmailAccount.owner == owner) + return query.filter(or_(EmailAccount.owner == None, EmailAccount.owner == "")) # noqa: E711 + + +def _discover_demo_scopes() -> set[str]: db = SessionLocal() try: - acct = db.query(EmailAccount).filter( - EmailAccount.name == NAME, EmailAccount.imap_user == IMAP_USER - ).first() + return { + row.owner or "" + for row in db.query(EmailAccount).filter( + EmailAccount.name == NAME, + EmailAccount.imap_user == IMAP_USER, + ).all() + } + finally: + db.close() + + +def _lock_and_load_demo_rows(db, scopes: set[str]): + """Reload Demo rows under every observed owner lock.""" + scopes = set(scopes) or {OWNER} + while True: + lock_email_account_owner_mutations(db, *scopes) + rows = ( + db.query(EmailAccount) + .filter( + EmailAccount.name == NAME, + EmailAccount.imap_user == IMAP_USER, + ) + .order_by(EmailAccount.created_at.asc(), EmailAccount.id.asc()) + .all() + ) + current_scopes = {row.owner or "" for row in rows} + if current_scopes.issubset(scopes) or db.get_bind().dialect.name == "sqlite": + return rows + db.rollback() + scopes.update(current_scopes) + + +def _promote_oldest_enabled(db, owner: str, excluded_ids: list[str]) -> None: + remaining = _owner_scope( + db.query(EmailAccount).filter( + EmailAccount.enabled == True, # noqa: E712 + ~EmailAccount.id.in_(excluded_ids), + ), + owner, + ) + if remaining.filter(EmailAccount.is_default == True).first() is not None: # noqa: E712 + return + promote = remaining.order_by( + EmailAccount.created_at.asc(), EmailAccount.id.asc() + ).first() + if promote is not None: + promote.is_default = True + + +def setup() -> int: + Base.metadata.create_all(bind=engine) + scopes = _discover_demo_scopes() | {OWNER} + db = SessionLocal() + try: + rows = _lock_and_load_demo_rows(db, scopes) + acct = rows[0] if rows else None if acct is None: acct = EmailAccount(id=uuid.uuid4().hex, name=NAME) db.add(acct) + old_scope = acct.owner or "" + was_default = bool(acct.is_default) + if old_scope != OWNER: + # Move a non-default row first so the unique index cannot see two + # defaults transiently while SQLAlchemy flushes the owner move and + # old-scope promotion in separate UPDATE statements. + acct.is_default = False + acct.owner = OWNER + db.flush() + if was_default: + _promote_oldest_enabled(db, old_scope, [acct.id]) + + target_default = _owner_scope( + db.query(EmailAccount).filter( + EmailAccount.id != acct.id, + EmailAccount.is_default == True, # noqa: E712 + ), + OWNER, + ).first() acct.owner = OWNER - acct.is_default = False # never default — user switches to it + # Keep Demo non-default when a real default exists. If it is the only + # enabled account, it must be default to preserve normal create + # semantics and avoid leaving the owner partition without one. + acct.is_default = target_default is None acct.enabled = True acct.imap_host = "localhost" acct.imap_port = 31143 @@ -57,20 +144,27 @@ def setup() -> int: acct.smtp_password = encrypt(IMAP_PASSWORD) acct.from_address = IMAP_USER db.commit() - print(f"'{NAME}' account ready (id={acct.id}, non-default, switchable).") + state = "default" if acct.is_default else "non-default" + print(f"'{NAME}' account ready (id={acct.id}, {state}, switchable).") return 0 finally: db.close() def teardown() -> int: + scopes = _discover_demo_scopes() db = SessionLocal() try: - rows = db.query(EmailAccount).filter( - EmailAccount.name == NAME, EmailAccount.imap_user == IMAP_USER - ).all() + rows = _lock_and_load_demo_rows(db, scopes) + deleted_ids = [row.id for row in rows] + default_scopes = {row.owner or "" for row in rows if row.is_default} for r in rows: db.delete(r) + # Ensure the old default DELETE reaches the database before a + # replacement UPDATE; the unique index is enforced per statement. + db.flush() + for owner in default_scopes: + _promote_oldest_enabled(db, owner, deleted_ids) db.commit() print(f"removed {len(rows)} '{NAME}' account row(s).") return 0 diff --git a/src/builtin_actions.py b/src/builtin_actions.py index 68817467f..216bb0360 100644 --- a/src/builtin_actions.py +++ b/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 - # through to default chat as a last resort). + # Resolve with the task owner as before, but defer the availability + # 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() - try: - from sqlalchemy import and_ as _and, or_ as _or - q = db.query(_EA).filter(_EA.enabled == True) # noqa: E712 - if owner: - unowned = _or(_EA.owner == None, _EA.owner == "") # noqa: E711 - same_mailbox = _or(_EA.imap_user == owner, _EA.from_address == owner) - q = q.filter(_or(_EA.owner == owner, _and(unowned, same_mailbox))) - if target_account_id: - q = q.filter(_EA.id == target_account_id) - accounts = q.all() - finally: - db.close() + def _enumerate_enabled_accounts(): + db = _SL() + try: + from sqlalchemy import and_ as _and, or_ as _or + q = db.query(_EA).filter(_EA.enabled == True) # noqa: E712 + if owner: + unowned = _or(_EA.owner == None, _EA.owner == "") # noqa: E711 + same_mailbox = _or( + _EA.imap_user == owner, + _EA.from_address == owner, + ) + q = q.filter( + _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 = ":" → {"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]: - return results - uids = data[0].split()[-30:] + if status != "OK": + return results, False + 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 - # scan window. Read messages remain cached because tags are useful - # on read mail too; unread state is refreshed per scan above. - seen_uids = {it["uid"] for it in items} - cache_uids = cache.get("uids", {}) - for stale in [u for u in cache_uids if u not in seen_uids]: - cache_uids.pop(stale, None) + if scan_complete: + # Only a complete account scan proves a cached UID left the + # recent window. Partial/failing scans preserve prior facts. + seen_uids = {it["uid"] for it in items} + cache_uids = cache.get("uids", {}) + for stale in [u for u in cache_uids if u not in seen_uids]: + 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. - try: - prior = _json.loads(STATE_PATH.read_text(encoding="utf-8")) if STATE_PATH.exists() else {} - except Exception: - 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] + # ── 5. Fire a reminder only when a previously-unnotified UID scores + # urgent. The read, decision, delivery, and checkpoint are serialized + # below so two scheduler workers cannot both act on the same stale + # state or overwrite each other's successful checkpoint. newly_notified = set() notify_failed = set() - if new_urgent: - title = "Urgent email" if total_urgent == 1 else f"{total_urgent} urgent emails" - # Build a real listing — subject · sender · reason for each urgent - # one — so the reminder email tells you which messages to act on, - # 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. + + def _urgency_reminder_payload(reminder_keys): + total = len(reminder_keys) + title = "Urgent email" if total == 1 else f"{total} urgent emails" sorted_urgent = sorted( - ((k, per_uid_scores[k]) for k in urgent_keys), - key=lambda kv: kv[1].get("score", 0), reverse=True, + ((key, per_uid_scores[key]) for key in reminder_keys), + 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:", ""] - for i, (k, v) in enumerate(sorted_urgent, 1): - subj = (v.get("subject") or "(no subject)")[:160] - frm = v.get("from") or "" - why = v.get("reason") or "" - uid_for_link = str(k).split(":", 1)[-1] + lines = [ + f"{total} email" + ("" if total == 1 else "s") + + " need an urgent reply:", + "", + ] + 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.") - body = "\n".join(lines) - try: - # Call dispatch_reminder DIRECTLY (no HTTP/auth roundtrip — the - # endpoint version 401's the background scheduler because it - # has no session cookie). - from routes.note_routes import dispatch_reminder - dispatch_result = await dispatch_reminder( - title=title, note_body=body, note_id="urgent-email", - owner=owner or "", - ) - 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) - else: + lines.append(f"…and {total - len(sorted_urgent)} more.") + return title, "\n".join(lines) + + async def _dispatch_urgency_reminder(reminder_keys): + # Call dispatch_reminder directly: a scheduler has no browser + # session cookie with which to call the HTTP endpoint. + from routes.note_routes import dispatch_reminder + title, body = _urgency_reminder_payload(reminder_keys) + return await dispatch_reminder( + title=title, + note_body=body, + note_id="urgent-email", + owner=owner or "", + ) + + async def _dispatch_and_checkpoint(prior): + notified_uids = _email_urgency_string_set( + prior.get("notified_uids", []) + ) + observed_accounts = { + _email_urgency_account_key(key) for key in per_uid_scores + } | 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 - # message with the same UID — rare but possible after archive→unarchive - # — can re-notify). Keep only UIDs still in `all_unread_keys`. - notified_uids = {u for u in notified_uids if u in all_unread_keys} + next_state = _merge_email_urgency_state( + prior, + owner=owner, + 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 diff --git a/src/llm_core.py b/src/llm_core.py index dd112cedc..30aff2e47 100644 --- a/src/llm_core.py +++ b/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 without opening tag _THINKING_MODEL_PATTERNS = ( - "qwen3", "qwq", "deepseek-r1", "deepseek-reasoner", "minimax", - "m2-reap", "gemma", "stepfun", "step-3", "step3", + "qwen3", "qwq", "deepseek-r1", "deepseek-reasoner", "deepseek-v4", + "minimax", "m2-reap", "gemma", "stepfun", "step-3", "step3", "magistral", "mistral-small", "mistral-medium", ) diff --git a/src/teacher_escalation.py b/src/teacher_escalation.py index 49134991c..4b6206a8d 100644 --- a/src/teacher_escalation.py +++ b/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: diff --git a/src/tools/calendar.py b/src/tools/calendar.py index e6572ba40..6dda5a0e3 100644 --- a/src/tools/calendar.py +++ b/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: diff --git a/tests/test_calendar_default_transaction.py b/tests/test_calendar_default_transaction.py new file mode 100644 index 000000000..ffd981e51 --- /dev/null +++ b/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() diff --git a/tests/test_cookbook_helpers.py b/tests/test_cookbook_helpers.py index bf6c47d4b..37620c88a 100644 --- a/tests/test_cookbook_helpers.py +++ b/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 diff --git a/tests/test_cookbook_local_serve_pid_winpid.py b/tests/test_cookbook_local_serve_pid_winpid.py new file mode 100644 index 000000000..7038fccbf --- /dev/null +++ b/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 diff --git a/tests/test_email_account_default_serialization.py b/tests/test_email_account_default_serialization.py new file mode 100644 index 000000000..6d2394378 --- /dev/null +++ b/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)] diff --git a/tests/test_email_urgency_checkpoint.py b/tests/test_email_urgency_checkpoint.py new file mode 100644 index 000000000..7fcec08a7 --- /dev/null +++ b/tests/test_email_urgency_checkpoint.py @@ -0,0 +1,1328 @@ +import asyncio +import json +import threading +from types import SimpleNamespace + +import pytest + + +class _Column: + def __eq__(self, _other): + return True + + def __ne__(self, _other): + return True + + +class _Query: + def __init__(self, rows): + self._rows = rows + + def filter(self, *_args, **_kwargs): + return self + + def all(self): + return list(self._rows) + + +class _Db: + def __init__(self, rows): + self._rows = rows + + def query(self, _model): + return _Query(self._rows()) + + def close(self): + return None + + +class _EmailAccount: + enabled = _Column() + owner = _Column() + imap_user = _Column() + from_address = _Column() + id = _Column() + + +class _FakeImap: + def __init__(self, account_id, failures, seen_accounts, search_uids): + self.account_id = account_id + self.failures = failures + self.seen_accounts = seen_accounts + self.search_uids = tuple(search_uids.get(account_id, ("1",))) + + def select(self, *_args, **_kwargs): + if self.account_id in self.failures: + raise RuntimeError(f"{self.account_id} unavailable") + return "OK", [] + + def uid(self, command, *_args): + if self.account_id in self.failures: + raise RuntimeError(f"{self.account_id} unavailable") + if command == "SEARCH": + return "OK", [" ".join(self.search_uids).encode()] + uid = _args[0] + uid = uid.decode() if isinstance(uid, bytes) else str(uid) + query = str(_args[-1]) if _args else "" + seen = "\\Seen" if self.account_id in self.seen_accounts else "" + flags = f"{uid} (UID {uid} FLAGS ({seen}))".encode() + if query == "(UID FLAGS)": + return "OK", [flags] + raw = ( + f"From: Sender {self.account_id} \r\n" + f"Subject: Urgent request for {self.account_id} uid {uid}\r\n" + f"Message-ID: <{self.account_id}-{uid}@example.com>\r\n" + "\r\n" + "Please reply immediately." + ).encode() + return "OK", [(flags, raw)] + + def logout(self): + return None + + +def _account(account_id): + return SimpleNamespace( + id=account_id, + enabled=True, + owner="alice", + imap_user="alice", + from_address="alice", + ) + + +def _configure_action(monkeypatch, tmp_path, account_ids): + from core import database + from routes import email_helpers + from src import builtin_actions, llm_core, settings, task_endpoint + + runtime = { + "accounts": list(account_ids), + "failures": set(), + "seen_accounts": set(), + "search_uids": {}, + "settings": { + "reminder_channel": "browser", + "reminder_llm_synthesis": False, + "app_public_url": "", + }, + } + + monkeypatch.setattr(builtin_actions, "DATA_DIR", str(tmp_path)) + monkeypatch.setattr( + builtin_actions, + "EMAIL_URGENCY_CACHE_DIR", + str(tmp_path / "urgency-cache"), + ) + monkeypatch.setattr(database, "EmailAccount", _EmailAccount) + monkeypatch.setattr( + database, + "SessionLocal", + lambda: _Db(lambda: [_account(value) for value in runtime["accounts"]]), + ) + monkeypatch.setattr( + task_endpoint, + "resolve_task_candidates", + lambda *args, **kwargs: [("http://llm", "model", {})], + ) + + async def fake_fallback(*_args, **_kwargs): + return '{"score": 3, "reason": "urgent"}' + + monkeypatch.setattr(llm_core, "llm_call_async_with_fallback", fake_fallback) + monkeypatch.setattr(settings, "load_settings", lambda: dict(runtime["settings"])) + monkeypatch.setattr( + email_helpers, + "SCHEDULED_DB", + tmp_path / "scheduled-emails.db", + ) + monkeypatch.setattr( + email_helpers, + "_imap_connect", + lambda account_id=None, **_kwargs: _FakeImap( + str(account_id), + runtime["failures"], + runtime["seen_accounts"], + runtime["search_uids"], + ), + ) + return builtin_actions, runtime + + +@pytest.mark.asyncio +async def test_urgency_state_transaction_serializes_decision_and_checkpoint(tmp_path): + """A later worker must observe the first worker's delivered UID.""" + from src.builtin_actions import _run_email_urgency_state_transaction + + state_path = tmp_path / "email_urgency_state_alice.json" + lock_db = tmp_path / "urgency.lock.sqlite3" + first_entered = asyncio.Event() + allow_first_to_finish = asyncio.Event() + second_entered = asyncio.Event() + deliveries = [] + + async def operation(name): + async def update(prior): + if name == "first": + first_entered.set() + await allow_first_to_finish.wait() + else: + second_entered.set() + + notified = set(prior.get("notified_uids", [])) + delivered = "acct:42" not in notified + if delivered: + deliveries.append(name) + notified.add("acct:42") + return delivered, { + "owner": "alice", + "notified_uids": sorted(notified), + } + + return await _run_email_urgency_state_transaction( + state_path, + lock_db, + update, + ) + + first = asyncio.create_task(operation("first")) + await asyncio.wait_for(first_entered.wait(), timeout=2) + second = asyncio.create_task(operation("second")) + await asyncio.sleep(0.1) + assert not second_entered.is_set() + allow_first_to_finish.set() + + assert await asyncio.wait_for(first, timeout=2) is True + assert await asyncio.wait_for(second, timeout=2) is False + assert deliveries == ["first"] + assert json.loads(state_path.read_text(encoding="utf-8")) == { + "owner": "alice", + "notified_uids": ["acct:42"], + } + + +def test_stale_complete_scan_preserves_newer_state_and_discards_stale_only_facts(): + from src.builtin_actions import _merge_email_urgency_state + + prior = { + "owner": "alice", + "per_uid": { + "acct:1": {"score": 0, "unread": True, "reason": "newer"}, + "acct:2": {"score": 2, "unread": True, "reason": "new UID"}, + }, + "notified_uids": ["acct:2"], + "account_generations": { + "acct": {"checkpoint": 1, "complete": 1}, + }, + } + + merged = _merge_email_urgency_state( + prior, + owner="alice", + per_uid_scores={ + "acct:1": {"score": 0, "unread": False, "reason": "older"}, + "acct:3": {"score": 3, "unread": True, "reason": "also observed"}, + }, + notified_uids={"acct:1", "acct:2", "acct:3"}, + all_unread_keys={"acct:3"}, + fully_scanned_account_ids={"acct"}, + base_account_generations={ + "acct": {"checkpoint": 0, "complete": 0}, + }, + timestamp=300.0, + ) + + assert merged["per_uid"]["acct:1"]["reason"] == "newer" + assert set(merged["per_uid"]) == {"acct:1", "acct:2"} + assert merged["notified_uids"] == ["acct:2"] + assert merged["account_generations"]["acct"] == { + "checkpoint": 1, + "complete": 1, + } + + +def test_newer_partial_checkpoint_fences_older_complete_snapshot(): + from src.builtin_actions import _merge_email_urgency_state + + legacy = { + "owner": "alice", + "per_uid": { + "acct:1": {"score": 3, "unread": True}, + }, + "notified_uids": ["acct:1"], + } + partial = _merge_email_urgency_state( + legacy, + owner="alice", + per_uid_scores={ + "acct:2": {"score": 3, "unread": True}, + }, + notified_uids={"acct:1", "acct:2"}, + all_unread_keys={"acct:2"}, + fully_scanned_account_ids=set(), + base_account_generations={}, + timestamp=200.0, + ) + assert partial["account_generations"]["acct"] == { + "checkpoint": 1, + "complete": 0, + } + + merged = _merge_email_urgency_state( + partial, + owner="alice", + per_uid_scores={ + "acct:1": {"score": 3, "unread": True}, + }, + notified_uids={"acct:1", "acct:2"}, + all_unread_keys={"acct:1"}, + fully_scanned_account_ids={"acct"}, + base_account_generations={}, + timestamp=300.0, + ) + + assert set(merged["per_uid"]) == {"acct:1", "acct:2"} + assert merged["notified_uids"] == ["acct:1", "acct:2"] + assert merged["account_generations"]["acct"] == { + "checkpoint": 1, + "complete": 0, + } + + +def test_authoritative_retirement_prunes_payload_and_recomputes_api_totals(): + from src.builtin_actions import _merge_email_urgency_state + + prior = { + "owner": "alice", + "per_uid": { + "acct-a:1": {"score": 1, "unread": True}, + "acct-b:1": {"score": 3, "unread": True}, + }, + "notified_uids": ["acct-b:1"], + "account_generations": { + "acct-a": {"checkpoint": 1, "complete": 1}, + "acct-b": {"checkpoint": 1, "complete": 1}, + }, + } + + merged = _merge_email_urgency_state( + prior, + owner="alice", + per_uid_scores={}, + notified_uids=prior["notified_uids"], + all_unread_keys=set(), + fully_scanned_account_ids=set(), + base_account_generations=prior["account_generations"], + timestamp=300.0, + retired_account_ids={"acct-b"}, + base_payload_account_ids={"acct-a", "acct-b"}, + ) + + assert merged["per_uid"] == { + "acct-a:1": {"score": 1, "unread": True}, + } + assert merged["notified_uids"] == [] + assert merged["total_unread"] == 1 + assert merged["total_urgent"] == 0 + assert merged["max_score"] == 1 + assert merged["account_generations"]["acct-b"] == { + "checkpoint": 2, + "complete": 1, + } + + +def test_authoritative_retirement_respects_generation_and_membership_fences(): + from src.builtin_actions import _merge_email_urgency_state + + newer = { + "owner": "alice", + "per_uid": { + "acct-b:2": {"score": 3, "unread": True, "reason": "newer"}, + }, + "notified_uids": ["acct-b:2"], + "account_generations": { + "acct-b": {"checkpoint": 2, "complete": 2}, + }, + } + generation_fenced = _merge_email_urgency_state( + newer, + owner="alice", + per_uid_scores={}, + notified_uids=newer["notified_uids"], + all_unread_keys=set(), + fully_scanned_account_ids=set(), + base_account_generations={ + "acct-b": {"checkpoint": 1, "complete": 1}, + }, + timestamp=300.0, + retired_account_ids={"acct-b"}, + base_payload_account_ids={"acct-b"}, + ) + assert generation_fenced["per_uid"] == newer["per_uid"] + assert generation_fenced["notified_uids"] == ["acct-b:2"] + assert generation_fenced["account_generations"]["acct-b"] == { + "checkpoint": 2, + "complete": 2, + } + + legacy_first_write = { + "owner": "alice", + "per_uid": { + "acct-b:3": {"score": 2, "unread": True, "reason": "concurrent"}, + }, + "notified_uids": [], + } + membership_fenced = _merge_email_urgency_state( + legacy_first_write, + owner="alice", + per_uid_scores={}, + notified_uids=[], + all_unread_keys=set(), + fully_scanned_account_ids=set(), + base_account_generations={}, + timestamp=300.0, + retired_account_ids={"acct-b"}, + base_payload_account_ids=set(), + ) + assert membership_fenced["per_uid"] == legacy_first_write["per_uid"] + + +@pytest.mark.asyncio +async def test_waiting_transaction_cancellation_does_not_leak_lock(tmp_path): + from src.builtin_actions import _run_email_urgency_state_transaction + + state_path = tmp_path / "email_urgency_state_alice.json" + lock_db = tmp_path / "urgency.lock.sqlite3" + first_entered = asyncio.Event() + release_first = asyncio.Event() + + async def first_operation(_prior): + first_entered.set() + await release_first.wait() + return None, {"notified_uids": ["acct:1"]} + + async def later_operation(prior): + return None, prior + + first = asyncio.create_task( + _run_email_urgency_state_transaction( + state_path, lock_db, first_operation + ) + ) + await asyncio.wait_for(first_entered.wait(), timeout=2) + waiting = asyncio.create_task( + _run_email_urgency_state_transaction( + state_path, lock_db, later_operation + ) + ) + await asyncio.sleep(0.05) + waiting.cancel() + with pytest.raises(asyncio.CancelledError): + await asyncio.wait_for(waiting, timeout=2) + + release_first.set() + await asyncio.wait_for(first, timeout=2) + await asyncio.wait_for( + _run_email_urgency_state_transaction( + state_path, lock_db, later_operation + ), + timeout=2, + ) + + +@pytest.mark.asyncio +async def test_action_dispatch_stays_on_app_loop_and_queues_browser_notification( + monkeypatch, + tmp_path, +): + import routes.note_routes as note_routes + from src import endpoint_resolver, llm_core + from src.task_scheduler import TaskScheduler + + builtin_actions, runtime = _configure_action( + monkeypatch, tmp_path, ["acct-a"] + ) + runtime["settings"]["reminder_llm_synthesis"] = True + monkeypatch.setattr(note_routes, "DATA_DIR", str(tmp_path)) + monkeypatch.setattr( + endpoint_resolver, + "resolve_endpoint", + lambda *_args, **_kwargs: ( + "https://api.openai.com/v1", + "utility-model", + {}, + ), + ) + + expected_loop = asyncio.get_running_loop() + expected_thread = threading.get_ident() + synthesis_loops = [] + shared_client = SimpleNamespace(is_closed=False) + monkeypatch.setattr(llm_core, "_http_client", shared_client) + monkeypatch.setattr(llm_core, "_response_cache", {}) + + class _Response: + is_success = True + status_code = 200 + text = "ok" + + @staticmethod + def json(): + return { + "choices": [ + {"message": {"content": "Synthesized urgency reminder."}} + ] + } + + async def fake_http_post(client, *_args, **_kwargs): + synthesis_loops.append(asyncio.get_running_loop()) + assert client is shared_client + return _Response() + + monkeypatch.setattr( + llm_core, + "httpx_post_kimi_aware_async", + fake_http_post, + ) + + scheduler = TaskScheduler(None) + notification_threads = [] + original_add = scheduler.add_notification + + def checked_add(*args, **kwargs): + notification_threads.append(threading.get_ident()) + return original_add(*args, **kwargs) + + monkeypatch.setattr(scheduler, "add_notification", checked_add) + monkeypatch.setattr(note_routes, "_scheduler_ref", scheduler) + + message, ok = await builtin_actions.action_check_email_urgency("alice") + + assert ok is True + assert "notified 1" in message + assert synthesis_loops == [expected_loop] + assert notification_threads == [expected_thread] + notifications = scheduler.pop_notifications(owner="alice") + assert len(notifications) == 1 + assert notifications[0]["body"] == "Synthesized urgency reminder." + + +@pytest.mark.asyncio +async def test_action_cancellation_rolls_back_without_checkpoint( + monkeypatch, + tmp_path, +): + import routes.note_routes as note_routes + + builtin_actions, _runtime = _configure_action( + monkeypatch, tmp_path, ["acct-a"] + ) + state_path = tmp_path / "email_urgency_state_alice.json" + state_path.write_text( + json.dumps({"owner": "alice", "per_uid": {}, "notified_uids": []}), + encoding="utf-8", + ) + entered = threading.Event() + release = threading.Event() + dispatch_cancelled = threading.Event() + + async def blocked_dispatch(**_kwargs): + entered.set() + try: + while not release.is_set(): + await asyncio.sleep(0.01) + except asyncio.CancelledError: + dispatch_cancelled.set() + raise + return { + "browser_sent": True, + "email_sent": False, + "ntfy_sent": False, + "webhook_sent": False, + } + + monkeypatch.setattr(note_routes, "dispatch_reminder", blocked_dispatch) + task = asyncio.create_task( + builtin_actions.action_check_email_urgency("alice") + ) + for _ in range(200): + if entered.is_set(): + break + await asyncio.sleep(0.01) + assert entered.is_set() + + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + release.set() + await asyncio.sleep(0.1) + + assert dispatch_cancelled.is_set() + state = json.loads(state_path.read_text(encoding="utf-8")) + assert state["owner"] == "alice" + assert state["per_uid"] == {} + assert state["notified_uids"] == [] + # The pre-scan active marker is not a delivered/checkpointed UID. It must + # survive cancellation so a concurrent deletion cleanup can fence this + # first-ever account scan. + assert state["account_generations"] == { + "acct-a": {"checkpoint": 0, "complete": 0}, + } + + +@pytest.mark.asyncio +async def test_first_scan_revalidates_account_deleted_before_registration( + monkeypatch, + tmp_path, +): + import routes.note_routes as note_routes + + builtin_actions, runtime = _configure_action( + monkeypatch, tmp_path, ["acct-a"] + ) + deliveries = [] + + async def delivered(**kwargs): + deliveries.append(kwargs["note_body"]) + return { + "browser_sent": True, + "email_sent": False, + "ntfy_sent": False, + "webhook_sent": False, + } + + monkeypatch.setattr(note_routes, "dispatch_reminder", delivered) + original_transaction = builtin_actions._run_email_urgency_state_transaction + + async def delete_before_registration(state_path, lock_db_path, operation): + if operation.__name__ == "_register_accounts": + # The initial DB read saw the account, but deletion commits before + # its first active marker. The post-registration enumeration must + # observe that absence and retire the marker without touching IMAP. + runtime["accounts"] = [] + return await original_transaction(state_path, lock_db_path, operation) + + monkeypatch.setattr( + builtin_actions, + "_run_email_urgency_state_transaction", + delete_before_registration, + ) + + with pytest.raises(builtin_actions.TaskNoop): + await builtin_actions.action_check_email_urgency("alice") + + state = json.loads( + (tmp_path / "email_urgency_state_alice.json").read_text(encoding="utf-8") + ) + assert deliveries == [] + assert state["per_uid"] == {} + assert state["notified_uids"] == [] + assert state["account_generations"]["acct-a"] == { + "checkpoint": 1, + "complete": 0, + } + + +@pytest.mark.asyncio +async def test_first_scan_deleted_after_revalidation_is_fenced_by_cleanup_pass( + monkeypatch, + tmp_path, +): + import routes.note_routes as note_routes + + builtin_actions, runtime = _configure_action( + monkeypatch, tmp_path, ["acct-a"] + ) + deliveries = [] + + async def delivered(**kwargs): + deliveries.append(kwargs["note_body"]) + return { + "browser_sent": True, + "email_sent": False, + "ntfy_sent": False, + "webhook_sent": False, + } + + monkeypatch.setattr(note_routes, "dispatch_reminder", delivered) + original_transaction = builtin_actions._run_email_urgency_state_transaction + scanned = asyncio.Event() + release_scan = asyncio.Event() + paused = False + + async def pause_first_delivery(state_path, lock_db_path, operation): + nonlocal paused + if operation.__name__ == "_dispatch_and_checkpoint" and not paused: + paused = True + scanned.set() + await release_scan.wait() + return await original_transaction(state_path, lock_db_path, operation) + + monkeypatch.setattr( + builtin_actions, + "_run_email_urgency_state_transaction", + pause_first_delivery, + ) + + stale = asyncio.create_task( + builtin_actions.action_check_email_urgency("alice") + ) + await asyncio.wait_for(scanned.wait(), timeout=2) + registered = json.loads( + (tmp_path / "email_urgency_state_alice.json").read_text(encoding="utf-8") + ) + assert registered["account_generations"]["acct-a"] == { + "checkpoint": 0, + "complete": 0, + } + + # Public-action boundary: deletion itself does not mutate urgency state. + # The authoritative zero-account pass is what advances the active marker + # before the paused scan reaches any reminder channel. + runtime["accounts"] = [] + with pytest.raises(builtin_actions.TaskNoop): + await builtin_actions.action_check_email_urgency("alice") + release_scan.set() + await asyncio.wait_for(stale, timeout=2) + + state = json.loads( + (tmp_path / "email_urgency_state_alice.json").read_text(encoding="utf-8") + ) + assert deliveries == [] + assert state["per_uid"] == {} + assert state["notified_uids"] == [] + assert state["total_unread"] == 0 + assert state["total_urgent"] == 0 + assert state["account_generations"]["acct-a"] == { + "checkpoint": 1, + "complete": 0, + } + + +def test_scan_keeps_registration_generation_when_cleanup_precedes_basis( + monkeypatch, + tmp_path, +): + """Cleanup after verification cannot become the stale scan's baseline.""" + from core import database + import routes.note_routes as note_routes + + builtin_actions, runtime = _configure_action( + monkeypatch, tmp_path, ["acct-a"] + ) + deliveries = [] + + async def delivered(**kwargs): + deliveries.append(kwargs["note_body"]) + return { + "browser_sent": True, + "email_sent": False, + "ntfy_sent": False, + "webhook_sent": False, + } + + monkeypatch.setattr(note_routes, "dispatch_reminder", delivered) + verified_account_selected = threading.Event() + release_verified_account = threading.Event() + stale_query_count = 0 + query_count_lock = threading.Lock() + + class _PausingQuery(_Query): + def all(self): + nonlocal stale_query_count + rows = list(self._rows) + if threading.current_thread().name == "stale-urgency-scan": + with query_count_lock: + stale_query_count += 1 + should_pause = stale_query_count == 2 + if should_pause: + # The second enumeration has selected the enabled row, but + # the action has not yet retained/used its checkpoint basis. + verified_account_selected.set() + assert release_verified_account.wait(5) + return rows + + class _PausingDb(_Db): + def query(self, _model): + return _PausingQuery(self._rows()) + + monkeypatch.setattr( + database, + "SessionLocal", + lambda: _PausingDb( + lambda: [_account(value) for value in runtime["accounts"]] + ), + ) + + stale_result = {} + + def run_stale_scan(): + try: + stale_result["value"] = asyncio.run( + builtin_actions.action_check_email_urgency("alice") + ) + except BaseException as exc: + stale_result["error"] = exc + + worker = threading.Thread( + target=run_stale_scan, + name="stale-urgency-scan", + ) + worker.start() + assert verified_account_selected.wait(5) + + runtime["accounts"] = [] + with pytest.raises(builtin_actions.TaskNoop): + asyncio.run(builtin_actions.action_check_email_urgency("alice")) + state_path = tmp_path / "email_urgency_state_alice.json" + retired = json.loads(state_path.read_text(encoding="utf-8")) + assert retired["account_generations"]["acct-a"] == { + "checkpoint": 1, + "complete": 0, + } + + release_verified_account.set() + worker.join(timeout=5) + assert not worker.is_alive() + assert "error" not in stale_result + + state = json.loads(state_path.read_text(encoding="utf-8")) + assert deliveries == [] + assert state["per_uid"] == {} + assert state["notified_uids"] == [] + assert state["account_generations"]["acct-a"] == { + "checkpoint": 1, + "complete": 0, + } + + +@pytest.mark.asyncio +async def test_payload_empty_tombstone_fences_reenable_redelete_and_can_recover( + monkeypatch, + tmp_path, +): + import routes.note_routes as note_routes + + builtin_actions, runtime = _configure_action( + monkeypatch, tmp_path, ["acct-a"] + ) + deliveries = [] + + async def delivered(**kwargs): + deliveries.append(kwargs["note_body"]) + return { + "browser_sent": True, + "email_sent": False, + "ntfy_sent": False, + "webhook_sent": False, + } + + monkeypatch.setattr(note_routes, "dispatch_reminder", delivered) + await builtin_actions.action_check_email_urgency("alice") + assert len(deliveries) == 1 + + runtime["accounts"] = [] + with pytest.raises(builtin_actions.TaskNoop): + await builtin_actions.action_check_email_urgency("alice") + first_tombstone = json.loads( + (tmp_path / "email_urgency_state_alice.json").read_text(encoding="utf-8") + ) + assert first_tombstone["per_uid"] == {} + assert first_tombstone["account_generations"]["acct-a"] == { + "checkpoint": 2, + "complete": 1, + } + + original_transaction = builtin_actions._run_email_urgency_state_transaction + scanned = asyncio.Event() + release_scan = asyncio.Event() + pause_reenabled = True + + async def pause_reenabled_delivery(state_path, lock_db_path, operation): + nonlocal pause_reenabled + if operation.__name__ == "_dispatch_and_checkpoint" and pause_reenabled: + pause_reenabled = False + scanned.set() + await release_scan.wait() + return await original_transaction(state_path, lock_db_path, operation) + + monkeypatch.setattr( + builtin_actions, + "_run_email_urgency_state_transaction", + pause_reenabled_delivery, + ) + deliveries.clear() + runtime["accounts"] = ["acct-a"] + stale_reenabled = asyncio.create_task( + builtin_actions.action_check_email_urgency("alice") + ) + await asyncio.wait_for(scanned.wait(), timeout=2) + + # Delete/disable again while the re-enabled scan is based on checkpoint 2. + # Discovery must include the payload-empty generation tombstone and advance + # it, otherwise the paused scan would deliver and resurrect acct-a:1. + runtime["accounts"] = [] + with pytest.raises(builtin_actions.TaskNoop): + await builtin_actions.action_check_email_urgency("alice") + release_scan.set() + await asyncio.wait_for(stale_reenabled, timeout=2) + + fenced = json.loads( + (tmp_path / "email_urgency_state_alice.json").read_text(encoding="utf-8") + ) + assert deliveries == [] + assert fenced["per_uid"] == {} + assert fenced["notified_uids"] == [] + assert fenced["account_generations"]["acct-a"] == { + "checkpoint": 3, + "complete": 1, + } + + # A later authoritative pass advances the empty tombstone again, fencing + # any scan that captured checkpoint 3 before this absence was confirmed. + with pytest.raises(builtin_actions.TaskNoop): + await builtin_actions.action_check_email_urgency("alice") + repeated = json.loads( + (tmp_path / "email_urgency_state_alice.json").read_text(encoding="utf-8") + ) + assert repeated["account_generations"]["acct-a"] == { + "checkpoint": 4, + "complete": 1, + } + + # Re-enabling after the latest tombstone captures checkpoint 4 and can + # publish fresh payload normally. + runtime["accounts"] = ["acct-a"] + await builtin_actions.action_check_email_urgency("alice") + recovered = json.loads( + (tmp_path / "email_urgency_state_alice.json").read_text(encoding="utf-8") + ) + assert len(deliveries) == 1 + assert set(recovered["per_uid"]) == {"acct-a:1"} + assert recovered["notified_uids"] == ["acct-a:1"] + assert recovered["account_generations"]["acct-a"] == { + "checkpoint": 5, + "complete": 2, + } + + +@pytest.mark.parametrize( + ("stale_uids", "newer_uids"), + [ + (["1"], ["1", "2"]), + (["1", "2"], ["1"]), + ], + ids=["preserve-newer-addition", "reject-stale-only-delivery"], +) +@pytest.mark.asyncio +async def test_stale_same_account_scan_cannot_override_newer_commit( + monkeypatch, + tmp_path, + stale_uids, + newer_uids, +): + import routes.note_routes as note_routes + + builtin_actions, runtime = _configure_action( + monkeypatch, tmp_path, ["acct-a"] + ) + runtime["search_uids"]["acct-a"] = stale_uids + deliveries = [] + + async def delivered(**kwargs): + deliveries.append(kwargs["note_body"]) + return { + "browser_sent": True, + "email_sent": False, + "ntfy_sent": False, + "webhook_sent": False, + } + + monkeypatch.setattr(note_routes, "dispatch_reminder", delivered) + original_transaction = builtin_actions._run_email_urgency_state_transaction + stale_scan_ready = asyncio.Event() + release_stale_scan = asyncio.Event() + transaction_count = 0 + + async def order_transactions(state_path, lock_db_path, operation): + nonlocal transaction_count + if operation.__name__ == "_dispatch_and_checkpoint": + transaction_count += 1 + if transaction_count == 1: + stale_scan_ready.set() + await release_stale_scan.wait() + return await original_transaction(state_path, lock_db_path, operation) + + monkeypatch.setattr( + builtin_actions, + "_run_email_urgency_state_transaction", + order_transactions, + ) + + stale = asyncio.create_task( + builtin_actions.action_check_email_urgency("alice") + ) + await asyncio.wait_for(stale_scan_ready.wait(), timeout=2) + + runtime["search_uids"]["acct-a"] = newer_uids + newer_result = await asyncio.wait_for( + builtin_actions.action_check_email_urgency("alice"), + timeout=2, + ) + release_stale_scan.set() + stale_result = await asyncio.wait_for(stale, timeout=2) + + state = json.loads( + (tmp_path / "email_urgency_state_alice.json").read_text(encoding="utf-8") + ) + assert newer_result[1] is True + assert stale_result[1] is True + assert len(deliveries) == 1 + for uid in newer_uids: + assert f"uid {uid}" in deliveries[0] + for uid in set(stale_uids) - set(newer_uids): + assert f"uid {uid}" not in deliveries[0] + expected_keys = {f"acct-a:{uid}" for uid in newer_uids} + assert set(state["per_uid"]) == expected_keys + assert state["notified_uids"] == sorted(expected_keys) + assert state["account_generations"]["acct-a"] == { + "checkpoint": 1, + "complete": 1, + } + + +@pytest.mark.asyncio +async def test_mixed_stale_and_fresh_accounts_exclude_stale_rows_from_delivery( + monkeypatch, + tmp_path, +): + import routes.note_routes as note_routes + + builtin_actions, runtime = _configure_action( + monkeypatch, tmp_path, ["acct-a", "acct-b"] + ) + deliveries = [] + + async def delivered(**kwargs): + deliveries.append(kwargs["note_body"]) + return { + "browser_sent": True, + "email_sent": False, + "ntfy_sent": False, + "webhook_sent": False, + } + + monkeypatch.setattr(note_routes, "dispatch_reminder", delivered) + original_transaction = builtin_actions._run_email_urgency_state_transaction + stale_scan_ready = asyncio.Event() + release_stale_scan = asyncio.Event() + transaction_count = 0 + + async def order_transactions(state_path, lock_db_path, operation): + nonlocal transaction_count + if operation.__name__ == "_dispatch_and_checkpoint": + transaction_count += 1 + if transaction_count == 1: + stale_scan_ready.set() + await release_stale_scan.wait() + return await original_transaction(state_path, lock_db_path, operation) + + monkeypatch.setattr( + builtin_actions, + "_run_email_urgency_state_transaction", + order_transactions, + ) + + stale = asyncio.create_task( + builtin_actions.action_check_email_urgency("alice") + ) + await asyncio.wait_for(stale_scan_ready.wait(), timeout=2) + + runtime["accounts"] = ["acct-a"] + await asyncio.wait_for( + builtin_actions.action_check_email_urgency( + "alice", + prompt='{"account_id":"acct-a"}', + ), + timeout=2, + ) + release_stale_scan.set() + await asyncio.wait_for(stale, timeout=2) + + state = json.loads( + (tmp_path / "email_urgency_state_alice.json").read_text(encoding="utf-8") + ) + assert len(deliveries) == 2 + assert "acct-a" in deliveries[0] + assert "acct-b" not in deliveries[0] + assert "acct-b" in deliveries[1] + assert "acct-a" not in deliveries[1] + assert set(state["per_uid"]) == {"acct-a:1", "acct-b:1"} + assert state["notified_uids"] == ["acct-a:1", "acct-b:1"] + + +@pytest.mark.asyncio +async def test_account_scoped_actions_merge_disjoint_checkpoints( + monkeypatch, + tmp_path, +): + import routes.note_routes as note_routes + + builtin_actions, runtime = _configure_action( + monkeypatch, tmp_path, ["acct-a"] + ) + deliveries = [] + + async def delivered(**kwargs): + deliveries.append(kwargs["note_body"]) + return { + "browser_sent": True, + "email_sent": False, + "ntfy_sent": False, + "webhook_sent": False, + } + + monkeypatch.setattr(note_routes, "dispatch_reminder", delivered) + + await builtin_actions.action_check_email_urgency( + "alice", prompt='{"account_id":"acct-a"}' + ) + runtime["accounts"] = ["acct-b"] + await builtin_actions.action_check_email_urgency( + "alice", prompt='{"account_id":"acct-b"}' + ) + runtime["accounts"] = ["acct-a"] + await builtin_actions.action_check_email_urgency( + "alice", prompt='{"account_id":"acct-a"}' + ) + + state = json.loads( + (tmp_path / "email_urgency_state_alice.json").read_text(encoding="utf-8") + ) + assert len(deliveries) == 2 + assert "Urgent request for acct-a" in deliveries[0] + assert "Urgent request for acct-b" in deliveries[1] + assert state["notified_uids"] == ["acct-a:1", "acct-b:1"] + assert set(state["per_uid"]) == {"acct-a:1", "acct-b:1"} + assert state["total_unread"] == 2 + assert state["total_urgent"] == 2 + + +@pytest.mark.asyncio +async def test_full_enumeration_retires_deleted_or_disabled_account( + monkeypatch, + tmp_path, +): + import routes.note_routes as note_routes + + builtin_actions, runtime = _configure_action( + monkeypatch, tmp_path, ["acct-a", "acct-b"] + ) + + async def delivered(**_kwargs): + return { + "browser_sent": True, + "email_sent": False, + "ntfy_sent": False, + "webhook_sent": False, + } + + monkeypatch.setattr(note_routes, "dispatch_reminder", delivered) + await builtin_actions.action_check_email_urgency("alice") + + # The production query returns only enabled, owner-visible accounts. A + # deleted row and a disabled row are therefore the same authoritative + # absence at this boundary. + runtime["accounts"] = ["acct-a"] + await builtin_actions.action_check_email_urgency("alice") + + state = json.loads( + (tmp_path / "email_urgency_state_alice.json").read_text(encoding="utf-8") + ) + assert set(state["per_uid"]) == {"acct-a:1"} + assert state["notified_uids"] == ["acct-a:1"] + assert state["total_unread"] == 1 + assert state["total_urgent"] == 1 + assert state["max_score"] == 3 + assert state["account_generations"]["acct-b"] == { + "checkpoint": 2, + "complete": 1, + } + + +@pytest.mark.asyncio +async def test_zero_enabled_accounts_cleanup_precedes_model_resolution( + monkeypatch, + tmp_path, +): + import routes.note_routes as note_routes + from src import task_endpoint + + builtin_actions, runtime = _configure_action( + monkeypatch, tmp_path, ["acct-a"] + ) + + async def delivered(**_kwargs): + return { + "browser_sent": True, + "email_sent": False, + "ntfy_sent": False, + "webhook_sent": False, + } + + monkeypatch.setattr(note_routes, "dispatch_reminder", delivered) + await builtin_actions.action_check_email_urgency("alice") + runtime["accounts"] = [] + + model_resolution_owners = [] + + def no_model_available(*_args, **kwargs): + model_resolution_owners.append(kwargs.get("owner")) + return [] + + monkeypatch.setattr( + task_endpoint, + "resolve_task_candidates", + no_model_available, + ) + with pytest.raises(builtin_actions.TaskNoop): + await builtin_actions.action_check_email_urgency("alice") + assert model_resolution_owners == ["alice"] + + state = json.loads( + (tmp_path / "email_urgency_state_alice.json").read_text(encoding="utf-8") + ) + assert state["per_uid"] == {} + assert state["notified_uids"] == [] + assert state["total_unread"] == 0 + assert state["total_urgent"] == 0 + assert state["max_score"] == 0 + assert state["account_generations"]["acct-a"] == { + "checkpoint": 2, + "complete": 1, + } + + +@pytest.mark.asyncio +async def test_scoped_missing_account_retires_only_selected_payload( + monkeypatch, + tmp_path, +): + import routes.note_routes as note_routes + + builtin_actions, runtime = _configure_action( + monkeypatch, tmp_path, ["acct-a", "acct-b"] + ) + + async def delivered(**_kwargs): + return { + "browser_sent": True, + "email_sent": False, + "ntfy_sent": False, + "webhook_sent": False, + } + + monkeypatch.setattr(note_routes, "dispatch_reminder", delivered) + await builtin_actions.action_check_email_urgency("alice") + + # A production account-id filter returns no row when the selected account + # was deleted or disabled. Other accounts are outside this scoped query and + # must remain untouched. + runtime["accounts"] = [] + with pytest.raises(builtin_actions.TaskNoop): + await builtin_actions.action_check_email_urgency( + "alice", + prompt='{"account_id":"acct-b"}', + ) + + state = json.loads( + (tmp_path / "email_urgency_state_alice.json").read_text(encoding="utf-8") + ) + assert set(state["per_uid"]) == {"acct-a:1"} + assert state["notified_uids"] == ["acct-a:1"] + assert state["total_unread"] == 1 + assert state["total_urgent"] == 1 + assert state["account_generations"]["acct-b"] == { + "checkpoint": 2, + "complete": 1, + } + + +@pytest.mark.asyncio +async def test_failed_account_scan_preserves_checkpoint_until_recovery( + monkeypatch, + tmp_path, +): + import routes.note_routes as note_routes + + builtin_actions, runtime = _configure_action( + monkeypatch, tmp_path, ["acct-a", "acct-b"] + ) + deliveries = [] + + async def delivered(**kwargs): + deliveries.append(kwargs["note_body"]) + return { + "browser_sent": True, + "email_sent": False, + "ntfy_sent": False, + "webhook_sent": False, + } + + monkeypatch.setattr(note_routes, "dispatch_reminder", delivered) + + await builtin_actions.action_check_email_urgency("alice") + runtime["failures"] = {"acct-b"} + await builtin_actions.action_check_email_urgency("alice") + + failed_state = json.loads( + (tmp_path / "email_urgency_state_alice.json").read_text(encoding="utf-8") + ) + assert failed_state["notified_uids"] == ["acct-a:1", "acct-b:1"] + assert set(failed_state["per_uid"]) == {"acct-a:1", "acct-b:1"} + assert failed_state["total_unread"] == 2 + assert failed_state["total_urgent"] == 2 + assert failed_state["account_generations"]["acct-b"] == { + "checkpoint": 1, + "complete": 1, + } + + runtime["failures"] = set() + runtime["accounts"] = ["acct-b"] + await builtin_actions.action_check_email_urgency( + "alice", prompt='{"account_id":"acct-b"}' + ) + + assert len(deliveries) == 1 + + +@pytest.mark.asyncio +async def test_cached_flags_refresh_prunes_checkpoint_after_message_is_read( + monkeypatch, + tmp_path, +): + import routes.note_routes as note_routes + + builtin_actions, runtime = _configure_action( + monkeypatch, tmp_path, ["acct-a"] + ) + + async def delivered(**_kwargs): + return { + "browser_sent": True, + "email_sent": False, + "ntfy_sent": False, + "webhook_sent": False, + } + + monkeypatch.setattr(note_routes, "dispatch_reminder", delivered) + await builtin_actions.action_check_email_urgency("alice") + runtime["seen_accounts"] = {"acct-a"} + await builtin_actions.action_check_email_urgency("alice") + + state = json.loads( + (tmp_path / "email_urgency_state_alice.json").read_text(encoding="utf-8") + ) + assert state["notified_uids"] == [] + assert state["per_uid"]["acct-a:1"]["unread"] is False + assert state["total_unread"] == 0 diff --git a/tests/test_llm_core_thinking_models.py b/tests/test_llm_core_thinking_models.py new file mode 100644 index 000000000..4ed4fc3c8 --- /dev/null +++ b/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 diff --git a/tests/test_rename_user_owner_sync.py b/tests/test_rename_user_owner_sync.py index 7e9e5d911..b7ac7e200 100644 --- a/tests/test_rename_user_owner_sync.py +++ b/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 diff --git a/tests/test_teacher_audit_owner_scope.py b/tests/test_teacher_audit_owner_scope.py index 5bd6228d9..8e398d9bb 100644 --- a/tests/test_teacher_audit_owner_scope.py +++ b/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):