.gitignore

@@ -15,3 +15,4 @@ planning/
 frontend/node_modules/
 frontend/dist/
 frontend/.vite/
+AGENTS.MD

AGENTS.MD

@@ -31,7 +31,7 @@ This file is the consolidated source of truth for project scope, constraints, sc
   - CSRF required for unsafe methods via `X-CSRF-Token`.
 - Logging:
   - structured request logging with request IDs.
-  - redaction controls for sensitive fields.
+  - redaction controls for sensitive fields. 
 
 ## 4. Scraping Contract (Probe-Based)
 

alembic/versions/20260219_0010_author_search_shared_state.py

@@ -0,0 +1,115 @@
+"""Add shared author-search runtime and cache tables.
+
+Revision ID: 20260219_0010
+Revises: 20260219_0009
+Create Date: 2026-02-19 22:20:00.000000
+"""
+
+from collections.abc import Sequence
+
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql
+
+
+# revision identifiers, used by Alembic.
+revision: str = "20260219_0010"
+down_revision: str | Sequence[str] | None = "20260219_0009"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    bind = op.get_bind()
+    inspector = sa.inspect(bind)
+    table_names = set(inspector.get_table_names())
+
+    if "author_search_runtime_state" not in table_names:
+        op.create_table(
+            "author_search_runtime_state",
+            sa.Column("state_key", sa.String(length=64), nullable=False),
+            sa.Column("last_live_request_at", sa.DateTime(timezone=True), nullable=True),
+            sa.Column("cooldown_until", sa.DateTime(timezone=True), nullable=True),
+            sa.Column(
+                "consecutive_blocked_count",
+                sa.Integer(),
+                nullable=False,
+                server_default=sa.text("0"),
+            ),
+            sa.Column(
+                "cooldown_rejection_count",
+                sa.Integer(),
+                nullable=False,
+                server_default=sa.text("0"),
+            ),
+            sa.Column(
+                "cooldown_alert_emitted",
+                sa.Boolean(),
+                nullable=False,
+                server_default=sa.text("false"),
+            ),
+            sa.Column(
+                "created_at",
+                sa.DateTime(timezone=True),
+                nullable=False,
+                server_default=sa.func.now(),
+            ),
+            sa.Column(
+                "updated_at",
+                sa.DateTime(timezone=True),
+                nullable=False,
+                server_default=sa.func.now(),
+            ),
+            sa.PrimaryKeyConstraint("state_key", name=op.f("pk_author_search_runtime_state")),
+        )
+
+    if "author_search_cache_entries" not in table_names:
+        op.create_table(
+            "author_search_cache_entries",
+            sa.Column("query_key", sa.String(length=256), nullable=False),
+            sa.Column(
+                "payload",
+                postgresql.JSONB(astext_type=sa.Text()),
+                nullable=False,
+            ),
+            sa.Column("expires_at", sa.DateTime(timezone=True), nullable=False),
+            sa.Column(
+                "cached_at",
+                sa.DateTime(timezone=True),
+                nullable=False,
+                server_default=sa.func.now(),
+            ),
+            sa.Column(
+                "updated_at",
+                sa.DateTime(timezone=True),
+                nullable=False,
+                server_default=sa.func.now(),
+            ),
+            sa.PrimaryKeyConstraint("query_key", name=op.f("pk_author_search_cache_entries")),
+        )
+        op.create_index(
+            "ix_author_search_cache_expires_at",
+            "author_search_cache_entries",
+            ["expires_at"],
+            unique=False,
+        )
+        op.create_index(
+            "ix_author_search_cache_cached_at",
+            "author_search_cache_entries",
+            ["cached_at"],
+            unique=False,
+        )
+
+
+def downgrade() -> None:
+    bind = op.get_bind()
+    inspector = sa.inspect(bind)
+    table_names = set(inspector.get_table_names())
+
+    if "author_search_cache_entries" in table_names:
+        op.drop_index("ix_author_search_cache_cached_at", table_name="author_search_cache_entries")
+        op.drop_index("ix_author_search_cache_expires_at", table_name="author_search_cache_entries")
+        op.drop_table("author_search_cache_entries")
+
+    if "author_search_runtime_state" in table_names:
+        op.drop_table("author_search_runtime_state")

app/api/routers/scholars.py

@@ -149,12 +149,14 @@ async def search_scholars(
     request: Request,
     query: str = Query(..., min_length=2, max_length=120),
     limit: int = Query(10, ge=1, le=25),
+    db_session: AsyncSession = Depends(get_db_session),
     source: ScholarSource = Depends(get_scholar_source),
     current_user: User = Depends(get_api_current_user),
 ):
     try:
         parsed = await scholar_service.search_author_candidates(
             source=source,
+            db_session=db_session,
             query=query,
             limit=limit,
             network_error_retries=settings.ingestion_network_error_retries,

app/auth/rate_limit.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from collections import defaultdict, deque
+from collections import deque
 from collections.abc import Callable
 from dataclasses import dataclass
 from math import ceil
@@ -25,14 +25,19 @@ class SlidingWindowRateLimiter:
         self._max_attempts = max_attempts
         self._window_seconds = window_seconds
         self._now = now
-        self._attempts: dict[str, deque[float]] = defaultdict(deque)
+        self._attempts: dict[str, deque[float]] = {}
         self._lock = Lock()
 
     def check(self, key: str) -> RateLimitDecision:
         with self._lock:
             now_value = self._now()
-            attempts = self._attempts[key]
+            attempts = self._attempts.get(key)
+            if attempts is None:
+                return RateLimitDecision(allowed=True)
             self._trim_expired(attempts, now_value)
+            if not attempts:
+                self._attempts.pop(key, None)
+                return RateLimitDecision(allowed=True)
             if len(attempts) >= self._max_attempts:
                 retry_after = self._window_seconds - (now_value - attempts[0])
                 return RateLimitDecision(
@@ -44,7 +49,10 @@ class SlidingWindowRateLimiter:
     def record_failure(self, key: str) -> None:
         with self._lock:
             now_value = self._now()
-            attempts = self._attempts[key]
+            attempts = self._attempts.get(key)
+            if attempts is None:
+                attempts = deque()
+                self._attempts[key] = attempts
             self._trim_expired(attempts, now_value)
             attempts.append(now_value)
 
@@ -59,4 +67,3 @@ class SlidingWindowRateLimiter:
     def _trim_expired(self, attempts: deque[float], now_value: float) -> None:
         while attempts and now_value - attempts[0] >= self._window_seconds:
             attempts.popleft()
-

app/db/models.py

@@ -319,3 +319,56 @@ class IngestionQueueItem(Base):
     updated_at: Mapped[datetime] = mapped_column(
         DateTime(timezone=True), nullable=False, server_default=func.now()
     )
+
+
+class AuthorSearchRuntimeState(Base):
+    __tablename__ = "author_search_runtime_state"
+
+    state_key: Mapped[str] = mapped_column(String(64), primary_key=True)
+    last_live_request_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+    cooldown_until: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+    consecutive_blocked_count: Mapped[int] = mapped_column(
+        Integer,
+        nullable=False,
+        server_default=text("0"),
+    )
+    cooldown_rejection_count: Mapped[int] = mapped_column(
+        Integer,
+        nullable=False,
+        server_default=text("0"),
+    )
+    cooldown_alert_emitted: Mapped[bool] = mapped_column(
+        Boolean,
+        nullable=False,
+        server_default=text("false"),
+    )
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False, server_default=func.now()
+    )
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False, server_default=func.now()
+    )
+
+
+class AuthorSearchCacheEntry(Base):
+    __tablename__ = "author_search_cache_entries"
+    __table_args__ = (
+        Index("ix_author_search_cache_expires_at", "expires_at"),
+        Index("ix_author_search_cache_cached_at", "cached_at"),
+    )
+
+    query_key: Mapped[str] = mapped_column(String(256), primary_key=True)
+    payload: Mapped[dict] = mapped_column(
+        JSONB,
+        nullable=False,
+    )
+    expires_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True),
+        nullable=False,
+    )
+    cached_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False, server_default=func.now()
+    )
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False, server_default=func.now()
+    )

app/db/session.py

@@ -1,5 +1,6 @@
 from collections.abc import AsyncIterator
 import logging
+import os
 
 from sqlalchemy import text
 from sqlalchemy.ext.asyncio import (
@@ -18,16 +19,50 @@ _engine: AsyncEngine | None = None
 _session_factory: async_sessionmaker[AsyncSession] | None = None
 
 
+def _normalized_pool_mode(raw_mode: str) -> str:
+    mode = (raw_mode or "").strip().lower()
+    if mode == "auto":
+        if os.getenv("PYTEST_CURRENT_TEST"):
+            return "null"
+        app_env = (os.getenv("APP_ENV") or "").strip().lower()
+        if app_env in {"test", "development", "dev", "local"}:
+            return "null"
+        return "queue"
+    if mode in {"null", "queue"}:
+        return mode
+    logger.warning(
+        "db.invalid_pool_mode_fallback",
+        extra={
+            "event": "db.invalid_pool_mode_fallback",
+            "database_pool_mode": raw_mode,
+            "fallback_mode": "queue",
+        },
+    )
+    return "queue"
+
+
 def get_engine() -> AsyncEngine:
     global _engine
     if _engine is None:
-        # NullPool avoids cross-event-loop connection reuse in tests and dev tools.
+        pool_mode = _normalized_pool_mode(settings.database_pool_mode)
-        _engine = create_async_engine(
+        engine_kwargs: dict[str, object] = {
-            settings.database_url,
+            "pool_pre_ping": True,
-            pool_pre_ping=True,
+        }
-            poolclass=NullPool,
+        if pool_mode == "null":
+            engine_kwargs["poolclass"] = NullPool
+        else:
+            engine_kwargs["pool_size"] = max(1, int(settings.database_pool_size))
+            engine_kwargs["max_overflow"] = max(0, int(settings.database_pool_max_overflow))
+            engine_kwargs["pool_timeout"] = max(1, int(settings.database_pool_timeout_seconds))
+
+        _engine = create_async_engine(settings.database_url, **engine_kwargs)
+        logger.info(
+            "db.engine_initialized",
+            extra={
+                "event": "db.engine_initialized",
+                "pool_mode": pool_mode,
+            },
         )
-        logger.info("db.engine_initialized", extra={"event": "db.engine_initialized"})
     return _engine
 
 

app/services/scholar_source.py

@@ -195,7 +195,7 @@ def _build_profile_url(*, scholar_id: str, cstart: int, pagesize: int) -> str:
     query: dict[str, int | str] = {"hl": "en", "user": scholar_id}
     if cstart > 0:
         query["cstart"] = int(cstart)
-    if pagesize > 0 and cstart > 0:
+    if pagesize > 0:
         query["pagesize"] = int(pagesize)
     return f"{SCHOLAR_PROFILE_URL}?{urlencode(query)}"
 

app/services/scholars.py

@@ -1,8 +1,6 @@
 from __future__ import annotations
 
 import asyncio
-from collections import OrderedDict
-from dataclasses import dataclass
 from dataclasses import replace
 from datetime import datetime
 from datetime import timedelta
@@ -11,16 +9,15 @@ import logging
 import os
 import random
 import re
-import time
 from pathlib import Path
 from urllib.parse import urlparse
 from uuid import uuid4
 
-from sqlalchemy import select
+from sqlalchemy import delete, func, select, text
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from app.db.models import ScholarProfile
+from app.db.models import AuthorSearchCacheEntry, AuthorSearchRuntimeState, ScholarProfile
 from app.services.scholar_parser import (
     ParseState,
     ParsedAuthorSearchPage,
@@ -40,6 +37,9 @@ DEFAULT_AUTHOR_SEARCH_MIN_INTERVAL_SECONDS = 3.0
 DEFAULT_AUTHOR_SEARCH_INTERVAL_JITTER_SECONDS = 1.0
 DEFAULT_AUTHOR_SEARCH_RETRY_ALERT_THRESHOLD = 2
 DEFAULT_AUTHOR_SEARCH_COOLDOWN_REJECTION_ALERT_THRESHOLD = 3
+AUTHOR_SEARCH_RUNTIME_STATE_KEY = "global"
+AUTHOR_SEARCH_LOCK_NAMESPACE = 3901
+AUTHOR_SEARCH_LOCK_KEY = 1
 ALLOWED_IMAGE_UPLOAD_CONTENT_TYPES = {
     "image/jpeg": ".jpg",
     "image/png": ".png",
@@ -91,22 +91,6 @@ STATE_REASON_HINTS: dict[str, str] = {
 logger = logging.getLogger(__name__)
 
 
-@dataclass(frozen=True)
-class _AuthorSearchCacheEntry:
-    parsed: ParsedAuthorSearchPage
-    expires_at_monotonic: float
-    cached_at_utc: datetime
-
-
-_AUTHOR_SEARCH_EXECUTION_LOCK = asyncio.Lock()
-_AUTHOR_SEARCH_CACHE: OrderedDict[str, _AuthorSearchCacheEntry] = OrderedDict()
-_AUTHOR_SEARCH_LAST_LIVE_REQUEST_MONOTONIC = 0.0
-_AUTHOR_SEARCH_COOLDOWN_UNTIL_UTC: datetime | None = None
-_AUTHOR_SEARCH_CONSECUTIVE_BLOCKED_COUNT = 0
-_AUTHOR_SEARCH_COOLDOWN_REJECTION_COUNT = 0
-_AUTHOR_SEARCH_COOLDOWN_ALERT_EMITTED = False
-
-
 class ScholarServiceError(ValueError):
     """Raised for expected scholar-management validation failures."""
 
@@ -235,64 +219,260 @@ def _policy_blocked_author_search_result(
     )
 
 
-def _cache_get_author_search_result(query_key: str, now_monotonic: float) -> _AuthorSearchCacheEntry | None:
+async def _acquire_author_search_lock(db_session: AsyncSession) -> None:
-    entry = _AUTHOR_SEARCH_CACHE.get(query_key)
+    await db_session.execute(
+        text("SELECT pg_advisory_xact_lock(:namespace, :lock_key)"),
+        {
+            "namespace": AUTHOR_SEARCH_LOCK_NAMESPACE,
+            "lock_key": AUTHOR_SEARCH_LOCK_KEY,
+        },
+    )
+
+
+async def _load_runtime_state_for_update(
+    db_session: AsyncSession,
+) -> AuthorSearchRuntimeState:
+    result = await db_session.execute(
+        select(AuthorSearchRuntimeState)
+        .where(AuthorSearchRuntimeState.state_key == AUTHOR_SEARCH_RUNTIME_STATE_KEY)
+        .with_for_update()
+    )
+    state = result.scalar_one_or_none()
+    if state is not None:
+        return state
+
+    state = AuthorSearchRuntimeState(state_key=AUTHOR_SEARCH_RUNTIME_STATE_KEY)
+    db_session.add(state)
+    await db_session.flush()
+    return state
+
+
+def _serialize_parsed_author_search_page(parsed: ParsedAuthorSearchPage) -> dict:
+    return {
+        "state": parsed.state.value,
+        "state_reason": parsed.state_reason,
+        "marker_counts": {str(key): int(value) for key, value in parsed.marker_counts.items()},
+        "warnings": [str(value) for value in parsed.warnings],
+        "candidates": [
+            {
+                "scholar_id": candidate.scholar_id,
+                "display_name": candidate.display_name,
+                "affiliation": candidate.affiliation,
+                "email_domain": candidate.email_domain,
+                "cited_by_count": candidate.cited_by_count,
+                "interests": [str(interest) for interest in candidate.interests],
+                "profile_url": candidate.profile_url,
+                "profile_image_url": candidate.profile_image_url,
+            }
+            for candidate in parsed.candidates
+        ],
+    }
+
+
+def _deserialize_parsed_author_search_page(payload: object) -> ParsedAuthorSearchPage | None:
+    if not isinstance(payload, dict):
+        return None
+
+    state_raw = str(payload.get("state", "")).strip()
+    try:
+        state = ParseState(state_raw)
+    except ValueError:
+        return None
+
+    marker_counts_payload = payload.get("marker_counts")
+    marker_counts = (
+        {str(key): int(value) for key, value in marker_counts_payload.items()}
+        if isinstance(marker_counts_payload, dict)
+        else {}
+    )
+    warnings_payload = payload.get("warnings")
+    warnings = (
+        [str(value) for value in warnings_payload if isinstance(value, str)]
+        if isinstance(warnings_payload, list)
+        else []
+    )
+    from app.services.scholar_parser import ScholarSearchCandidate
+
+    candidates_payload = payload.get("candidates")
+    normalized_candidates = []
+    for value in candidates_payload if isinstance(candidates_payload, list) else []:
+        if not isinstance(value, dict):
+            continue
+        scholar_id = str(value.get("scholar_id", "")).strip()
+        display_name = str(value.get("display_name", "")).strip()
+        profile_url = str(value.get("profile_url", "")).strip()
+        if not scholar_id or not display_name or not profile_url:
+            continue
+        interests_payload = value.get("interests")
+        interests = (
+            [str(item) for item in interests_payload if isinstance(item, str)]
+            if isinstance(interests_payload, list)
+            else []
+        )
+        cited_by_count = value.get("cited_by_count")
+        parsed_cited_by_count: int | None = None
+        if isinstance(cited_by_count, int):
+            parsed_cited_by_count = cited_by_count
+        elif isinstance(cited_by_count, str) and cited_by_count.strip():
+            try:
+                parsed_cited_by_count = int(cited_by_count)
+            except ValueError:
+                parsed_cited_by_count = None
+        normalized_candidates.append(
+            {
+                "scholar_id": scholar_id,
+                "display_name": display_name,
+                "affiliation": str(value.get("affiliation")).strip()
+                if value.get("affiliation") is not None
+                else None,
+                "email_domain": str(value.get("email_domain")).strip()
+                if value.get("email_domain") is not None
+                else None,
+                "cited_by_count": parsed_cited_by_count,
+                "interests": interests,
+                "profile_url": profile_url,
+                "profile_image_url": str(value.get("profile_image_url")).strip()
+                if value.get("profile_image_url") is not None
+                else None,
+            }
+        )
+
+    return ParsedAuthorSearchPage(
+        state=state,
+        state_reason=str(payload.get("state_reason", "")).strip() or "unknown",
+        candidates=[
+            ScholarSearchCandidate(
+                scholar_id=item["scholar_id"],
+                display_name=item["display_name"],
+                affiliation=item["affiliation"],
+                email_domain=item["email_domain"],
+                cited_by_count=item["cited_by_count"],
+                interests=item["interests"],
+                profile_url=item["profile_url"],
+                profile_image_url=item["profile_image_url"],
+            )
+            for item in normalized_candidates
+        ],
+        marker_counts=marker_counts,
+        warnings=warnings,
+    )
+
+
+async def _cache_get_author_search_result(
+    db_session: AsyncSession,
+    *,
+    query_key: str,
+    now_utc: datetime,
+) -> ParsedAuthorSearchPage | None:
+    result = await db_session.execute(
+        select(AuthorSearchCacheEntry).where(AuthorSearchCacheEntry.query_key == query_key)
+    )
+    entry = result.scalar_one_or_none()
     if entry is None:
         return None
-    if entry.expires_at_monotonic <= now_monotonic:
+    expires_at = entry.expires_at
-        _AUTHOR_SEARCH_CACHE.pop(query_key, None)
+    if expires_at.tzinfo is None:
+        expires_at = expires_at.replace(tzinfo=timezone.utc)
+    if expires_at <= now_utc:
+        await db_session.delete(entry)
         return None
-    _AUTHOR_SEARCH_CACHE.move_to_end(query_key)
+    parsed = _deserialize_parsed_author_search_page(entry.payload)
-    return entry
+    if parsed is None:
+        await db_session.delete(entry)
+        return None
+    return parsed
 
 
-def _cache_set_author_search_result(
+async def _cache_set_author_search_result(
+    db_session: AsyncSession,
     *,
     query_key: str,
     parsed: ParsedAuthorSearchPage,
     ttl_seconds: float,
     max_entries: int,
+    now_utc: datetime,
 ) -> None:
     ttl = max(float(ttl_seconds), 0.0)
+    existing_result = await db_session.execute(
+        select(AuthorSearchCacheEntry).where(AuthorSearchCacheEntry.query_key == query_key)
+    )
+    existing = existing_result.scalar_one_or_none()
+
     if ttl <= 0.0:
-        _AUTHOR_SEARCH_CACHE.pop(query_key, None)
+        if existing is not None:
+            await db_session.delete(existing)
         return
 
-    _AUTHOR_SEARCH_CACHE[query_key] = _AuthorSearchCacheEntry(
+    expires_at = now_utc + timedelta(seconds=ttl)
-        parsed=parsed,
+    payload = _serialize_parsed_author_search_page(parsed)
-        expires_at_monotonic=time.monotonic() + ttl,
+    if existing is None:
-        cached_at_utc=datetime.now(timezone.utc),
+        db_session.add(
-    )
+            AuthorSearchCacheEntry(
-    _AUTHOR_SEARCH_CACHE.move_to_end(query_key)
+                query_key=query_key,
+                payload=payload,
+                expires_at=expires_at,
+                cached_at=now_utc,
+                updated_at=now_utc,
+            )
+        )
+    else:
+        existing.payload = payload
+        existing.expires_at = expires_at
+        existing.cached_at = now_utc
+        existing.updated_at = now_utc
 
+    await _prune_author_search_cache(db_session, now_utc=now_utc, max_entries=max_entries)
+
+
+async def _prune_author_search_cache(
+    db_session: AsyncSession,
+    *,
+    now_utc: datetime,
+    max_entries: int,
+) -> None:
+    await db_session.execute(
+        delete(AuthorSearchCacheEntry).where(AuthorSearchCacheEntry.expires_at <= now_utc)
+    )
     bounded_max_entries = max(1, int(max_entries))
-    while len(_AUTHOR_SEARCH_CACHE) > bounded_max_entries:
+    count_result = await db_session.execute(
-        _AUTHOR_SEARCH_CACHE.popitem(last=False)
+        select(func.count()).select_from(AuthorSearchCacheEntry)
+    )
+    entry_count = int(count_result.scalar_one() or 0)
+    overflow = max(0, entry_count - bounded_max_entries)
+    if overflow <= 0:
+        return
+    stale_keys_result = await db_session.execute(
+        select(AuthorSearchCacheEntry.query_key)
+        .order_by(AuthorSearchCacheEntry.cached_at.asc())
+        .limit(overflow)
+    )
+    stale_keys = [str(row[0]) for row in stale_keys_result.all()]
+    if stale_keys:
+        await db_session.execute(
+            delete(AuthorSearchCacheEntry).where(AuthorSearchCacheEntry.query_key.in_(stale_keys))
+        )
 
 
 def _is_author_search_block_state(parsed: ParsedAuthorSearchPage) -> bool:
     return parsed.state == ParseState.BLOCKED_OR_CAPTCHA
 
 
-def _author_search_cooldown_remaining_seconds(now_utc: datetime) -> int:
+def _author_search_cooldown_remaining_seconds(
-    if _AUTHOR_SEARCH_COOLDOWN_UNTIL_UTC is None:
+    runtime_state: AuthorSearchRuntimeState,
+    now_utc: datetime,
+) -> int:
+    cooldown_until = runtime_state.cooldown_until
+    if cooldown_until is None:
         return 0
-    remaining_seconds = int((_AUTHOR_SEARCH_COOLDOWN_UNTIL_UTC - now_utc).total_seconds())
+    if cooldown_until.tzinfo is None:
+        cooldown_until = cooldown_until.replace(tzinfo=timezone.utc)
+    remaining_seconds = int((cooldown_until - now_utc).total_seconds())
     return max(0, remaining_seconds)
 
 
 def _reset_author_search_runtime_state_for_tests() -> None:
-    global _AUTHOR_SEARCH_LAST_LIVE_REQUEST_MONOTONIC
+    # Runtime state now lives in the database; tests should reset via DB fixtures.
-    global _AUTHOR_SEARCH_COOLDOWN_UNTIL_UTC
+    return None
-    global _AUTHOR_SEARCH_CONSECUTIVE_BLOCKED_COUNT
-    global _AUTHOR_SEARCH_COOLDOWN_REJECTION_COUNT
-    global _AUTHOR_SEARCH_COOLDOWN_ALERT_EMITTED
-    _AUTHOR_SEARCH_CACHE.clear()
-    _AUTHOR_SEARCH_LAST_LIVE_REQUEST_MONOTONIC = 0.0
-    _AUTHOR_SEARCH_COOLDOWN_UNTIL_UTC = None
-    _AUTHOR_SEARCH_CONSECUTIVE_BLOCKED_COUNT = 0
-    _AUTHOR_SEARCH_COOLDOWN_REJECTION_COUNT = 0
-    _AUTHOR_SEARCH_COOLDOWN_ALERT_EMITTED = False
 
 
 async def list_scholars_for_user(
@@ -375,6 +555,7 @@ async def delete_scholar(
 async def search_author_candidates(
     *,
     source: ScholarSource,
+    db_session: AsyncSession,
     query: str,
     limit: int,
     network_error_retries: int = 1,
@@ -390,12 +571,6 @@ async def search_author_candidates(
     retry_alert_threshold: int = DEFAULT_AUTHOR_SEARCH_RETRY_ALERT_THRESHOLD,
     cooldown_rejection_alert_threshold: int = DEFAULT_AUTHOR_SEARCH_COOLDOWN_REJECTION_ALERT_THRESHOLD,
 ) -> ParsedAuthorSearchPage:
-    global _AUTHOR_SEARCH_LAST_LIVE_REQUEST_MONOTONIC
-    global _AUTHOR_SEARCH_COOLDOWN_UNTIL_UTC
-    global _AUTHOR_SEARCH_CONSECUTIVE_BLOCKED_COUNT
-    global _AUTHOR_SEARCH_COOLDOWN_REJECTION_COUNT
-    global _AUTHOR_SEARCH_COOLDOWN_ALERT_EMITTED
-
     normalized_query = query.strip()
     if len(normalized_query) < 2:
         raise ScholarServiceError("Search query must be at least 2 characters.")
@@ -416,208 +591,234 @@ async def search_author_candidates(
             limit=bounded_limit,
         )
 
-    async with _AUTHOR_SEARCH_EXECUTION_LOCK:
+    await _acquire_author_search_lock(db_session)
-        now_utc = datetime.now(timezone.utc)
+    runtime_state = await _load_runtime_state_for_update(db_session)
-        now_monotonic = time.monotonic()
+    runtime_state_updated = False
+    now_utc = datetime.now(timezone.utc)
 
-        if (
+    if runtime_state.cooldown_until is not None:
-            _AUTHOR_SEARCH_COOLDOWN_UNTIL_UTC is not None
+        cooldown_until = runtime_state.cooldown_until
-            and now_utc >= _AUTHOR_SEARCH_COOLDOWN_UNTIL_UTC
+        if cooldown_until.tzinfo is None:
-        ):
+            cooldown_until = cooldown_until.replace(tzinfo=timezone.utc)
+            runtime_state.cooldown_until = cooldown_until
+            runtime_state_updated = True
+        if now_utc >= cooldown_until:
             logger.info(
                 "scholar_search.cooldown_expired",
                 extra={
                     "event": "scholar_search.cooldown_expired",
-                    "cooldown_until_utc": _AUTHOR_SEARCH_COOLDOWN_UNTIL_UTC.isoformat(),
+                    "cooldown_until_utc": cooldown_until.isoformat(),
                 },
             )
-            _AUTHOR_SEARCH_COOLDOWN_UNTIL_UTC = None
+            runtime_state.cooldown_until = None
-            _AUTHOR_SEARCH_COOLDOWN_REJECTION_COUNT = 0
+            runtime_state.cooldown_rejection_count = 0
-            _AUTHOR_SEARCH_COOLDOWN_ALERT_EMITTED = False
+            runtime_state.cooldown_alert_emitted = False
+            runtime_state_updated = True
 
-        cooldown_remaining_seconds = _author_search_cooldown_remaining_seconds(now_utc)
+    cooldown_remaining_seconds = _author_search_cooldown_remaining_seconds(runtime_state, now_utc)
-        if cooldown_remaining_seconds > 0:
+    if cooldown_remaining_seconds > 0:
-            _AUTHOR_SEARCH_COOLDOWN_REJECTION_COUNT += 1
+        runtime_state.cooldown_rejection_count = int(runtime_state.cooldown_rejection_count) + 1
-            bounded_cooldown_rejection_alert_threshold = max(
+        bounded_cooldown_rejection_alert_threshold = max(
-                1,
+            1,
-                int(cooldown_rejection_alert_threshold),
+            int(cooldown_rejection_alert_threshold),
-            )
+        )
-            if (
+        if (
-                _AUTHOR_SEARCH_COOLDOWN_REJECTION_COUNT
+            int(runtime_state.cooldown_rejection_count) >= bounded_cooldown_rejection_alert_threshold
-                >= bounded_cooldown_rejection_alert_threshold
+            and not bool(runtime_state.cooldown_alert_emitted)
-                and not _AUTHOR_SEARCH_COOLDOWN_ALERT_EMITTED
+        ):
-            ):
+            logger.error(
-                logger.error(
+                "scholar_search.cooldown_rejection_threshold_exceeded",
-                    "scholar_search.cooldown_rejection_threshold_exceeded",
-                    extra={
-                        "event": "scholar_search.cooldown_rejection_threshold_exceeded",
-                        "query": normalized_query,
-                        "cooldown_rejection_count": _AUTHOR_SEARCH_COOLDOWN_REJECTION_COUNT,
-                        "threshold": bounded_cooldown_rejection_alert_threshold,
-                        "cooldown_until_utc": _AUTHOR_SEARCH_COOLDOWN_UNTIL_UTC.isoformat()
-                        if _AUTHOR_SEARCH_COOLDOWN_UNTIL_UTC
-                        else None,
-                    },
-                )
-                _AUTHOR_SEARCH_COOLDOWN_ALERT_EMITTED = True
-
-            logger.warning(
-                "scholar_search.cooldown_active",
                 extra={
-                    "event": "scholar_search.cooldown_active",
+                    "event": "scholar_search.cooldown_rejection_threshold_exceeded",
                     "query": normalized_query,
-                    "cooldown_remaining_seconds": cooldown_remaining_seconds,
+                    "cooldown_rejection_count": int(runtime_state.cooldown_rejection_count),
-                    "cooldown_until_utc": _AUTHOR_SEARCH_COOLDOWN_UNTIL_UTC.isoformat()
+                    "threshold": bounded_cooldown_rejection_alert_threshold,
-                    if _AUTHOR_SEARCH_COOLDOWN_UNTIL_UTC
+                    "cooldown_until_utc": runtime_state.cooldown_until.isoformat()
+                    if runtime_state.cooldown_until
                     else None,
                 },
             )
-            warning_codes = [
+            runtime_state.cooldown_alert_emitted = True
-                "author_search_cooldown_active",
+        runtime_state_updated = True
-                f"author_search_cooldown_remaining_{cooldown_remaining_seconds}s",
-            ]
-            if _AUTHOR_SEARCH_COOLDOWN_ALERT_EMITTED:
-                warning_codes.append("author_search_cooldown_alert_threshold_exceeded")
-            return _policy_blocked_author_search_result(
-                reason=SEARCH_COOLDOWN_REASON,
-                warning_codes=warning_codes,
-                limit=bounded_limit,
-            )
 
-        cached_entry = _cache_get_author_search_result(query_key, now_monotonic)
+        logger.warning(
-        if cached_entry is not None:
+            "scholar_search.cooldown_active",
-            cached = cached_entry.parsed
+            extra={
-            state_reason_override = (
+                "event": "scholar_search.cooldown_active",
-                SEARCH_CACHED_BLOCK_REASON if _is_author_search_block_state(cached) else None
+                "query": normalized_query,
-            )
+                "cooldown_remaining_seconds": cooldown_remaining_seconds,
-            logger.info(
+                "cooldown_until_utc": runtime_state.cooldown_until.isoformat()
-                "scholar_search.cache_hit",
+                if runtime_state.cooldown_until
-                extra={
+                else None,
-                    "event": "scholar_search.cache_hit",
+            },
-                    "query": normalized_query,
+        )
-                    "state": cached.state.value,
+        warning_codes = [
-                    "state_reason": cached.state_reason,
+            "author_search_cooldown_active",
-                },
+            f"author_search_cooldown_remaining_{cooldown_remaining_seconds}s",
-            )
+        ]
-            return _trim_author_search_result(
+        if bool(runtime_state.cooldown_alert_emitted):
-                cached,
+            warning_codes.append("author_search_cooldown_alert_threshold_exceeded")
-                limit=bounded_limit,
+        if runtime_state_updated:
-                extra_warnings=["author_search_served_from_cache"],
+            await db_session.commit()
-                state_reason_override=state_reason_override,
+        return _policy_blocked_author_search_result(
-            )
+            reason=SEARCH_COOLDOWN_REASON,
+            warning_codes=warning_codes,
+            limit=bounded_limit,
+        )
 
+    cached = await _cache_get_author_search_result(
+        db_session,
+        query_key=query_key,
+        now_utc=now_utc,
+    )
+    if cached is not None:
+        state_reason_override = (
+            SEARCH_CACHED_BLOCK_REASON if _is_author_search_block_state(cached) else None
+        )
+        logger.info(
+            "scholar_search.cache_hit",
+            extra={
+                "event": "scholar_search.cache_hit",
+                "query": normalized_query,
+                "state": cached.state.value,
+                "state_reason": cached.state_reason,
+            },
+        )
+        await db_session.commit()
+        return _trim_author_search_result(
+            cached,
+            limit=bounded_limit,
+            extra_warnings=["author_search_served_from_cache"],
+            state_reason_override=state_reason_override,
+        )
+
+    if runtime_state.last_live_request_at is not None:
+        last_live_request_at = runtime_state.last_live_request_at
+        if last_live_request_at.tzinfo is None:
+            last_live_request_at = last_live_request_at.replace(tzinfo=timezone.utc)
+            runtime_state.last_live_request_at = last_live_request_at
+            runtime_state_updated = True
         enforced_wait_seconds = (
-            (_AUTHOR_SEARCH_LAST_LIVE_REQUEST_MONOTONIC + max(float(min_interval_seconds), 0.0))
+            last_live_request_at + timedelta(seconds=max(float(min_interval_seconds), 0.0)) - now_utc
-            - now_monotonic
+        ).total_seconds()
+    else:
+        enforced_wait_seconds = 0.0
+
+    jitter_seconds = random.uniform(0.0, max(float(interval_jitter_seconds), 0.0))
+    sleep_seconds = max(0.0, float(enforced_wait_seconds)) + jitter_seconds
+    if sleep_seconds > 0.0:
+        logger.info(
+            "scholar_search.throttle_wait",
+            extra={
+                "event": "scholar_search.throttle_wait",
+                "query": normalized_query,
+                "sleep_seconds": round(sleep_seconds, 3),
+            },
         )
-        jitter_seconds = random.uniform(0.0, max(float(interval_jitter_seconds), 0.0))
+        await asyncio.sleep(sleep_seconds)
-        sleep_seconds = max(0.0, enforced_wait_seconds) + jitter_seconds
-        if sleep_seconds > 0.0:
-            logger.info(
-                "scholar_search.throttle_wait",
-                extra={
-                    "event": "scholar_search.throttle_wait",
-                    "query": normalized_query,
-                    "sleep_seconds": round(sleep_seconds, 3),
-                },
-            )
-            await asyncio.sleep(sleep_seconds)
 
-        max_attempts = max(1, int(network_error_retries) + 1)
+    max_attempts = max(1, int(network_error_retries) + 1)
-        parsed: ParsedAuthorSearchPage | None = None
+    parsed: ParsedAuthorSearchPage | None = None
-        retry_warnings: list[str] = []
+    retry_warnings: list[str] = []
-        retry_scheduled_count = 0
+    retry_scheduled_count = 0
 
-        for attempt_index in range(max_attempts):
+    for attempt_index in range(max_attempts):
-            fetch_result = await source.fetch_author_search_html(normalized_query, start=0)
+        fetch_result = await source.fetch_author_search_html(normalized_query, start=0)
-            parsed = parse_author_search_page(fetch_result)
+        parsed = parse_author_search_page(fetch_result)
-            if parsed.state != ParseState.NETWORK_ERROR or attempt_index >= max_attempts - 1:
+        if parsed.state != ParseState.NETWORK_ERROR or attempt_index >= max_attempts - 1:
-                break
+            break
 
-            retry_warnings.append("network_retry_scheduled_for_author_search")
+        retry_warnings.append("network_retry_scheduled_for_author_search")
-            retry_scheduled_count += 1
+        retry_scheduled_count += 1
-            sleep_seconds = max(float(retry_backoff_seconds), 0.0) * (2**attempt_index)
+        retry_sleep_seconds = max(float(retry_backoff_seconds), 0.0) * (2**attempt_index)
-            if sleep_seconds > 0:
+        if retry_sleep_seconds > 0:
-                await asyncio.sleep(sleep_seconds)
+            await asyncio.sleep(retry_sleep_seconds)
 
-        _AUTHOR_SEARCH_LAST_LIVE_REQUEST_MONOTONIC = time.monotonic()
+    runtime_state.last_live_request_at = datetime.now(timezone.utc)
+    runtime_state_updated = True
 
-        if parsed is None:
+    if parsed is None:
-            raise ScholarServiceError("Unable to complete scholar author search.")
+        raise ScholarServiceError("Unable to complete scholar author search.")
 
+    merged_parsed = replace(
+        parsed,
+        warnings=_merge_warnings(parsed.warnings, retry_warnings),
+    )
+
+    bounded_retry_alert_threshold = max(1, int(retry_alert_threshold))
+    if retry_scheduled_count >= bounded_retry_alert_threshold:
+        logger.warning(
+            "scholar_search.retry_threshold_exceeded",
+            extra={
+                "event": "scholar_search.retry_threshold_exceeded",
+                "query": normalized_query,
+                "retry_scheduled_count": retry_scheduled_count,
+                "threshold": bounded_retry_alert_threshold,
+                "final_state": merged_parsed.state.value,
+                "final_state_reason": merged_parsed.state_reason,
+            },
+        )
         merged_parsed = replace(
-            parsed,
+            merged_parsed,
-            warnings=_merge_warnings(parsed.warnings, retry_warnings),
+            warnings=_merge_warnings(
+                merged_parsed.warnings,
+                [f"author_search_retry_threshold_exceeded_{retry_scheduled_count}"],
+            ),
         )
 
-        bounded_retry_alert_threshold = max(1, int(retry_alert_threshold))
+    if _is_author_search_block_state(merged_parsed):
-        if retry_scheduled_count >= bounded_retry_alert_threshold:
+        runtime_state.consecutive_blocked_count = int(runtime_state.consecutive_blocked_count) + 1
-            logger.warning(
+        logger.warning(
-                "scholar_search.retry_threshold_exceeded",
+            "scholar_search.block_detected",
-                extra={
+            extra={
-                    "event": "scholar_search.retry_threshold_exceeded",
+                "event": "scholar_search.block_detected",
-                    "query": normalized_query,
+                "query": normalized_query,
-                    "retry_scheduled_count": retry_scheduled_count,
+                "state_reason": merged_parsed.state_reason,
-                    "threshold": bounded_retry_alert_threshold,
+                "consecutive_blocked_count": int(runtime_state.consecutive_blocked_count),
-                    "final_state": merged_parsed.state.value,
+            },
-                    "final_state_reason": merged_parsed.state_reason,
+        )
-                },
+        if int(runtime_state.consecutive_blocked_count) >= max(1, int(cooldown_block_threshold)):
+            runtime_state.cooldown_until = datetime.now(timezone.utc) + timedelta(
+                seconds=max(60, int(cooldown_seconds))
             )
+            runtime_state.consecutive_blocked_count = 0
+            runtime_state.cooldown_rejection_count = 0
+            runtime_state.cooldown_alert_emitted = False
             merged_parsed = replace(
                 merged_parsed,
                 warnings=_merge_warnings(
                     merged_parsed.warnings,
-                    [f"author_search_retry_threshold_exceeded_{retry_scheduled_count}"],
+                    ["author_search_circuit_breaker_armed"],
                 ),
             )
-
+            logger.error(
-        if _is_author_search_block_state(merged_parsed):
+                "scholar_search.cooldown_activated",
-            _AUTHOR_SEARCH_CONSECUTIVE_BLOCKED_COUNT += 1
-            logger.warning(
-                "scholar_search.block_detected",
                 extra={
-                    "event": "scholar_search.block_detected",
+                    "event": "scholar_search.cooldown_activated",
                     "query": normalized_query,
-                    "state_reason": merged_parsed.state_reason,
+                    "cooldown_until_utc": runtime_state.cooldown_until.isoformat()
-                    "consecutive_blocked_count": _AUTHOR_SEARCH_CONSECUTIVE_BLOCKED_COUNT,
+                    if runtime_state.cooldown_until
+                    else None,
                 },
             )
-            if _AUTHOR_SEARCH_CONSECUTIVE_BLOCKED_COUNT >= max(1, int(cooldown_block_threshold)):
+    else:
-                _AUTHOR_SEARCH_COOLDOWN_UNTIL_UTC = datetime.now(timezone.utc) + timedelta(
+        runtime_state.consecutive_blocked_count = 0
-                    seconds=max(60, int(cooldown_seconds))
-                )
-                _AUTHOR_SEARCH_CONSECUTIVE_BLOCKED_COUNT = 0
-                _AUTHOR_SEARCH_COOLDOWN_REJECTION_COUNT = 0
-                _AUTHOR_SEARCH_COOLDOWN_ALERT_EMITTED = False
-                merged_parsed = replace(
-                    merged_parsed,
-                    warnings=_merge_warnings(
-                        merged_parsed.warnings,
-                        ["author_search_circuit_breaker_armed"],
-                    ),
-                )
-                logger.error(
-                    "scholar_search.cooldown_activated",
-                    extra={
-                        "event": "scholar_search.cooldown_activated",
-                        "query": normalized_query,
-                        "cooldown_until_utc": _AUTHOR_SEARCH_COOLDOWN_UNTIL_UTC.isoformat(),
-                    },
-                )
-        else:
-            _AUTHOR_SEARCH_CONSECUTIVE_BLOCKED_COUNT = 0
 
-        ttl_seconds = (
+    ttl_seconds = (
-            min(max(1, int(blocked_cache_ttl_seconds)), max(1, int(cache_ttl_seconds)))
+        min(max(1, int(blocked_cache_ttl_seconds)), max(1, int(cache_ttl_seconds)))
-            if _is_author_search_block_state(merged_parsed)
+        if _is_author_search_block_state(merged_parsed)
-            else max(1, int(cache_ttl_seconds))
+        else max(1, int(cache_ttl_seconds))
-        )
+    )
-        _cache_set_author_search_result(
+    await _cache_set_author_search_result(
-            query_key=query_key,
+        db_session,
-            parsed=merged_parsed,
+        query_key=query_key,
-            ttl_seconds=float(ttl_seconds),
+        parsed=merged_parsed,
-            max_entries=cache_max_entries,
+        ttl_seconds=float(ttl_seconds),
-        )
+        max_entries=cache_max_entries,
+        now_utc=datetime.now(timezone.utc),
+    )
+    if runtime_state_updated:
+        await db_session.commit()
 
     return _trim_author_search_result(
         merged_parsed,

app/settings.py

@@ -64,6 +64,10 @@ class Settings:
         "DATABASE_URL",
         "postgresql+asyncpg://scholar:scholar@db:5432/scholar",
     )
+    database_pool_mode: str = _env_str("DATABASE_POOL_MODE", "auto")
+    database_pool_size: int = _env_int("DATABASE_POOL_SIZE", 5)
+    database_pool_max_overflow: int = _env_int("DATABASE_POOL_MAX_OVERFLOW", 10)
+    database_pool_timeout_seconds: int = _env_int("DATABASE_POOL_TIMEOUT_SECONDS", 30)
     session_secret_key: str = os.getenv("SESSION_SECRET_KEY", "dev-insecure-session-key")
     session_cookie_secure: bool = _env_bool("SESSION_COOKIE_SECURE", False)
     security_headers_enabled: bool = _env_bool("SECURITY_HEADERS_ENABLED", True)

frontend/src/stores/user_settings.test.ts

@@ -0,0 +1,50 @@
+import { beforeEach, describe, expect, it } from "vitest";
+import { createPinia, setActivePinia } from "pinia";
+
+import { useUserSettingsStore } from "@/stores/user_settings";
+
+describe("user settings store", () => {
+  beforeEach(() => {
+    setActivePinia(createPinia());
+  });
+
+  it("falls back network failure threshold to 2 when policy value is missing", () => {
+    const store = useUserSettingsStore();
+
+    store.applySettings({
+      auto_run_enabled: true,
+      run_interval_minutes: 60,
+      request_delay_seconds: 2,
+      nav_visible_pages: ["dashboard", "scholars", "settings"],
+      policy: {
+        min_run_interval_minutes: 15,
+        min_request_delay_seconds: 2,
+        automation_allowed: true,
+        manual_run_allowed: true,
+        blocked_failure_threshold: 1,
+        network_failure_threshold: Number.NaN,
+        cooldown_blocked_seconds: 1800,
+        cooldown_network_seconds: 900,
+      },
+      safety_state: {
+        cooldown_active: false,
+        cooldown_reason: null,
+        cooldown_reason_label: null,
+        cooldown_until: null,
+        cooldown_remaining_seconds: 0,
+        recommended_action: null,
+        counters: {
+          consecutive_blocked_runs: 0,
+          consecutive_network_runs: 0,
+          cooldown_entry_count: 0,
+          blocked_start_count: 0,
+          last_blocked_failure_count: 0,
+          last_network_failure_count: 0,
+          last_evaluated_run_id: null,
+        },
+      },
+    });
+
+    expect(store.networkFailureThreshold).toBe(2);
+  });
+});

frontend/src/stores/user_settings.ts

@@ -84,7 +84,7 @@ export const useUserSettingsStore = defineStore("userSettings", {
         : 1;
       this.networkFailureThreshold = Number.isFinite(settings.policy?.network_failure_threshold)
         ? Math.max(1, settings.policy.network_failure_threshold)
-        : 1;
+        : 2;
       this.cooldownBlockedSeconds = Number.isFinite(settings.policy?.cooldown_blocked_seconds)
         ? Math.max(60, settings.policy.cooldown_blocked_seconds)
         : 1800;

tests/conftest.py

@@ -19,6 +19,8 @@ from app.settings import settings
 RESET_SQL = text(
     """
     TRUNCATE TABLE
+      author_search_cache_entries,
+      author_search_runtime_state,
       ingestion_queue_items,
       scholar_publications,
       crawl_runs,

tests/integration/test_migrations.py

@@ -11,10 +11,12 @@ EXPECTED_TABLES = {
     "scholar_publications",
     "crawl_runs",
     "ingestion_queue_items",
+    "author_search_runtime_state",
+    "author_search_cache_entries",
 }
 
 EXPECTED_ENUMS = {"run_status", "run_trigger_type"}
-EXPECTED_REVISION = "20260219_0009"
+EXPECTED_REVISION = "20260219_0010"
 
 
 @pytest.mark.integration

tests/unit/test_auth_primitives.py

@@ -32,3 +32,22 @@ def test_rate_limiter_enforces_windowed_attempt_budget() -> None:
     clock["now"] = 11.0
     assert limiter.check(key).allowed
 
+
+def test_rate_limiter_check_does_not_create_or_retain_empty_keys() -> None:
+    clock = {"now": 0.0}
+    limiter = SlidingWindowRateLimiter(
+        max_attempts=2,
+        window_seconds=10,
+        now=lambda: clock["now"],
+    )
+    key = "127.0.0.1:never-failed@example.com"
+
+    assert limiter.check(key).allowed
+    assert key not in limiter._attempts
+
+    limiter.record_failure(key)
+    assert key in limiter._attempts
+
+    clock["now"] = 11.0
+    assert limiter.check(key).allowed
+    assert key not in limiter._attempts

tests/unit/test_scholar_search_safety.py

@@ -1,11 +1,14 @@
 from __future__ import annotations
 
 import pytest
+from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.services import scholars as scholar_service
 from app.services.scholar_parser import ParseState
 from app.services.scholar_source import FetchResult
 
+pytestmark = [pytest.mark.integration, pytest.mark.db]
+
 
 class StubScholarSource:
     def __init__(self, fetch_results: list[FetchResult]) -> None:
@@ -21,13 +24,6 @@ class StubScholarSource:
         return self._fetch_results[index]
 
 
-@pytest.fixture(autouse=True)
-def reset_author_search_runtime_state() -> None:
-    scholar_service._reset_author_search_runtime_state_for_tests()
-    yield
-    scholar_service._reset_author_search_runtime_state_for_tests()
-
-
 def _ok_author_search_fetch() -> FetchResult:
     body = (
         "<html><body>"
@@ -74,11 +70,14 @@ def _network_timeout_fetch() -> FetchResult:
 
 
 @pytest.mark.asyncio
-async def test_search_author_candidates_serves_cached_response_for_same_query() -> None:
+async def test_search_author_candidates_serves_cached_response_for_same_query(
+    db_session: AsyncSession,
+) -> None:
     source = StubScholarSource([_ok_author_search_fetch()])
 
     first = await scholar_service.search_author_candidates(
         source=source,
+        db_session=db_session,
         query="Ada Lovelace",
         limit=10,
         network_error_retries=0,
@@ -94,6 +93,7 @@ async def test_search_author_candidates_serves_cached_response_for_same_query()
     )
     second = await scholar_service.search_author_candidates(
         source=source,
+        db_session=db_session,
         query="Ada Lovelace",
         limit=10,
         network_error_retries=0,
@@ -117,11 +117,14 @@ async def test_search_author_candidates_serves_cached_response_for_same_query()
 
 
 @pytest.mark.asyncio
-async def test_search_author_candidates_trips_cooldown_after_blocked_responses() -> None:
+async def test_search_author_candidates_trips_cooldown_after_blocked_responses(
+    db_session: AsyncSession,
+) -> None:
     source = StubScholarSource([_blocked_author_search_fetch()])
 
     first = await scholar_service.search_author_candidates(
         source=source,
+        db_session=db_session,
         query="Blocked Query",
         limit=10,
         network_error_retries=0,
@@ -137,6 +140,7 @@ async def test_search_author_candidates_trips_cooldown_after_blocked_responses()
     )
     second = await scholar_service.search_author_candidates(
         source=source,
+        db_session=db_session,
         query="Blocked Query",
         limit=10,
         network_error_retries=0,
@@ -161,11 +165,14 @@ async def test_search_author_candidates_trips_cooldown_after_blocked_responses()
 
 
 @pytest.mark.asyncio
-async def test_search_author_candidates_emits_cooldown_alert_warning_after_threshold() -> None:
+async def test_search_author_candidates_emits_cooldown_alert_warning_after_threshold(
+    db_session: AsyncSession,
+) -> None:
     source = StubScholarSource([_blocked_author_search_fetch()])
 
     _ = await scholar_service.search_author_candidates(
         source=source,
+        db_session=db_session,
         query="Blocked Query",
         limit=10,
         network_error_retries=0,
@@ -182,6 +189,7 @@ async def test_search_author_candidates_emits_cooldown_alert_warning_after_thres
     )
     second = await scholar_service.search_author_candidates(
         source=source,
+        db_session=db_session,
         query="Blocked Query",
         limit=10,
         network_error_retries=0,
@@ -202,7 +210,9 @@ async def test_search_author_candidates_emits_cooldown_alert_warning_after_thres
 
 
 @pytest.mark.asyncio
-async def test_search_author_candidates_adds_retry_threshold_warning() -> None:
+async def test_search_author_candidates_adds_retry_threshold_warning(
+    db_session: AsyncSession,
+) -> None:
     source = StubScholarSource(
         [
             _network_timeout_fetch(),
@@ -213,6 +223,7 @@ async def test_search_author_candidates_adds_retry_threshold_warning() -> None:
 
     result = await scholar_service.search_author_candidates(
         source=source,
+        db_session=db_session,
         query="Ada Lovelace",
         limit=10,
         network_error_retries=2,
@@ -234,11 +245,14 @@ async def test_search_author_candidates_adds_retry_threshold_warning() -> None:
 
 
 @pytest.mark.asyncio
-async def test_search_author_candidates_can_be_disabled_by_configuration() -> None:
+async def test_search_author_candidates_can_be_disabled_by_configuration(
+    db_session: AsyncSession,
+) -> None:
     source = StubScholarSource([_ok_author_search_fetch()])
 
     parsed = await scholar_service.search_author_candidates(
         source=source,
+        db_session=db_session,
         query="Ada Lovelace",
         limit=5,
         network_error_retries=0,

tests/unit/test_scholar_source.py

@@ -0,0 +1,13 @@
+from app.services.scholar_source import _build_profile_url
+
+
+def test_build_profile_url_includes_pagesize_for_initial_page() -> None:
+    url = _build_profile_url(
+        scholar_id="abcDEF123456",
+        cstart=0,
+        pagesize=100,
+    )
+
+    assert "user=abcDEF123456" in url
+    assert "pagesize=100" in url
+    assert "cstart=" not in url