feat(settings): sync user preferences across authenticated clients

README.md

@@ -394,6 +394,21 @@ DeerFlow runs the agent runtime inside the Gateway API. Development mode enables
 
 Gateway owns `/api/langgraph/*` and translates those public LangGraph-compatible paths to its native `/api/*` routers behind nginx.
 
+With Gateway authentication enabled, the Web UI also persists an allowlisted
+set of user-level display and model defaults through the owner-scoped
+`/api/user-preferences` API. An existing server record is authoritative at
+sign-in; otherwise DeerFlow imports the browser's valid local base settings
+once. Local storage remains the offline fallback. Thread overrides, browser
+notification permission/system state, workspace data, and credentials stay
+local and are never accepted by this API. Auth-disabled deployments keep the
+original local-only settings behavior. Authenticated fallback caches are keyed
+by account. On browsers with cross-tab Web Locks, only one authenticated
+account can claim the old unscoped cache; without that lock, DeerFlow safely
+skips the ambiguous legacy import instead of copying it across accounts.
+If an older or manually edited server record no longer matches the current
+settings schema, DeerFlow clears only that observed revision and rebuilds it
+from the browser's complete current fallback without dropping a pending edit.
+
 #### LangGraph Studio (Optional)
 
 The default `make dev` topology uses DeerFlow's Gateway-embedded runtime and

backend/app/gateway/AGENTS.md

@@ -44,7 +44,8 @@ reads/searches.
 | Router | Endpoints |
 |--------|-----------|
 | **Models** (`/api/models`) | `GET /` - list models; `GET /{name}` - model details |
-| **Features** (`/api/features`) | `GET /` - UI capabilities: hot-reloaded agents, guarded browser, startup MCP tasks, and separate batch repository/worker states so history stays readable without a worker |
+| **Features** (`/api/features`) | Flags for agents, browser, MCP tasks, and subagent-batch repository/worker availability |
+| **Preferences** (`/api/user-preferences`) | Owner-scoped allowlisted settings sync; corrupt records are CAS-cleared for client recovery |
 | **Console** (`/api/console`) | Read-only cross-thread observability for the current user (the data layer for an operations dashboard or external monitoring): `GET /stats` - headline counters (runs/threads/agents/tokens/cost); `GET /runs` - paginated run history joined with thread titles (per-run cost); `GET /usage` - zero-filled daily token series + per-model breakdown with spend. Queries `runs`/`threads_meta` directly as a reporting layer (no new `RunStore` methods); requires a SQL database backend — returns 503 on `database.backend: memory`. Real-cost estimation reads optional `models[*].pricing` (`currency`, `input_per_million`, `output_per_million`, `input_cache_hit_per_million`; `ModelConfig` is `extra="allow"`, so no schema change) and prices each run from its `token_usage_by_model` input/output split. Pricing is **cache-aware**: `RunJournal` accumulates prompt-cache hits from `usage_metadata.input_token_details.cache_read` into a sparse `cache_read_tokens` bucket key (also threaded through `SubagentTokenCollector` → `record_external_llm_usage_records`), and cache-hit input tokens are billed at `input_cache_hit_per_million` (omitted → billed at the miss price, a conservative upper bound). All priced models must use one currency; mixed currencies disable cost reporting and leave cost/currency fields null instead of producing invalid aggregates. Legacy rows fall back to run-level totals at `model_name`; unpriced models yield `cost: null` and cost fields are null when no pricing is configured |
 | **MCP** (`/api/mcp`) | `GET /config` - get config; `PUT /config` - replace the full config with whole-payload stdio validation; `PATCH /config` - toggle one server while preserving the raw extensions config and validating only an enabled target; both writes reload config and reset the process-local MCP cache |
 | **MCP Tasks** (`/api/threads/{id}/mcp-tasks`) | `GET /` - current user's durable tasks for one owned thread; `GET /{task_id}` - bounded result/input/status-error/cancellation-error detail, including cancellation attempt count, without remote task IDs or driver configuration |

backend/app/gateway/app.py

@@ -40,6 +40,7 @@
     thread_runs,
     threads,
     uploads,
+    user_preferences,
 )
 from app.gateway.trace_middleware import TraceMiddleware, resolve_trace_enabled
 from deerflow.config import app_config as deerflow_app_config
@@ -841,6 +842,9 @@ def _resolve_extension_principal(request):
     # Auth API is mounted at /api/v1/auth
     app.include_router(auth.router)
 
+    # Authenticated user-level UI settings (server-backed, owner-scoped)
+    app.include_router(user_preferences.router)
+
     # Feedback API is mounted at /api/threads/{thread_id}/runs/{run_id}/feedback
     app.include_router(feedback.router)
 

backend/app/gateway/auth/repositories/base.py

@@ -15,6 +15,14 @@ class UserNotFoundError(LookupError):
     """
 
 
+class UserPreferencesNotInitializedError(LookupError):
+    """Raised when a partial preference update precedes initialization."""
+
+
+class UserPreferencesWriteConflict(RuntimeError):
+    """Raised after bounded optimistic preference-update retries fail."""
+
+
 class UserRepository(ABC):
     """Abstract interface for user data storage.
 
@@ -105,3 +113,23 @@ async def get_user_by_oauth(self, provider: str, oauth_id: str) -> User | None:
             User if found, None otherwise
         """
         raise NotImplementedError
+
+    @abstractmethod
+    async def get_user_preferences(self, user_id: str) -> tuple[dict | None, int]:
+        """Return the user's persisted UI preferences and revision."""
+        raise NotImplementedError
+
+    @abstractmethod
+    async def initialize_user_preferences(self, user_id: str, settings: dict) -> tuple[dict, int]:
+        """Persist settings only when the user has no preference record yet."""
+        raise NotImplementedError
+
+    @abstractmethod
+    async def merge_user_preferences(self, user_id: str, patch: dict) -> tuple[dict | None, int]:
+        """Atomically merge a patch, or clear a structurally corrupt record."""
+        raise NotImplementedError
+
+    @abstractmethod
+    async def reset_user_preferences_if_revision(self, user_id: str, revision: int) -> tuple[dict | None, int]:
+        """Clear an invalid preference record unless another writer replaced it."""
+        raise NotImplementedError

backend/app/gateway/auth/repositories/sqlite.py

@@ -12,17 +12,26 @@
 
 from __future__ import annotations
 
+import sqlite3
+from copy import deepcopy
 from datetime import UTC
 from uuid import UUID
 
-from sqlalchemy import func, select
-from sqlalchemy.exc import IntegrityError
+from sqlalchemy import func, select, update
+from sqlalchemy.exc import IntegrityError, OperationalError
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
 
 from app.gateway.auth.models import User
-from app.gateway.auth.repositories.base import UserNotFoundError, UserRepository
+from app.gateway.auth.repositories.base import (
+    UserNotFoundError,
+    UserPreferencesNotInitializedError,
+    UserPreferencesWriteConflict,
+    UserRepository,
+)
 from deerflow.persistence.user.model import UserRow
 
+_PREFERENCE_WRITE_MAX_ATTEMPTS = 5
+
 
 def _normalize_email(email: str) -> str:
     """Canonicalise an email address for storage and lookup.
@@ -179,3 +188,151 @@ async def get_user_by_oauth(self, provider: str, oauth_id: str) -> User | None:
             result = await session.execute(stmt)
             row = result.scalar_one_or_none()
             return self._row_to_user(row) if row is not None else None
+
+    async def get_user_preferences(self, user_id: str) -> tuple[dict | None, int]:
+        stmt = select(UserRow.preferences, UserRow.preferences_revision).where(UserRow.id == user_id)
+        async with self._sf() as session:
+            row = (await session.execute(stmt)).one_or_none()
+        if row is None:
+            raise UserNotFoundError(f"User {user_id} no longer exists")
+        preferences, revision = row
+        return deepcopy(preferences), int(revision)
+
+    async def initialize_user_preferences(self, user_id: str, settings: dict) -> tuple[dict, int]:
+        """Set the first preference record without overwriting another client.
+
+        The conditional update is the cross-process arbiter: two tabs or Gateway
+        workers can both observe NULL, but only one can change it. The loser
+        reads and returns the winner's server value.
+        """
+        candidate = deepcopy(settings)
+        async with self._sf() as session:
+            result = await session.execute(
+                update(UserRow)
+                .where(UserRow.id == user_id, UserRow.preferences.is_(None))
+                .values(
+                    preferences=candidate,
+                    preferences_revision=UserRow.preferences_revision + 1,
+                )
+            )
+            if result.rowcount == 0:
+                row = (await session.execute(select(UserRow.preferences, UserRow.preferences_revision).where(UserRow.id == user_id))).one_or_none()
+                if row is None:
+                    raise UserNotFoundError(f"User {user_id} no longer exists")
+                existing, revision = row
+                if existing is None:
+                    # A concurrent transaction may have lost before committing;
+                    # let a normal retry from the client resolve that rare race.
+                    raise UserPreferencesWriteConflict(f"Preferences for user {user_id} are being initialized")
+                return deepcopy(existing), int(revision)
+
+            row = (await session.execute(select(UserRow.preferences, UserRow.preferences_revision).where(UserRow.id == user_id))).one()
+            await session.commit()
+            stored, revision = row
+            return deepcopy(stored), int(revision)
+
+    async def merge_user_preferences(self, user_id: str, patch: dict) -> tuple[dict | None, int]:
+        """Merge with an optimistic revision CAS supported by SQLite/Postgres.
+
+        A structurally corrupt JSON value cannot be deep-merged safely. Clear
+        it in the same CAS write so the client receives the normal absent-record
+        recovery signal instead of a server error.
+        """
+        for _attempt in range(_PREFERENCE_WRITE_MAX_ATTEMPTS):
+            async with self._sf() as session:
+                dialect_name = session.get_bind().dialect.name
+                try:
+                    row = (await session.execute(select(UserRow.preferences, UserRow.preferences_revision).where(UserRow.id == user_id))).one_or_none()
+                    if row is None:
+                        raise UserNotFoundError(f"User {user_id} no longer exists")
+                    current, revision = row
+                    if current is None:
+                        raise UserPreferencesNotInitializedError(f"Preferences for user {user_id} have not been initialized")
+
+                    merged = _merge_preferences(current, patch)
+                    result = await session.execute(
+                        update(UserRow)
+                        .where(
+                            UserRow.id == user_id,
+                            UserRow.preferences_revision == revision,
+                        )
+                        .values(
+                            preferences=merged,
+                            preferences_revision=revision + 1,
+                        )
+                    )
+                    if result.rowcount == 1:
+                        await session.commit()
+                        return merged, int(revision) + 1
+                    await session.rollback()
+                except OperationalError as exc:
+                    # In WAL mode, a transaction that SELECTed before another
+                    # writer committed cannot upgrade its stale read snapshot.
+                    # SQLite reports SQLITE_BUSY_SNAPSHOT immediately; the
+                    # connection busy_timeout cannot make that snapshot valid.
+                    # Roll back the snapshot and rerun the complete read/merge/
+                    # CAS cycle so disjoint client patches are not lost. Other
+                    # SQLite failures and every Postgres failure retain their
+                    # existing error semantics.
+                    if dialect_name != "sqlite" or not _is_sqlite_busy_error(exc):
+                        raise
+                    await session.rollback()
+
+        raise UserPreferencesWriteConflict(f"Concurrent preference updates for user {user_id} did not settle")
+
+    async def reset_user_preferences_if_revision(self, user_id: str, revision: int) -> tuple[dict | None, int]:
+        """Clear a corrupt record without erasing a concurrent valid update."""
+        async with self._sf() as session:
+            result = await session.execute(
+                update(UserRow)
+                .where(
+                    UserRow.id == user_id,
+                    UserRow.preferences_revision == revision,
+                )
+                .values(
+                    preferences=None,
+                    preferences_revision=revision + 1,
+                )
+            )
+            if result.rowcount == 1:
+                await session.commit()
+                return None, revision + 1
+
+            row = (await session.execute(select(UserRow.preferences, UserRow.preferences_revision).where(UserRow.id == user_id))).one_or_none()
+            if row is None:
+                raise UserNotFoundError(f"User {user_id} no longer exists")
+            settings, current_revision = row
+            return deepcopy(settings), int(current_revision)
+
+
+def _is_sqlite_busy_error(exc: OperationalError) -> bool:
+    """Match SQLITE_BUSY and its extended result codes from sqlite3."""
+    error_code = getattr(exc.orig, "sqlite_errorcode", None)
+    if isinstance(error_code, int):
+        return error_code & 0xFF == sqlite3.SQLITE_BUSY
+
+    # Python 3.12's sqlite3 exceptions expose ``sqlite_errorcode``. Keep the
+    # message fallback for compatible DBAPI adapters that omit that attribute.
+    return "database is locked" in str(exc.orig).lower()
+
+
+def _merge_preferences(current: object, patch: dict) -> dict | None:
+    """Deep-merge allowlisted sections, or flag an unsafe stored shape."""
+    if not isinstance(current, dict):
+        return None
+
+    merged = deepcopy(current)
+    for section, values in patch.items():
+        if not isinstance(values, dict):
+            return None
+        if section not in merged:
+            merged[section] = {}
+        target = merged[section]
+        if not isinstance(target, dict):
+            return None
+        for key, value in values.items():
+            if value is None:
+                target.pop(key, None)
+            else:
+                target[key] = deepcopy(value)
+    return merged

backend/app/gateway/deps.py

@@ -752,6 +752,13 @@ def get_local_provider() -> LocalAuthProvider:
     return _cached_local_provider
 
 
+def get_user_repository() -> SQLiteUserRepository:
+    """Return the shared SQL-backed user repository after engine startup."""
+    get_local_provider()
+    assert _cached_repo is not None
+    return _cached_repo
+
+
 async def get_current_user_from_request(request: Request):
     """Get the current authenticated user from the request cookie.