Merge pull request #7 from JustinZeus/feat/scrape-safety-observability-and-run-state-ux

arden scrape safety telemetry and polish run state UX
2026-02-19 22:10:46 +01:00 · 2026-02-19 22:10:46 +01:00 · 1b1595c837
commit 1b1595c837
parent bd33e797a2 d7404abf18
33 changed files with 2074 additions and 96 deletions
--- a/.env.example
+++ b/.env.example
@ -46,6 +46,10 @@ LOG_REQUEST_SKIP_PATHS=/healthz
 LOG_REDACT_FIELDS=
 SCHEDULER_ENABLED=1
 SCHEDULER_TICK_SECONDS=60
+INGESTION_AUTOMATION_ALLOWED=1
+INGESTION_MANUAL_RUN_ALLOWED=1
+INGESTION_MIN_RUN_INTERVAL_MINUTES=15
+INGESTION_MIN_REQUEST_DELAY_SECONDS=2
 INGESTION_NETWORK_ERROR_RETRIES=1
 INGESTION_RETRY_BACKOFF_SECONDS=1.0
 INGESTION_MAX_PAGES_PER_SCHOLAR=30
@ -53,6 +57,8 @@ INGESTION_PAGE_SIZE=100
 INGESTION_ALERT_BLOCKED_FAILURE_THRESHOLD=1
 INGESTION_ALERT_NETWORK_FAILURE_THRESHOLD=2
 INGESTION_ALERT_RETRY_SCHEDULED_THRESHOLD=3
+INGESTION_SAFETY_COOLDOWN_BLOCKED_SECONDS=1800
+INGESTION_SAFETY_COOLDOWN_NETWORK_SECONDS=900
 INGESTION_CONTINUATION_QUEUE_ENABLED=1
 INGESTION_CONTINUATION_BASE_DELAY_SECONDS=120
 INGESTION_CONTINUATION_MAX_DELAY_SECONDS=3600
--- a/README.md
+++ b/README.md
@ -70,6 +70,7 @@ docker compose -f docker-compose.yml -f docker-compose.dev.yml down
 - `Dockerfile`: multi-stage build (frontend + backend runtime).
 - `README.md`: deployment and operations reference.
 - `CONTRIBUTING.md`: contribution policy and merge checklist.
+- `docs/ops/scrape_safety_runbook.md`: scrape cooldown and blocked-IP operations guide.
 - `scripts/check_no_generated_artifacts.sh`: tracked-artifact guard used by CI.

 ## Data Model Notes
@ -170,6 +171,10 @@ Notes:
 | --- | --- | --- | --- | --- |
 | `SCHEDULER_ENABLED` | `1` | boolean | deploy, dev | Enable background scheduler loop. |
 | `SCHEDULER_TICK_SECONDS` | `60` | integer >= 1 | deploy, dev | Scheduler interval in seconds. |
+| `INGESTION_AUTOMATION_ALLOWED` | `1` | boolean | deploy, dev | Global safety switch for scheduled/automatic checks. When disabled, auto-run settings are forced off. |
+| `INGESTION_MANUAL_RUN_ALLOWED` | `1` | boolean | deploy, dev | Global safety switch for manual checks from API/UI. |
+| `INGESTION_MIN_RUN_INTERVAL_MINUTES` | `15` | integer >= 15 | deploy, dev | Server-enforced minimum for user-configured automatic check interval. |
+| `INGESTION_MIN_REQUEST_DELAY_SECONDS` | `2` | integer >= 2 | deploy, dev | Server-enforced minimum delay between scholar requests. |
 | `SCHEDULER_QUEUE_BATCH_SIZE` | `10` | integer >= 1 | deploy, dev | Queue items processed per scheduler tick. |
 | `INGESTION_NETWORK_ERROR_RETRIES` | `1` | integer >= 0 | deploy, dev | Retries for transient ingestion network errors. |
 | `INGESTION_RETRY_BACKOFF_SECONDS` | `1.0` | float >= 0 | deploy, dev | Backoff delay for retry attempts. |
@ -178,6 +183,8 @@ Notes:
 | `INGESTION_ALERT_BLOCKED_FAILURE_THRESHOLD` | `1` | integer >= 1 | deploy, dev | Trigger blocked/captcha scrape alert flag when this many blocked failures occur in a run. |
 | `INGESTION_ALERT_NETWORK_FAILURE_THRESHOLD` | `2` | integer >= 1 | deploy, dev | Trigger network scrape alert flag when this many network failures occur in a run. |
 | `INGESTION_ALERT_RETRY_SCHEDULED_THRESHOLD` | `3` | integer >= 1 | deploy, dev | Trigger retry alert flag when scheduled retry count reaches this threshold in a run. |
+| `INGESTION_SAFETY_COOLDOWN_BLOCKED_SECONDS` | `1800` | integer >= 60 | deploy, dev | Cooldown duration applied when blocked-failure threshold is exceeded. |
+| `INGESTION_SAFETY_COOLDOWN_NETWORK_SECONDS` | `900` | integer >= 60 | deploy, dev | Cooldown duration applied when network-failure threshold is exceeded. |
 | `INGESTION_CONTINUATION_QUEUE_ENABLED` | `1` | boolean | deploy, dev | Enable continuation queue for long runs. |
 | `INGESTION_CONTINUATION_BASE_DELAY_SECONDS` | `120` | integer >= 0 | deploy, dev | Initial delay before retrying continuation queue items. |
 | `INGESTION_CONTINUATION_MAX_DELAY_SECONDS` | `3600` | integer >= 0 | deploy, dev | Maximum continuation retry delay. |
@ -200,6 +207,16 @@ Notes:
 | `SCHOLAR_NAME_SEARCH_ALERT_RETRY_COUNT_THRESHOLD` | `2` | integer >= 1 | deploy, dev | Emit retry-threshold observability warning when name-search retry count reaches this value. |
 | `SCHOLAR_NAME_SEARCH_ALERT_COOLDOWN_REJECTIONS_THRESHOLD` | `3` | integer >= 1 | deploy, dev | Emit cooldown-threshold observability alert after this many requests are rejected during active cooldown. |

+### Scrape Safety Operations
+
+- Structured safety events are emitted for:
+  - policy/manual run blocks,
+  - cooldown entered/cleared transitions,
+  - blocked/network/retry threshold trips,
+  - scheduler cooldown skips/deferments.
+- Use `LOG_FORMAT=json` and ship logs to your preferred collector for alerting.
+- Runbook: `docs/ops/scrape_safety_runbook.md`.
+
 ### Startup Bootstrap and DB Wait

 | Variable | Default | Options | Scope | Description |
--- a/alembic/versions/20260219_0009_user_settings_scrape_safety_state.py
+++ b/alembic/versions/20260219_0009_user_settings_scrape_safety_state.py
@ -0,0 +1,66 @@
+"""Add scrape safety state fields to user settings.
+
+Revision ID: 20260219_0009
+Revises: 20260219_0008
+Create Date: 2026-02-19 21: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_0009"
+down_revision: str | Sequence[str] | None = "20260219_0008"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+TABLE_NAME = "user_settings"
+
+
+def upgrade() -> None:
+    bind = op.get_bind()
+    inspector = sa.inspect(bind)
+    columns = {column["name"] for column in inspector.get_columns(TABLE_NAME)}
+
+    if "scrape_safety_state" not in columns:
+        op.add_column(
+            TABLE_NAME,
+            sa.Column(
+                "scrape_safety_state",
+                postgresql.JSONB(astext_type=sa.Text()),
+                nullable=False,
+                server_default=sa.text("'{}'::jsonb"),
+            ),
+        )
+
+    if "scrape_cooldown_until" not in columns:
+        op.add_column(
+            TABLE_NAME,
+            sa.Column("scrape_cooldown_until", sa.DateTime(timezone=True), nullable=True),
+        )
+
+    if "scrape_cooldown_reason" not in columns:
+        op.add_column(
+            TABLE_NAME,
+            sa.Column("scrape_cooldown_reason", sa.String(length=64), nullable=True),
+        )
+
+
+def downgrade() -> None:
+    bind = op.get_bind()
+    inspector = sa.inspect(bind)
+    columns = {column["name"] for column in inspector.get_columns(TABLE_NAME)}
+
+    if "scrape_cooldown_reason" in columns:
+        op.drop_column(TABLE_NAME, "scrape_cooldown_reason")
+
+    if "scrape_cooldown_until" in columns:
+        op.drop_column(TABLE_NAME, "scrape_cooldown_until")
+
+    if "scrape_safety_state" in columns:
+        op.drop_column(TABLE_NAME, "scrape_safety_state")
--- a/app/api/routers/runs.py
+++ b/app/api/routers/runs.py
@ -22,6 +22,7 @@ from app.api.schemas import (
 from app.db.models import RunStatus, RunTriggerType, User
 from app.db.session import get_db_session
 from app.services import ingestion as ingestion_service
+from app.services import run_safety as run_safety_service
 from app.services import runs as run_service
 from app.services import user_settings as user_settings_service
 from app.settings import settings
@ -253,6 +254,7 @@ def _manual_run_payload_from_run(
    *,
    idempotency_key: str | None,
    reused_existing_run: bool,
+    safety_state: dict[str, Any],
 ) -> dict[str, Any]:
    summary = run_service.extract_run_summary(run.error_log)
    return {
@ -265,9 +267,37 @@ def _manual_run_payload_from_run(
        "new_publication_count": int(run.new_pub_count or 0),
        "reused_existing_run": reused_existing_run,
        "idempotency_key": idempotency_key,
+        "safety_state": safety_state,
    }


+async def _load_safety_state(
+    db_session: AsyncSession,
+    *,
+    user_id: int,
+) -> dict[str, Any]:
+    user_settings = await user_settings_service.get_or_create_settings(
+        db_session,
+        user_id=user_id,
+    )
+    previous_safety_state = run_safety_service.get_safety_event_context(user_settings)
+    if run_safety_service.clear_expired_cooldown(user_settings):
+        await db_session.commit()
+        await db_session.refresh(user_settings)
+        logger.info(
+            "api.runs.safety_cooldown_cleared",
+            extra={
+                "event": "api.runs.safety_cooldown_cleared",
+                "user_id": user_id,
+                "reason": previous_safety_state.get("cooldown_reason"),
+                "cooldown_until": previous_safety_state.get("cooldown_until"),
+                "metric_name": "api_runs_safety_cooldown_cleared_total",
+                "metric_value": 1,
+            },
+        )
+    return run_safety_service.get_safety_state_payload(user_settings)
+
+
@router.get(
    "",
    response_model=RunsListEnvelope,
@ -285,10 +315,15 @@ async def list_runs(
        limit=limit,
        failed_only=failed_only,
    )
+    safety_state = await _load_safety_state(
+        db_session,
+        user_id=current_user.id,
+    )
    return success_payload(
        request,
        data={
            "runs": [_serialize_run(run) for run in runs],
+            "safety_state": safety_state,
        },
    )

@ -318,12 +353,17 @@ async def get_run(
    scholar_results = error_log.get("scholar_results")
    if not isinstance(scholar_results, list):
        scholar_results = []
+    safety_state = await _load_safety_state(
+        db_session,
+        user_id=current_user.id,
+    )
    return success_payload(
        request,
        data={
            "run": _serialize_run(run),
            "summary": run_service.extract_run_summary(error_log),
            "scholar_results": [_normalize_scholar_result(item) for item in scholar_results],
+            "safety_state": safety_state,
        },
    )

@ -338,6 +378,34 @@ async def run_manual(
    current_user: User = Depends(get_api_current_user),
    ingest_service: ingestion_service.ScholarIngestionService = Depends(get_ingestion_service),
 ):
+    safety_state = await _load_safety_state(
+        db_session,
+        user_id=current_user.id,
+    )
+    if not settings.ingestion_manual_run_allowed:
+        logger.warning(
+            "api.runs.manual_blocked_policy",
+            extra={
+                "event": "api.runs.manual_blocked_policy",
+                "user_id": current_user.id,
+                "policy": {"manual_run_allowed": False},
+                "safety_state": safety_state,
+                "metric_name": "api_runs_manual_blocked_policy_total",
+                "metric_value": 1,
+            },
+        )
+        raise ApiException(
+            status_code=403,
+            code="manual_runs_disabled",
+            message="Manual checks are disabled by server policy.",
+            details={
+                "policy": {
+                    "manual_run_allowed": False,
+                },
+                "safety_state": safety_state,
+            },
+        )
+
    idempotency_key = _normalize_idempotency_key(request.headers.get(IDEMPOTENCY_HEADER))
    if idempotency_key is not None:
        previous_run = await run_service.get_manual_run_by_idempotency_key(
@ -362,6 +430,7 @@ async def run_manual(
                    previous_run,
                    idempotency_key=idempotency_key,
                    reused_existing_run=True,
+                    safety_state=safety_state,
                ),
            )

@ -393,6 +462,28 @@ async def run_manual(
            code="run_in_progress",
            message="A run is already in progress for this account.",
        ) from exc
+    except ingestion_service.RunBlockedBySafetyPolicyError as exc:
+        await db_session.rollback()
+        logger.info(
+            "api.runs.manual_blocked_safety",
+            extra={
+                "event": "api.runs.manual_blocked_safety",
+                "user_id": current_user.id,
+                "reason": exc.safety_state.get("cooldown_reason"),
+                "cooldown_until": exc.safety_state.get("cooldown_until"),
+                "cooldown_remaining_seconds": exc.safety_state.get("cooldown_remaining_seconds"),
+                "metric_name": "api_runs_manual_blocked_safety_total",
+                "metric_value": 1,
+            },
+        )
+        raise ApiException(
+            status_code=429,
+            code=exc.code,
+            message=exc.message,
+            details={
+                "safety_state": exc.safety_state,
+            },
+        ) from exc
    except IntegrityError as exc:
        await db_session.rollback()
        if idempotency_key is not None:
@ -418,6 +509,10 @@ async def run_manual(
                        existing_run,
                        idempotency_key=idempotency_key,
                        reused_existing_run=True,
+                        safety_state=await _load_safety_state(
+                            db_session,
+                            user_id=current_user.id,
+                        ),
                    ),
                )
        logger.exception(
@ -447,6 +542,10 @@ async def run_manual(
            message="Manual run failed.",
        ) from exc

+    current_safety_state = await _load_safety_state(
+        db_session,
+        user_id=current_user.id,
+    )
    return success_payload(
        request,
        data={
@ -459,6 +558,7 @@ async def run_manual(
            "new_publication_count": run_summary.new_publication_count,
            "reused_existing_run": False,
            "idempotency_key": idempotency_key,
+            "safety_state": current_safety_state,
        },
    )

--- a/app/api/routers/settings.py
+++ b/app/api/routers/settings.py
@ -11,19 +11,38 @@ from app.api.responses import success_payload
 from app.api.schemas import SettingsEnvelope, SettingsUpdateRequest
 from app.db.models import User
 from app.db.session import get_db_session
+from app.services import run_safety as run_safety_service
 from app.services import user_settings as user_settings_service
+from app.settings import settings as settings_module

 logger = logging.getLogger(__name__)

 router = APIRouter(prefix="/settings", tags=["api-settings"])


-def _serialize_settings(settings) -> dict[str, object]:
+def _serialize_settings(user_settings) -> dict[str, object]:
+    min_run_interval_minutes = user_settings_service.resolve_run_interval_minimum(
+        settings_module.ingestion_min_run_interval_minutes
+    )
+    min_request_delay_seconds = user_settings_service.resolve_request_delay_minimum(
+        settings_module.ingestion_min_request_delay_seconds
+    )
    return {
-        "auto_run_enabled": bool(settings.auto_run_enabled),
-        "run_interval_minutes": int(settings.run_interval_minutes),
-        "request_delay_seconds": int(settings.request_delay_seconds),
-        "nav_visible_pages": list(settings.nav_visible_pages or []),
+        "auto_run_enabled": bool(user_settings.auto_run_enabled) and settings_module.ingestion_automation_allowed,
+        "run_interval_minutes": int(user_settings.run_interval_minutes),
+        "request_delay_seconds": int(user_settings.request_delay_seconds),
+        "nav_visible_pages": list(user_settings.nav_visible_pages or []),
+        "policy": {
+            "min_run_interval_minutes": min_run_interval_minutes,
+            "min_request_delay_seconds": min_request_delay_seconds,
+            "automation_allowed": bool(settings_module.ingestion_automation_allowed),
+            "manual_run_allowed": bool(settings_module.ingestion_manual_run_allowed),
+            "blocked_failure_threshold": max(1, int(settings_module.ingestion_alert_blocked_failure_threshold)),
+            "network_failure_threshold": max(1, int(settings_module.ingestion_alert_network_failure_threshold)),
+            "cooldown_blocked_seconds": max(60, int(settings_module.ingestion_safety_cooldown_blocked_seconds)),
+            "cooldown_network_seconds": max(60, int(settings_module.ingestion_safety_cooldown_network_seconds)),
+        },
+        "safety_state": run_safety_service.get_safety_state_payload(user_settings),
    }


@ -36,13 +55,28 @@ async def get_settings(
    db_session: AsyncSession = Depends(get_db_session),
    current_user: User = Depends(get_api_current_user),
 ):
-    settings = await user_settings_service.get_or_create_settings(
+    user_settings = await user_settings_service.get_or_create_settings(
        db_session,
        user_id=current_user.id,
    )
+    previous_safety_state = run_safety_service.get_safety_event_context(user_settings)
+    if run_safety_service.clear_expired_cooldown(user_settings):
+        await db_session.commit()
+        await db_session.refresh(user_settings)
+        logger.info(
+            "api.settings.safety_cooldown_cleared",
+            extra={
+                "event": "api.settings.safety_cooldown_cleared",
+                "user_id": current_user.id,
+                "reason": previous_safety_state.get("cooldown_reason"),
+                "cooldown_until": previous_safety_state.get("cooldown_until"),
+                "metric_name": "api_settings_safety_cooldown_cleared_total",
+                "metric_value": 1,
+            },
+        )
    return success_payload(
        request,
-        data=_serialize_settings(settings),
+        data=_serialize_settings(user_settings),
    )


@ -56,22 +90,31 @@ async def update_settings(
    db_session: AsyncSession = Depends(get_db_session),
    current_user: User = Depends(get_api_current_user),
 ):
-    settings = await user_settings_service.get_or_create_settings(
+    user_settings = await user_settings_service.get_or_create_settings(
        db_session,
        user_id=current_user.id,
    )

+    min_run_interval_minutes = user_settings_service.resolve_run_interval_minimum(
+        settings_module.ingestion_min_run_interval_minutes
+    )
+    min_request_delay_seconds = user_settings_service.resolve_request_delay_minimum(
+        settings_module.ingestion_min_request_delay_seconds
+    )
+
    try:
        parsed_interval = user_settings_service.parse_run_interval_minutes(
-            str(payload.run_interval_minutes)
+            str(payload.run_interval_minutes),
+            minimum=min_run_interval_minutes,
        )
        parsed_delay = user_settings_service.parse_request_delay_seconds(
-            str(payload.request_delay_seconds)
+            str(payload.request_delay_seconds),
+            minimum=min_request_delay_seconds,
        )
        parsed_nav_visible_pages = user_settings_service.parse_nav_visible_pages(
            payload.nav_visible_pages
            if payload.nav_visible_pages is not None
-            else list(settings.nav_visible_pages or user_settings_service.DEFAULT_NAV_VISIBLE_PAGES)
+            else list(user_settings.nav_visible_pages or user_settings_service.DEFAULT_NAV_VISIBLE_PAGES)
        )
    except user_settings_service.UserSettingsServiceError as exc:
        raise ApiException(
@ -80,14 +123,33 @@ async def update_settings(
            message=str(exc),
        ) from exc

+    effective_auto_run_enabled = (
+        bool(payload.auto_run_enabled) and settings_module.ingestion_automation_allowed
+    )
+
    updated = await user_settings_service.update_settings(
        db_session,
-        settings=settings,
-        auto_run_enabled=bool(payload.auto_run_enabled),
+        settings=user_settings,
+        auto_run_enabled=effective_auto_run_enabled,
        run_interval_minutes=parsed_interval,
        request_delay_seconds=parsed_delay,
        nav_visible_pages=parsed_nav_visible_pages,
    )
+    previous_safety_state = run_safety_service.get_safety_event_context(updated)
+    if run_safety_service.clear_expired_cooldown(updated):
+        await db_session.commit()
+        await db_session.refresh(updated)
+        logger.info(
+            "api.settings.safety_cooldown_cleared",
+            extra={
+                "event": "api.settings.safety_cooldown_cleared",
+                "user_id": current_user.id,
+                "reason": previous_safety_state.get("cooldown_reason"),
+                "cooldown_until": previous_safety_state.get("cooldown_until"),
+                "metric_name": "api_settings_safety_cooldown_cleared_total",
+                "metric_value": 1,
+            },
+        )
    logger.info(
        "api.settings.updated",
        extra={
--- a/app/api/schemas.py
+++ b/app/api/schemas.py
@ -202,6 +202,7 @@ class RunListItemData(BaseModel):

 class RunsListData(BaseModel):
    runs: list[RunListItemData]
+    safety_state: ScrapeSafetyStateData

    model_config = ConfigDict(extra="forbid")

@ -227,6 +228,30 @@ class RunSummaryData(BaseModel):
    model_config = ConfigDict(extra="forbid")


+class ScrapeSafetyCountersData(BaseModel):
+    consecutive_blocked_runs: int = 0
+    consecutive_network_runs: int = 0
+    cooldown_entry_count: int = 0
+    blocked_start_count: int = 0
+    last_blocked_failure_count: int = 0
+    last_network_failure_count: int = 0
+    last_evaluated_run_id: int | None = None
+
+    model_config = ConfigDict(extra="forbid")
+
+
+class ScrapeSafetyStateData(BaseModel):
+    cooldown_active: bool
+    cooldown_reason: str | None = None
+    cooldown_reason_label: str | None = None
+    cooldown_until: datetime | None = None
+    cooldown_remaining_seconds: int = 0
+    recommended_action: str | None = None
+    counters: ScrapeSafetyCountersData = Field(default_factory=ScrapeSafetyCountersData)
+
+    model_config = ConfigDict(extra="forbid")
+
+
 class RunAttemptLogData(BaseModel):
    attempt: int
    cstart: int
@ -293,6 +318,7 @@ class RunDetailData(BaseModel):
    run: RunListItemData
    summary: RunSummaryData
    scholar_results: list[RunScholarResultData]
+    safety_state: ScrapeSafetyStateData

    model_config = ConfigDict(extra="forbid")

@ -314,6 +340,7 @@ class ManualRunData(BaseModel):
    new_publication_count: int
    reused_existing_run: bool
    idempotency_key: str | None = None
+    safety_state: ScrapeSafetyStateData

    model_config = ConfigDict(extra="forbid")

@ -429,11 +456,26 @@ class AdminResetPasswordRequest(BaseModel):
    model_config = ConfigDict(extra="forbid")


+class SettingsPolicyData(BaseModel):
+    min_run_interval_minutes: int
+    min_request_delay_seconds: int
+    automation_allowed: bool
+    manual_run_allowed: bool
+    blocked_failure_threshold: int
+    network_failure_threshold: int
+    cooldown_blocked_seconds: int
+    cooldown_network_seconds: int
+
+    model_config = ConfigDict(extra="forbid")
+
+
 class SettingsData(BaseModel):
    auto_run_enabled: bool
    run_interval_minutes: int
    request_delay_seconds: int
    nav_visible_pages: list[str]
+    policy: SettingsPolicyData
+    safety_state: ScrapeSafetyStateData

    model_config = ConfigDict(extra="forbid")

--- a/app/db/models.py
+++ b/app/db/models.py
@ -105,6 +105,13 @@ class UserSetting(Base):
            '\'["dashboard","scholars","publications","settings","style-guide","runs","users"]\'::jsonb'
        ),
    )
+    scrape_safety_state: Mapped[dict] = mapped_column(
+        JSONB,
+        nullable=False,
+        server_default=text("'{}'::jsonb"),
+    )
+    scrape_cooldown_until: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+    scrape_cooldown_reason: Mapped[str | None] = mapped_column(String(64))
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now()
    )
--- a/app/services/ingestion.py
+++ b/app/services/ingestion.py
@ -22,6 +22,8 @@ from app.db.models import (
    ScholarPublication,
 )
 from app.services import continuation_queue as queue_service
+from app.services import run_safety as run_safety_service
+from app.services import user_settings as user_settings_service
 from app.services.scholar_parser import (
    ParseState,
    ParsedProfilePage,
@ -29,6 +31,7 @@ from app.services.scholar_parser import (
    parse_profile_page,
 )
 from app.services.scholar_source import FetchResult, ScholarSource
+from app.settings import settings

 TITLE_ALNUM_RE = re.compile(r"[^a-z0-9]+")
 WORD_RE = re.compile(r"[a-z0-9]+")
@ -88,6 +91,20 @@ class RunAlreadyInProgressError(RuntimeError):
    """Raised when a run lock for a user is already held by another process."""


+class RunBlockedBySafetyPolicyError(RuntimeError):
+    def __init__(
+        self,
+        *,
+        code: str,
+        message: str,
+        safety_state: dict[str, Any],
+    ) -> None:
+        super().__init__(message)
+        self.code = code
+        self.message = message
+        self.safety_state = safety_state
+
+
 def _int_or_default(value: Any, default: int = 0) -> int:
    try:
        return int(value)
@ -134,6 +151,61 @@ class ScholarIngestionService:
        alert_network_failure_threshold: int = 2,
        alert_retry_scheduled_threshold: int = 3,
    ) -> RunExecutionSummary:
+        user_settings = await user_settings_service.get_or_create_settings(
+            db_session,
+            user_id=user_id,
+        )
+        now_utc = datetime.now(timezone.utc)
+        previous_safety_state = run_safety_service.get_safety_event_context(
+            user_settings,
+            now_utc=now_utc,
+        )
+        if run_safety_service.clear_expired_cooldown(
+            user_settings,
+            now_utc=now_utc,
+        ):
+            await db_session.commit()
+            await db_session.refresh(user_settings)
+            logger.info(
+                "ingestion.safety_cooldown_cleared",
+                extra={
+                    "event": "ingestion.safety_cooldown_cleared",
+                    "user_id": user_id,
+                    "reason": previous_safety_state.get("cooldown_reason"),
+                    "cooldown_until": previous_safety_state.get("cooldown_until"),
+                    "metric_name": "ingestion_safety_cooldown_cleared_total",
+                    "metric_value": 1,
+                },
+            )
+            now_utc = datetime.now(timezone.utc)
+        if run_safety_service.is_cooldown_active(user_settings, now_utc=now_utc):
+            safety_state = run_safety_service.register_cooldown_blocked_start(
+                user_settings,
+                now_utc=now_utc,
+            )
+            await db_session.commit()
+            logger.warning(
+                "ingestion.safety_policy_blocked_run_start",
+                extra={
+                    "event": "ingestion.safety_policy_blocked_run_start",
+                    "user_id": user_id,
+                    "trigger_type": trigger_type.value,
+                    "reason": safety_state.get("cooldown_reason"),
+                    "cooldown_until": safety_state.get("cooldown_until"),
+                    "cooldown_remaining_seconds": safety_state.get("cooldown_remaining_seconds"),
+                    "blocked_start_count": (
+                        (safety_state.get("counters") or {}).get("blocked_start_count")
+                    ),
+                    "metric_name": "ingestion_safety_run_start_blocked_total",
+                    "metric_value": 1,
+                },
+            )
+            raise RunBlockedBySafetyPolicyError(
+                code="scrape_cooldown_active",
+                message="Scrape safety cooldown is active; run start is temporarily blocked.",
+                safety_state=safety_state,
+            )
+
        lock_acquired = await self._try_acquire_user_lock(
            db_session,
            user_id=user_id,
@ -505,6 +577,8 @@ class ScholarIngestionService:
                    "crawl_run_id": run.id,
                    "blocked_failure_count": blocked_failure_count,
                    "threshold": bounded_blocked_threshold,
+                    "metric_name": "ingestion_blocked_failure_threshold_exceeded_total",
+                    "metric_value": 1,
                },
            )
        if alert_flags["network_failure_threshold_exceeded"]:
@ -516,6 +590,8 @@ class ScholarIngestionService:
                    "crawl_run_id": run.id,
                    "network_failure_count": network_failure_count,
                    "threshold": bounded_network_threshold,
+                    "metric_name": "ingestion_network_failure_threshold_exceeded_total",
+                    "metric_value": 1,
                },
            )
        if alert_flags["retry_scheduled_threshold_exceeded"]:
@ -527,6 +603,56 @@ class ScholarIngestionService:
                    "crawl_run_id": run.id,
                    "retries_scheduled_count": retries_scheduled_count,
                    "threshold": bounded_retry_threshold,
+                    "metric_name": "ingestion_retry_scheduled_threshold_exceeded_total",
+                    "metric_value": 1,
+                },
+            )
+
+        pre_apply_safety_state = run_safety_service.get_safety_event_context(
+            user_settings,
+            now_utc=datetime.now(timezone.utc),
+        )
+        safety_state, cooldown_trigger_reason = run_safety_service.apply_run_safety_outcome(
+            user_settings,
+            run_id=int(run.id),
+            blocked_failure_count=blocked_failure_count,
+            network_failure_count=network_failure_count,
+            blocked_failure_threshold=bounded_blocked_threshold,
+            network_failure_threshold=bounded_network_threshold,
+            blocked_cooldown_seconds=settings.ingestion_safety_cooldown_blocked_seconds,
+            network_cooldown_seconds=settings.ingestion_safety_cooldown_network_seconds,
+            now_utc=datetime.now(timezone.utc),
+        )
+        if cooldown_trigger_reason is not None:
+            logger.warning(
+                "ingestion.safety_cooldown_entered",
+                extra={
+                    "event": "ingestion.safety_cooldown_entered",
+                    "user_id": user_id,
+                    "crawl_run_id": run.id,
+                    "reason": cooldown_trigger_reason,
+                    "blocked_failure_count": blocked_failure_count,
+                    "network_failure_count": network_failure_count,
+                    "blocked_failure_threshold": bounded_blocked_threshold,
+                    "network_failure_threshold": bounded_network_threshold,
+                    "cooldown_until": safety_state.get("cooldown_until"),
+                    "cooldown_remaining_seconds": safety_state.get("cooldown_remaining_seconds"),
+                    "safety_counters": safety_state.get("counters", {}),
+                    "metric_name": "ingestion_safety_cooldown_entered_total",
+                    "metric_value": 1,
+                },
+            )
+        elif pre_apply_safety_state.get("cooldown_active") and not safety_state.get("cooldown_active"):
+            logger.info(
+                "ingestion.safety_cooldown_cleared",
+                extra={
+                    "event": "ingestion.safety_cooldown_cleared",
+                    "user_id": user_id,
+                    "crawl_run_id": run.id,
+                    "reason": pre_apply_safety_state.get("cooldown_reason"),
+                    "cooldown_until": pre_apply_safety_state.get("cooldown_until"),
+                    "metric_name": "ingestion_safety_cooldown_cleared_total",
+                    "metric_value": 1,
                },
            )

--- a/app/services/run_safety.py
+++ b/app/services/run_safety.py
@ -0,0 +1,239 @@
+from __future__ import annotations
+
+from datetime import datetime, timedelta, timezone
+from typing import Any
+
+from app.db.models import UserSetting
+
+COOLDOWN_REASON_BLOCKED_FAILURE_THRESHOLD = "blocked_failure_threshold_exceeded"
+COOLDOWN_REASON_NETWORK_FAILURE_THRESHOLD = "network_failure_threshold_exceeded"
+
+_COUNTER_CONSECUTIVE_BLOCKED_RUNS = "consecutive_blocked_runs"
+_COUNTER_CONSECUTIVE_NETWORK_RUNS = "consecutive_network_runs"
+_COUNTER_COOLDOWN_ENTRY_COUNT = "cooldown_entry_count"
+_COUNTER_BLOCKED_START_COUNT = "blocked_start_count"
+_COUNTER_LAST_BLOCKED_FAILURE_COUNT = "last_blocked_failure_count"
+_COUNTER_LAST_NETWORK_FAILURE_COUNT = "last_network_failure_count"
+_COUNTER_LAST_EVALUATED_RUN_ID = "last_evaluated_run_id"
+
+
+def _utcnow() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+def _safe_int(value: Any, default: int = 0) -> int:
+    try:
+        return int(value)
+    except (TypeError, ValueError):
+        return default
+
+
+def _safe_optional_int(value: Any) -> int | None:
+    if value is None:
+        return None
+    try:
+        return int(value)
+    except (TypeError, ValueError):
+        return None
+
+
+def _normalize_datetime(value: datetime | None) -> datetime | None:
+    if value is None:
+        return None
+    if value.tzinfo is None:
+        return value.replace(tzinfo=timezone.utc)
+    return value.astimezone(timezone.utc)
+
+
+def _state_dict(settings: UserSetting) -> dict[str, Any]:
+    state = settings.scrape_safety_state
+    if isinstance(state, dict):
+        return state
+    return {}
+
+
+def _counters_from_state(settings: UserSetting) -> dict[str, Any]:
+    state = _state_dict(settings)
+    return {
+        _COUNTER_CONSECUTIVE_BLOCKED_RUNS: max(
+            0,
+            _safe_int(state.get(_COUNTER_CONSECUTIVE_BLOCKED_RUNS), 0),
+        ),
+        _COUNTER_CONSECUTIVE_NETWORK_RUNS: max(
+            0,
+            _safe_int(state.get(_COUNTER_CONSECUTIVE_NETWORK_RUNS), 0),
+        ),
+        _COUNTER_COOLDOWN_ENTRY_COUNT: max(
+            0,
+            _safe_int(state.get(_COUNTER_COOLDOWN_ENTRY_COUNT), 0),
+        ),
+        _COUNTER_BLOCKED_START_COUNT: max(
+            0,
+            _safe_int(state.get(_COUNTER_BLOCKED_START_COUNT), 0),
+        ),
+        _COUNTER_LAST_BLOCKED_FAILURE_COUNT: max(
+            0,
+            _safe_int(state.get(_COUNTER_LAST_BLOCKED_FAILURE_COUNT), 0),
+        ),
+        _COUNTER_LAST_NETWORK_FAILURE_COUNT: max(
+            0,
+            _safe_int(state.get(_COUNTER_LAST_NETWORK_FAILURE_COUNT), 0),
+        ),
+        _COUNTER_LAST_EVALUATED_RUN_ID: _safe_optional_int(
+            state.get(_COUNTER_LAST_EVALUATED_RUN_ID),
+        ),
+    }
+
+
+def _cooldown_reason_label(reason: str | None) -> str | None:
+    if reason == COOLDOWN_REASON_BLOCKED_FAILURE_THRESHOLD:
+        return "Blocked responses exceeded safety threshold"
+    if reason == COOLDOWN_REASON_NETWORK_FAILURE_THRESHOLD:
+        return "Network failures exceeded safety threshold"
+    return None
+
+
+def _recommended_action(reason: str | None) -> str | None:
+    if reason == COOLDOWN_REASON_BLOCKED_FAILURE_THRESHOLD:
+        return (
+            "Google Scholar appears to be blocking requests. Wait for cooldown to expire, "
+            "increase request delay, and avoid repeated manual retries."
+        )
+    if reason == COOLDOWN_REASON_NETWORK_FAILURE_THRESHOLD:
+        return (
+            "Network failures crossed the threshold. Verify connectivity and retry after cooldown."
+        )
+    return None
+
+
+def is_cooldown_active(
+    settings: UserSetting,
+    *,
+    now_utc: datetime | None = None,
+) -> bool:
+    now = now_utc or _utcnow()
+    cooldown_until = _normalize_datetime(settings.scrape_cooldown_until)
+    if cooldown_until is None:
+        return False
+    return cooldown_until > now
+
+
+def clear_expired_cooldown(
+    settings: UserSetting,
+    *,
+    now_utc: datetime | None = None,
+) -> bool:
+    now = now_utc or _utcnow()
+    cooldown_until = _normalize_datetime(settings.scrape_cooldown_until)
+    if cooldown_until is None:
+        return False
+    if cooldown_until > now:
+        return False
+    settings.scrape_cooldown_until = None
+    settings.scrape_cooldown_reason = None
+    return True
+
+
+def register_cooldown_blocked_start(
+    settings: UserSetting,
+    *,
+    now_utc: datetime | None = None,
+) -> dict[str, Any]:
+    now = now_utc or _utcnow()
+    counters = _counters_from_state(settings)
+    counters[_COUNTER_BLOCKED_START_COUNT] = int(counters[_COUNTER_BLOCKED_START_COUNT]) + 1
+    settings.scrape_safety_state = counters
+    return get_safety_state_payload(settings, now_utc=now)
+
+
+def apply_run_safety_outcome(
+    settings: UserSetting,
+    *,
+    run_id: int,
+    blocked_failure_count: int,
+    network_failure_count: int,
+    blocked_failure_threshold: int,
+    network_failure_threshold: int,
+    blocked_cooldown_seconds: int,
+    network_cooldown_seconds: int,
+    now_utc: datetime | None = None,
+) -> tuple[dict[str, Any], str | None]:
+    now = now_utc or _utcnow()
+    counters = _counters_from_state(settings)
+
+    bounded_blocked_failures = max(0, int(blocked_failure_count))
+    bounded_network_failures = max(0, int(network_failure_count))
+
+    counters[_COUNTER_LAST_BLOCKED_FAILURE_COUNT] = bounded_blocked_failures
+    counters[_COUNTER_LAST_NETWORK_FAILURE_COUNT] = bounded_network_failures
+    counters[_COUNTER_LAST_EVALUATED_RUN_ID] = int(run_id)
+    counters[_COUNTER_CONSECUTIVE_BLOCKED_RUNS] = (
+        int(counters[_COUNTER_CONSECUTIVE_BLOCKED_RUNS]) + 1
+        if bounded_blocked_failures > 0
+        else 0
+    )
+    counters[_COUNTER_CONSECUTIVE_NETWORK_RUNS] = (
+        int(counters[_COUNTER_CONSECUTIVE_NETWORK_RUNS]) + 1
+        if bounded_network_failures > 0
+        else 0
+    )
+
+    reason: str | None = None
+    cooldown_seconds = 0
+
+    if bounded_blocked_failures >= max(1, int(blocked_failure_threshold)):
+        reason = COOLDOWN_REASON_BLOCKED_FAILURE_THRESHOLD
+        cooldown_seconds = max(60, int(blocked_cooldown_seconds))
+    elif bounded_network_failures >= max(1, int(network_failure_threshold)):
+        reason = COOLDOWN_REASON_NETWORK_FAILURE_THRESHOLD
+        cooldown_seconds = max(60, int(network_cooldown_seconds))
+
+    if reason is not None:
+        settings.scrape_cooldown_reason = reason
+        settings.scrape_cooldown_until = now + timedelta(seconds=cooldown_seconds)
+        counters[_COUNTER_COOLDOWN_ENTRY_COUNT] = int(counters[_COUNTER_COOLDOWN_ENTRY_COUNT]) + 1
+    else:
+        clear_expired_cooldown(settings, now_utc=now)
+
+    settings.scrape_safety_state = counters
+    return get_safety_state_payload(settings, now_utc=now), reason
+
+
+def get_safety_state_payload(
+    settings: UserSetting,
+    *,
+    now_utc: datetime | None = None,
+) -> dict[str, Any]:
+    now = now_utc or _utcnow()
+    cooldown_until = _normalize_datetime(settings.scrape_cooldown_until)
+    cooldown_active = bool(cooldown_until is not None and cooldown_until > now)
+    cooldown_remaining_seconds = 0
+    if cooldown_active and cooldown_until is not None:
+        cooldown_remaining_seconds = max(0, int((cooldown_until - now).total_seconds()))
+
+    reason = settings.scrape_cooldown_reason if cooldown_active else None
+
+    return {
+        "cooldown_active": cooldown_active,
+        "cooldown_reason": reason,
+        "cooldown_reason_label": _cooldown_reason_label(reason),
+        "cooldown_until": cooldown_until,
+        "cooldown_remaining_seconds": cooldown_remaining_seconds,
+        "recommended_action": _recommended_action(reason),
+        "counters": _counters_from_state(settings),
+    }
+
+
+def get_safety_event_context(
+    settings: UserSetting,
+    *,
+    now_utc: datetime | None = None,
+) -> dict[str, Any]:
+    payload = get_safety_state_payload(settings, now_utc=now_utc)
+    return {
+        "cooldown_active": bool(payload.get("cooldown_active")),
+        "cooldown_reason": payload.get("cooldown_reason"),
+        "cooldown_until": payload.get("cooldown_until"),
+        "cooldown_remaining_seconds": int(payload.get("cooldown_remaining_seconds") or 0),
+        "safety_counters": payload.get("counters", {}),
+    }
--- a/app/services/scheduler.py
+++ b/app/services/scheduler.py
@ -18,7 +18,11 @@ from app.db.models import (
 )
 from app.db.session import get_session_factory
 from app.services import continuation_queue as queue_service
-from app.services.ingestion import RunAlreadyInProgressError, ScholarIngestionService
+from app.services.ingestion import (
+    RunAlreadyInProgressError,
+    RunBlockedBySafetyPolicyError,
+    ScholarIngestionService,
+)
 from app.services.scholar_source import LiveScholarSource
 from app.settings import settings

@ -30,6 +34,8 @@ class _AutoRunCandidate:
    user_id: int
    run_interval_minutes: int
    request_delay_seconds: int
+    cooldown_until: datetime | None
+    cooldown_reason: str | None


 class SchedulerService:
@ -134,6 +140,9 @@ class SchedulerService:
            await self._run_candidate(candidate)

    async def _load_candidates(self) -> list[_AutoRunCandidate]:
+        if not settings.ingestion_automation_allowed:
+            return []
+
        session_factory = get_session_factory()
        async with session_factory() as session:
            result = await session.execute(
@ -141,6 +150,8 @@ class SchedulerService:
                    UserSetting.user_id,
                    UserSetting.run_interval_minutes,
                    UserSetting.request_delay_seconds,
+                    UserSetting.scrape_cooldown_until,
+                    UserSetting.scrape_cooldown_reason,
                )
                .join(User, User.id == UserSetting.user_id)
                .where(
@ -150,14 +161,35 @@ class SchedulerService:
                .order_by(UserSetting.user_id.asc())
            )
            rows = result.all()
-        return [
-            _AutoRunCandidate(
-                user_id=int(user_id),
-                run_interval_minutes=int(run_interval_minutes),
-                request_delay_seconds=int(request_delay_seconds),
+        now_utc = datetime.now(timezone.utc)
+        candidates: list[_AutoRunCandidate] = []
+        for user_id, run_interval_minutes, request_delay_seconds, cooldown_until, cooldown_reason in rows:
+            if cooldown_until is not None and cooldown_until.tzinfo is None:
+                cooldown_until = cooldown_until.replace(tzinfo=timezone.utc)
+            if cooldown_until is not None and cooldown_until > now_utc:
+                logger.info(
+                    "scheduler.run_skipped_safety_cooldown_precheck",
+                    extra={
+                        "event": "scheduler.run_skipped_safety_cooldown_precheck",
+                        "user_id": int(user_id),
+                        "reason": cooldown_reason,
+                        "cooldown_until": cooldown_until,
+                        "cooldown_remaining_seconds": int((cooldown_until - now_utc).total_seconds()),
+                        "metric_name": "scheduler_run_skipped_safety_cooldown_total",
+                        "metric_value": 1,
+                    },
+                )
+                continue
+            candidates.append(
+                _AutoRunCandidate(
+                    user_id=int(user_id),
+                    run_interval_minutes=int(run_interval_minutes),
+                    request_delay_seconds=int(request_delay_seconds),
+                    cooldown_until=cooldown_until,
+                    cooldown_reason=(str(cooldown_reason).strip() if cooldown_reason else None),
+                )
            )
-            for user_id, run_interval_minutes, request_delay_seconds in rows
-        ]
+        return candidates

    async def _is_due(self, candidate: _AutoRunCandidate, *, now: datetime) -> bool:
        session_factory = get_session_factory()
@ -210,6 +242,21 @@ class SchedulerService:
                    },
                )
                return
+            except RunBlockedBySafetyPolicyError as exc:
+                await session.rollback()
+                logger.info(
+                    "scheduler.run_skipped_safety_cooldown",
+                    extra={
+                        "event": "scheduler.run_skipped_safety_cooldown",
+                        "user_id": candidate.user_id,
+                        "reason": exc.safety_state.get("cooldown_reason"),
+                        "cooldown_until": exc.safety_state.get("cooldown_until"),
+                        "cooldown_remaining_seconds": exc.safety_state.get("cooldown_remaining_seconds"),
+                        "metric_name": "scheduler_run_skipped_safety_cooldown_total",
+                        "metric_value": 1,
+                    },
+                )
+                return
            except Exception:
                await session.rollback()
                logger.exception(
@ -354,6 +401,34 @@ class SchedulerService:
                    },
                )
                return
+            except RunBlockedBySafetyPolicyError as exc:
+                await session.rollback()
+                cooldown_remaining_seconds = max(
+                    self._tick_seconds,
+                    int(exc.safety_state.get("cooldown_remaining_seconds") or 0),
+                )
+                async with session_factory() as recovery_session:
+                    await queue_service.reschedule_job(
+                        recovery_session,
+                        job_id=job.id,
+                        delay_seconds=max(self._tick_seconds, cooldown_remaining_seconds),
+                        reason="scrape_safety_cooldown",
+                        error=str(exc.message),
+                    )
+                    await recovery_session.commit()
+                logger.info(
+                    "scheduler.queue_item_deferred_safety_cooldown",
+                    extra={
+                        "event": "scheduler.queue_item_deferred_safety_cooldown",
+                        "queue_item_id": job.id,
+                        "user_id": job.user_id,
+                        "reason": exc.safety_state.get("cooldown_reason"),
+                        "cooldown_remaining_seconds": cooldown_remaining_seconds,
+                        "metric_name": "scheduler_queue_item_deferred_safety_cooldown_total",
+                        "metric_value": 1,
+                    },
+                )
+                return
            except Exception as exc:
                await session.rollback()
                async with session_factory() as recovery_session:
--- a/app/services/user_settings.py
+++ b/app/services/user_settings.py
@ -33,25 +33,49 @@ REQUIRED_NAV_PAGES = (
    NAV_PAGE_SETTINGS,
 )
 DEFAULT_NAV_VISIBLE_PAGES = list(ALLOWED_NAV_PAGES)
+HARD_MIN_RUN_INTERVAL_MINUTES = 15
+HARD_MIN_REQUEST_DELAY_SECONDS = 2


-def parse_run_interval_minutes(value: str) -> int:
+def resolve_run_interval_minimum(configured_minimum: int | None) -> int:
+    try:
+        parsed = int(configured_minimum) if configured_minimum is not None else HARD_MIN_RUN_INTERVAL_MINUTES
+    except (TypeError, ValueError):
+        parsed = HARD_MIN_RUN_INTERVAL_MINUTES
+    return max(HARD_MIN_RUN_INTERVAL_MINUTES, parsed)
+
+
+def resolve_request_delay_minimum(configured_minimum: int | None) -> int:
+    try:
+        parsed = int(configured_minimum) if configured_minimum is not None else HARD_MIN_REQUEST_DELAY_SECONDS
+    except (TypeError, ValueError):
+        parsed = HARD_MIN_REQUEST_DELAY_SECONDS
+    return max(HARD_MIN_REQUEST_DELAY_SECONDS, parsed)
+
+
+def parse_run_interval_minutes(value: str, *, minimum: int = HARD_MIN_RUN_INTERVAL_MINUTES) -> int:
    try:
        parsed = int(value)
    except ValueError as exc:
        raise UserSettingsServiceError("Check interval must be a whole number.") from exc
-    if parsed < 15:
-        raise UserSettingsServiceError("Check interval must be at least 15 minutes.")
+    effective_minimum = resolve_run_interval_minimum(minimum)
+    if parsed < effective_minimum:
+        raise UserSettingsServiceError(
+            f"Check interval must be at least {effective_minimum} minutes."
+        )
    return parsed


-def parse_request_delay_seconds(value: str) -> int:
+def parse_request_delay_seconds(value: str, *, minimum: int = HARD_MIN_REQUEST_DELAY_SECONDS) -> int:
    try:
        parsed = int(value)
    except ValueError as exc:
        raise UserSettingsServiceError("Request delay must be a whole number.") from exc
-    if parsed < 2:
-        raise UserSettingsServiceError("Request delay must be at least 2 seconds.")
+    effective_minimum = resolve_request_delay_minimum(minimum)
+    if parsed < effective_minimum:
+        raise UserSettingsServiceError(
+            f"Request delay must be at least {effective_minimum} seconds."
+        )
    return parsed


--- a/app/settings.py
+++ b/app/settings.py
@ -115,6 +115,22 @@ class Settings:
    log_redact_fields: str = os.getenv("LOG_REDACT_FIELDS", "")
    scheduler_enabled: bool = _env_bool("SCHEDULER_ENABLED", True)
    scheduler_tick_seconds: int = _env_int("SCHEDULER_TICK_SECONDS", 60)
+    ingestion_automation_allowed: bool = _env_bool(
+        "INGESTION_AUTOMATION_ALLOWED",
+        True,
+    )
+    ingestion_manual_run_allowed: bool = _env_bool(
+        "INGESTION_MANUAL_RUN_ALLOWED",
+        True,
+    )
+    ingestion_min_run_interval_minutes: int = _env_int(
+        "INGESTION_MIN_RUN_INTERVAL_MINUTES",
+        15,
+    )
+    ingestion_min_request_delay_seconds: int = _env_int(
+        "INGESTION_MIN_REQUEST_DELAY_SECONDS",
+        2,
+    )
    ingestion_network_error_retries: int = _env_int("INGESTION_NETWORK_ERROR_RETRIES", 1)
    ingestion_retry_backoff_seconds: float = _env_float(
        "INGESTION_RETRY_BACKOFF_SECONDS",
@ -137,6 +153,14 @@ class Settings:
        "INGESTION_ALERT_RETRY_SCHEDULED_THRESHOLD",
        3,
    )
+    ingestion_safety_cooldown_blocked_seconds: int = _env_int(
+        "INGESTION_SAFETY_COOLDOWN_BLOCKED_SECONDS",
+        1800,
+    )
+    ingestion_safety_cooldown_network_seconds: int = _env_int(
+        "INGESTION_SAFETY_COOLDOWN_NETWORK_SECONDS",
+        900,
+    )
    ingestion_continuation_queue_enabled: bool = _env_bool(
        "INGESTION_CONTINUATION_QUEUE_ENABLED",
        True,
--- a/docker-compose.yml
+++ b/docker-compose.yml
@ -40,6 +40,10 @@ services:
      LOG_REDACT_FIELDS: ${LOG_REDACT_FIELDS:-}
      SCHEDULER_ENABLED: ${SCHEDULER_ENABLED:-1}
      SCHEDULER_TICK_SECONDS: ${SCHEDULER_TICK_SECONDS:-60}
+      INGESTION_AUTOMATION_ALLOWED: ${INGESTION_AUTOMATION_ALLOWED:-1}
+      INGESTION_MANUAL_RUN_ALLOWED: ${INGESTION_MANUAL_RUN_ALLOWED:-1}
+      INGESTION_MIN_RUN_INTERVAL_MINUTES: ${INGESTION_MIN_RUN_INTERVAL_MINUTES:-15}
+      INGESTION_MIN_REQUEST_DELAY_SECONDS: ${INGESTION_MIN_REQUEST_DELAY_SECONDS:-2}
      INGESTION_NETWORK_ERROR_RETRIES: ${INGESTION_NETWORK_ERROR_RETRIES:-1}
      INGESTION_RETRY_BACKOFF_SECONDS: ${INGESTION_RETRY_BACKOFF_SECONDS:-1.0}
      INGESTION_MAX_PAGES_PER_SCHOLAR: ${INGESTION_MAX_PAGES_PER_SCHOLAR:-30}
@ -47,6 +51,8 @@ services:
      INGESTION_ALERT_BLOCKED_FAILURE_THRESHOLD: ${INGESTION_ALERT_BLOCKED_FAILURE_THRESHOLD:-1}
      INGESTION_ALERT_NETWORK_FAILURE_THRESHOLD: ${INGESTION_ALERT_NETWORK_FAILURE_THRESHOLD:-2}
      INGESTION_ALERT_RETRY_SCHEDULED_THRESHOLD: ${INGESTION_ALERT_RETRY_SCHEDULED_THRESHOLD:-3}
+      INGESTION_SAFETY_COOLDOWN_BLOCKED_SECONDS: ${INGESTION_SAFETY_COOLDOWN_BLOCKED_SECONDS:-1800}
+      INGESTION_SAFETY_COOLDOWN_NETWORK_SECONDS: ${INGESTION_SAFETY_COOLDOWN_NETWORK_SECONDS:-900}
      INGESTION_CONTINUATION_QUEUE_ENABLED: ${INGESTION_CONTINUATION_QUEUE_ENABLED:-1}
      INGESTION_CONTINUATION_BASE_DELAY_SECONDS: ${INGESTION_CONTINUATION_BASE_DELAY_SECONDS:-120}
      INGESTION_CONTINUATION_MAX_DELAY_SECONDS: ${INGESTION_CONTINUATION_MAX_DELAY_SECONDS:-3600}
--- a/docs/ops/scrape_safety_runbook.md
+++ b/docs/ops/scrape_safety_runbook.md
@ -0,0 +1,74 @@
+# Scrape Safety Runbook
+
+This runbook covers operational handling for scrape safety cooldowns, threshold alerts, and blocked-IP events.
+
+## Key Signals
+
+Use structured log `event` fields (recommended with `LOG_FORMAT=json`):
+
+- `ingestion.safety_policy_blocked_run_start`
+- `ingestion.safety_cooldown_entered`
+- `ingestion.safety_cooldown_cleared`
+- `ingestion.alert_blocked_failure_threshold_exceeded`
+- `ingestion.alert_network_failure_threshold_exceeded`
+- `ingestion.alert_retry_scheduled_threshold_exceeded`
+- `api.runs.manual_blocked_policy`
+- `api.runs.manual_blocked_safety`
+- `scheduler.run_skipped_safety_cooldown`
+- `scheduler.run_skipped_safety_cooldown_precheck`
+- `scheduler.queue_item_deferred_safety_cooldown`
+
+Each event includes metric-style fields (`metric_name`, `metric_value`) for straightforward log-based alert rules.
+
+## Recommended Alert Rules
+
+- Cooldown enters:
+  - Trigger on `event=ingestion.safety_cooldown_entered`.
+- Repeated start blocks:
+  - Trigger on high rate of `event=api.runs.manual_blocked_safety`.
+- Threshold trips:
+  - Trigger on any of:
+    - `ingestion.alert_blocked_failure_threshold_exceeded`
+    - `ingestion.alert_network_failure_threshold_exceeded`
+    - `ingestion.alert_retry_scheduled_threshold_exceeded`
+- Scheduler pressure:
+  - Trigger on sustained `scheduler.queue_item_deferred_safety_cooldown`.
+
+## If Your IP Appears Blocked
+
+Symptoms:
+
+- cooldown reason `blocked_failure_threshold_exceeded`
+- parse state `blocked_or_captcha`
+- redirects toward Google account sign-in flows
+
+Actions:
+
+1. Stop manual retries immediately.
+2. Let cooldown expire; do not spam retriggers.
+3. Increase `INGESTION_MIN_REQUEST_DELAY_SECONDS` and user request delay values.
+4. Reduce concurrency pressure (keep one scheduler instance).
+5. Keep name-search disabled/WIP for now if login-gated responses persist.
+6. Resume with a small monitored run and verify blocked rate drops.
+
+Avoid:
+
+- aggressive rapid retries
+- rotating through risky scraping patterns that increase challenge rates
+- bypass/captcha-solving workflows that may violate source platform rules
+
+## Environment Controls
+
+Policy floors and safety controls:
+
+- `INGESTION_MIN_REQUEST_DELAY_SECONDS`
+- `INGESTION_MIN_RUN_INTERVAL_MINUTES`
+- `INGESTION_ALERT_BLOCKED_FAILURE_THRESHOLD`
+- `INGESTION_ALERT_NETWORK_FAILURE_THRESHOLD`
+- `INGESTION_ALERT_RETRY_SCHEDULED_THRESHOLD`
+- `INGESTION_SAFETY_COOLDOWN_BLOCKED_SECONDS`
+- `INGESTION_SAFETY_COOLDOWN_NETWORK_SECONDS`
+- `INGESTION_MANUAL_RUN_ALLOWED`
+- `INGESTION_AUTOMATION_ALLOWED`
+
+Apply stricter values first, then relax slowly only after sustained healthy runs.
--- a/frontend/src/components/layout/AppNav.vue
+++ b/frontend/src/components/layout/AppNav.vue
@ -3,7 +3,9 @@ import { computed } from "vue";
 import { useRouter } from "vue-router";

 import RunStatusBadge from "@/components/patterns/RunStatusBadge.vue";
+import ScrapeSafetyBadge from "@/components/patterns/ScrapeSafetyBadge.vue";
 import AppButton from "@/components/ui/AppButton.vue";
+import { formatCooldownCountdown } from "@/features/safety";
 import { useAuthStore } from "@/stores/auth";
 import { useRunStatusStore } from "@/stores/run_status";
 import { useUserSettingsStore } from "@/stores/user_settings";
@ -31,21 +33,34 @@ const links = computed(() => {
    return userSettings.isPageVisible(item.id);
  });
 });
-const navRunStatus = computed(() => {
+const navSafetyText = computed(() => {
+  if (!userSettings.manualRunAllowed) {
+    return "Manual checks disabled";
+  }
+  if (runStatus.safetyState.cooldown_active) {
+    return `Cooldown: ${formatCooldownCountdown(runStatus.safetyState.cooldown_remaining_seconds)}`;
+  }
+  return "Safety ready";
+});
+const navRunStateStatus = computed(() => {
+  if (runStatus.isSubmitting && !runStatus.isLikelyRunning) {
+    return "starting";
+  }
  if (runStatus.isLikelyRunning) {
    return "running";
  }
  return "idle";
 });
-const navRunText = computed(() => {
-  const label = navRunStatus.value
-    .replaceAll("_", " ")
-    .split(" ")
+const navRunStateLabel = computed(() =>
+  navRunStateStatus.value
+    .split("_")
    .filter((segment) => segment.length > 0)
    .map((segment) => segment.charAt(0).toUpperCase() + segment.slice(1))
-    .join(" ");
-  return `State: ${label}`;
-});
+    .join(" "),
+);
+const showSafetyRow = computed(
+  () => runStatus.safetyState.cooldown_active || !userSettings.manualRunAllowed,
+);

 async function onLogout(): Promise<void> {
  await auth.logout();
@ -71,12 +86,18 @@ async function onLogout(): Promise<void> {
      </nav>

      <div class="mt-auto grid gap-2 border-t border-stroke-subtle pt-3">
-        <div
-          v-if="auth.isAdmin"
-          class="flex items-center justify-between gap-2 rounded-lg border border-stroke-default bg-surface-card-muted px-2.5 py-2 text-xs text-secondary"
-        >
-          <span class="truncate">{{ navRunText }}</span>
-          <RunStatusBadge :status="navRunStatus" />
+        <div class="grid gap-1 rounded-lg border border-stroke-default bg-surface-card-muted px-2.5 py-2 text-xs text-secondary">
+          <div class="flex items-center justify-between gap-2">
+            <span class="truncate">State: {{ navRunStateLabel }}</span>
+            <RunStatusBadge :status="navRunStateStatus" />
+          </div>
+          <div
+            v-if="showSafetyRow"
+            class="flex items-center justify-between gap-2 border-t border-stroke-subtle pt-1"
+          >
+            <span class="truncate">{{ navSafetyText }}</span>
+            <ScrapeSafetyBadge :state="runStatus.safetyState" />
+          </div>
        </div>
        <AppButton variant="ghost" class="w-full justify-start" @click="onLogout">Logout</AppButton>
      </div>
--- a/frontend/src/components/patterns/RunStatusBadge.vue
+++ b/frontend/src/components/patterns/RunStatusBadge.vue
@ -20,6 +20,9 @@ const toneClass = computed(() => {
  if (props.status === "running") {
    return "border-state-info-border bg-state-info-bg text-state-info-text";
  }
+  if (props.status === "starting") {
+    return "border-state-info-border bg-state-info-bg text-state-info-text";
+  }
  return "border-stroke-default bg-surface-card-muted text-ink-secondary";
 });
 </script>
--- a/frontend/src/components/patterns/ScrapeSafetyBadge.vue
+++ b/frontend/src/components/patterns/ScrapeSafetyBadge.vue
@ -0,0 +1,27 @@
+<script setup lang="ts">
+import { computed } from "vue";
+
+import { type ScrapeSafetyState } from "@/features/safety";
+
+const props = defineProps<{
+  state: ScrapeSafetyState;
+}>();
+
+const label = computed(() => (props.state.cooldown_active ? "Safety cooldown" : "Safety ready"));
+
+const toneClass = computed(() => {
+  if (!props.state.cooldown_active) {
+    return "border-state-success-border bg-state-success-bg text-state-success-text";
+  }
+  if (props.state.cooldown_reason === "blocked_failure_threshold_exceeded") {
+    return "border-state-danger-border bg-state-danger-bg text-state-danger-text";
+  }
+  return "border-state-warning-border bg-state-warning-bg text-state-warning-text";
+});
+</script>
+
+<template>
+  <span class="inline-flex items-center rounded-full border px-2 py-0.5 text-xs font-semibold" :class="toneClass">
+    {{ label }}
+  </span>
+</template>
--- a/frontend/src/components/patterns/ScrapeSafetyBanner.vue
+++ b/frontend/src/components/patterns/ScrapeSafetyBanner.vue
@ -0,0 +1,102 @@
+<script setup lang="ts">
+import { computed, onBeforeUnmount, onMounted, ref } from "vue";
+
+import AppAlert from "@/components/ui/AppAlert.vue";
+import {
+  formatCooldownCountdown,
+  type ScrapeSafetyState,
+} from "@/features/safety";
+
+const props = withDefaults(
+  defineProps<{
+    safetyState: ScrapeSafetyState;
+    manualRunAllowed?: boolean;
+  }>(),
+  {
+    manualRunAllowed: true,
+  },
+);
+
+const now = ref(Date.now());
+let timer: ReturnType<typeof setInterval> | null = null;
+
+const cooldownUntilMs = computed(() => {
+  if (!props.safetyState.cooldown_until) {
+    return null;
+  }
+  const parsed = new Date(props.safetyState.cooldown_until).getTime();
+  if (Number.isNaN(parsed)) {
+    return null;
+  }
+  return parsed;
+});
+
+const cooldownRemainingSeconds = computed(() => {
+  if (!props.safetyState.cooldown_active) {
+    return 0;
+  }
+
+  if (cooldownUntilMs.value !== null) {
+    return Math.max(0, Math.floor((cooldownUntilMs.value - now.value) / 1000));
+  }
+
+  return Math.max(0, props.safetyState.cooldown_remaining_seconds || 0);
+});
+
+const isCooldownBlocked = computed(() => props.safetyState.cooldown_active && cooldownRemainingSeconds.value > 0);
+const isPolicyBlocked = computed(() => !props.manualRunAllowed);
+
+const isVisible = computed(() => isPolicyBlocked.value || isCooldownBlocked.value);
+const tone = computed(() => {
+  if (isPolicyBlocked.value) {
+    return "warning" as const;
+  }
+  if (props.safetyState.cooldown_reason === "blocked_failure_threshold_exceeded") {
+    return "danger" as const;
+  }
+  return "warning" as const;
+});
+
+const title = computed(() => {
+  if (isPolicyBlocked.value) {
+    return "Manual checks disabled by server policy";
+  }
+  return props.safetyState.cooldown_reason_label || "Safety cooldown active";
+});
+
+const detailText = computed(() => {
+  if (isPolicyBlocked.value) {
+    return "This server currently disallows manual checks. Automatic checks may still run if enabled by policy.";
+  }
+  const countdown = formatCooldownCountdown(cooldownRemainingSeconds.value);
+  return `Manual and scheduled checks are paused for ${countdown}.`;
+});
+
+const actionText = computed(() => {
+  if (isPolicyBlocked.value) {
+    return "Ask an admin to enable manual checks in server environment policy.";
+  }
+  return props.safetyState.recommended_action;
+});
+
+onMounted(() => {
+  timer = setInterval(() => {
+    now.value = Date.now();
+  }, 1000);
+});
+
+onBeforeUnmount(() => {
+  if (timer !== null) {
+    clearInterval(timer);
+    timer = null;
+  }
+});
+</script>
+
+<template>
+  <AppAlert v-if="isVisible" :tone="tone">
+    <template #title>{{ title }}</template>
+    <p>{{ detailText }}</p>
+    <p v-if="actionText" class="text-secondary">{{ actionText }}</p>
+  </AppAlert>
+</template>
--- a/frontend/src/features/dashboard/index.ts
+++ b/frontend/src/features/dashboard/index.ts
@ -4,6 +4,7 @@ import {
  type PublicationMode,
 } from "@/features/publications";
 import { listQueueItems, listRuns, type RunListItem } from "@/features/runs";
+import { type ScrapeSafetyState } from "@/features/safety";

 export interface QueueHealth {
  queued: number;
@ -19,6 +20,7 @@ export interface DashboardSnapshot {
  recentRuns: RunListItem[];
  recentPublications: PublicationItem[];
  queue: QueueHealth;
+  safetyState: ScrapeSafetyState;
 }

 function countQueueStatuses(statuses: string[]): QueueHealth {
@ -38,11 +40,12 @@ function countQueueStatuses(statuses: string[]): QueueHealth {
 }

 export async function fetchDashboardSnapshot(): Promise<DashboardSnapshot> {
-  const [publications, runs, queueItems] = await Promise.all([
+  const [publications, runsPayload, queueItems] = await Promise.all([
    listPublications({ mode: "new", limit: 20 }),
    listRuns({ limit: 20 }),
    listQueueItems(200),
  ]);
+  const runs = runsPayload.runs;

  const queueHealth = countQueueStatuses(queueItems.map((item) => item.status));

@ -54,5 +57,6 @@ export async function fetchDashboardSnapshot(): Promise<DashboardSnapshot> {
    recentRuns: runs,
    recentPublications: publications.publications,
    queue: queueHealth,
+    safetyState: runsPayload.safety_state,
  };
 }
--- a/frontend/src/features/runs/index.ts
+++ b/frontend/src/features/runs/index.ts
@ -1,4 +1,5 @@
 import { apiRequest } from "@/lib/api/client";
+import { type ScrapeSafetyState } from "@/features/safety";

 export interface RunListItem {
  id: number;
@ -49,6 +50,7 @@ export interface RunDetail {
  run: RunListItem;
  summary: RunSummary;
  scholar_results: RunScholarResult[];
+  safety_state: ScrapeSafetyState;
 }

 export interface QueueItem {
@ -68,6 +70,7 @@ export interface QueueItem {

 interface RunsListData {
  runs: RunListItem[];
+  safety_state: ScrapeSafetyState;
 }

 interface QueueListData {
@ -79,7 +82,7 @@ export interface RunsListQuery {
  limit?: number;
 }

-export async function listRuns(query: RunsListQuery = {}): Promise<RunListItem[]> {
+export async function listRuns(query: RunsListQuery = {}): Promise<RunsListData> {
  const params = new URLSearchParams();
  if (query.failedOnly) {
    params.set("failed_only", "true");
@ -92,7 +95,7 @@ export async function listRuns(query: RunsListQuery = {}): Promise<RunListItem[]
  const response = await apiRequest<RunsListData>(`/runs${suffix ? `?${suffix}` : ""}`, {
    method: "GET",
  });
-  return response.data.runs;
+  return response.data;
 }

 export async function getRunDetail(runId: number): Promise<RunDetail> {
@ -118,6 +121,7 @@ export async function triggerManualRun(): Promise<{
  new_publication_count: number;
  reused_existing_run: boolean;
  idempotency_key: string | null;
+  safety_state: ScrapeSafetyState;
 }> {
  const headers: Record<string, string> = {
    "Idempotency-Key": generateIdempotencyKey(),
@ -133,6 +137,7 @@ export async function triggerManualRun(): Promise<{
    new_publication_count: number;
    reused_existing_run: boolean;
    idempotency_key: string | null;
+    safety_state: ScrapeSafetyState;
  }>("/runs/manual", {
    method: "POST",
    headers,
--- a/frontend/src/features/safety/index.ts
+++ b/frontend/src/features/safety/index.ts
@ -0,0 +1,112 @@
+export interface ScrapeSafetyCounters {
+  consecutive_blocked_runs: number;
+  consecutive_network_runs: number;
+  cooldown_entry_count: number;
+  blocked_start_count: number;
+  last_blocked_failure_count: number;
+  last_network_failure_count: number;
+  last_evaluated_run_id: number | null;
+}
+
+export interface ScrapeSafetyState {
+  cooldown_active: boolean;
+  cooldown_reason: string | null;
+  cooldown_reason_label: string | null;
+  cooldown_until: string | null;
+  cooldown_remaining_seconds: number;
+  recommended_action: string | null;
+  counters: ScrapeSafetyCounters;
+}
+
+export function createDefaultSafetyCounters(): ScrapeSafetyCounters {
+  return {
+    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,
+  };
+}
+
+export function createDefaultSafetyState(): ScrapeSafetyState {
+  return {
+    cooldown_active: false,
+    cooldown_reason: null,
+    cooldown_reason_label: null,
+    cooldown_until: null,
+    cooldown_remaining_seconds: 0,
+    recommended_action: null,
+    counters: createDefaultSafetyCounters(),
+  };
+}
+
+function parseNumber(value: unknown, fallback: number): number {
+  if (typeof value === "number" && Number.isFinite(value)) {
+    return value;
+  }
+  if (typeof value === "string" && value.trim().length > 0) {
+    const parsed = Number(value);
+    if (Number.isFinite(parsed)) {
+      return parsed;
+    }
+  }
+  return fallback;
+}
+
+function parseNullableString(value: unknown): string | null {
+  return typeof value === "string" && value.trim().length > 0 ? value : null;
+}
+
+export function normalizeSafetyState(value: unknown): ScrapeSafetyState {
+  if (!value || typeof value !== "object") {
+    return createDefaultSafetyState();
+  }
+
+  const raw = value as Record<string, unknown>;
+  const rawCounters = raw.counters;
+  const counters = rawCounters && typeof rawCounters === "object"
+    ? (rawCounters as Record<string, unknown>)
+    : {};
+
+  return {
+    cooldown_active: Boolean(raw.cooldown_active),
+    cooldown_reason: parseNullableString(raw.cooldown_reason),
+    cooldown_reason_label: parseNullableString(raw.cooldown_reason_label),
+    cooldown_until: parseNullableString(raw.cooldown_until),
+    cooldown_remaining_seconds: Math.max(0, parseNumber(raw.cooldown_remaining_seconds, 0)),
+    recommended_action: parseNullableString(raw.recommended_action),
+    counters: {
+      consecutive_blocked_runs: Math.max(0, parseNumber(counters.consecutive_blocked_runs, 0)),
+      consecutive_network_runs: Math.max(0, parseNumber(counters.consecutive_network_runs, 0)),
+      cooldown_entry_count: Math.max(0, parseNumber(counters.cooldown_entry_count, 0)),
+      blocked_start_count: Math.max(0, parseNumber(counters.blocked_start_count, 0)),
+      last_blocked_failure_count: Math.max(0, parseNumber(counters.last_blocked_failure_count, 0)),
+      last_network_failure_count: Math.max(0, parseNumber(counters.last_network_failure_count, 0)),
+      last_evaluated_run_id:
+        counters.last_evaluated_run_id === null
+          ? null
+          : Math.max(0, parseNumber(counters.last_evaluated_run_id, 0)),
+    },
+  };
+}
+
+export function formatCooldownCountdown(seconds: number): string {
+  const bounded = Number.isFinite(seconds) ? Math.max(0, Math.floor(seconds)) : 0;
+  if (bounded <= 0) {
+    return "0s";
+  }
+
+  const hours = Math.floor(bounded / 3600);
+  const minutes = Math.floor((bounded % 3600) / 60);
+  const secs = bounded % 60;
+
+  if (hours > 0) {
+    return `${hours}h ${minutes}m`;
+  }
+  if (minutes > 0) {
+    return `${minutes}m ${secs}s`;
+  }
+  return `${secs}s`;
+}
--- a/frontend/src/features/settings/index.ts
+++ b/frontend/src/features/settings/index.ts
@ -1,10 +1,31 @@
 import { apiRequest } from "@/lib/api/client";
+import { type ScrapeSafetyState } from "@/features/safety";
+
+export interface UserSettingsPolicy {
+  min_run_interval_minutes: number;
+  min_request_delay_seconds: number;
+  automation_allowed: boolean;
+  manual_run_allowed: boolean;
+  blocked_failure_threshold: number;
+  network_failure_threshold: number;
+  cooldown_blocked_seconds: number;
+  cooldown_network_seconds: number;
+}

 export interface UserSettings {
  auto_run_enabled: boolean;
  run_interval_minutes: number;
  request_delay_seconds: number;
  nav_visible_pages: string[];
+  policy: UserSettingsPolicy;
+  safety_state: ScrapeSafetyState;
+}
+
+export interface UserSettingsUpdate {
+  auto_run_enabled: boolean;
+  run_interval_minutes: number;
+  request_delay_seconds: number;
+  nav_visible_pages: string[];
 }

 export interface ChangePasswordPayload {
@ -18,7 +39,7 @@ export async function fetchSettings(): Promise<UserSettings> {
  return response.data;
 }

-export async function updateSettings(payload: UserSettings): Promise<UserSettings> {
+export async function updateSettings(payload: UserSettingsUpdate): Promise<UserSettings> {
  const response = await apiRequest<UserSettings>("/settings", {
    method: "PUT",
    body: payload,
--- a/frontend/src/pages/DashboardPage.vue
+++ b/frontend/src/pages/DashboardPage.vue
@ -1,5 +1,5 @@
 <script setup lang="ts">
-import { computed, onMounted, ref } from "vue";
+import { computed, onMounted, ref, watch } from "vue";

 import { fetchDashboardSnapshot, type DashboardSnapshot } from "@/features/dashboard";
 import { ApiRequestError } from "@/lib/api/errors";
@ -8,22 +8,51 @@ import AsyncStateGate from "@/components/patterns/AsyncStateGate.vue";
 import QueueHealthBadge from "@/components/patterns/QueueHealthBadge.vue";
 import RequestStateAlerts from "@/components/patterns/RequestStateAlerts.vue";
 import RunStatusBadge from "@/components/patterns/RunStatusBadge.vue";
+import ScrapeSafetyBanner from "@/components/patterns/ScrapeSafetyBanner.vue";
 import AppButton from "@/components/ui/AppButton.vue";
 import AppCard from "@/components/ui/AppCard.vue";
 import AppEmptyState from "@/components/ui/AppEmptyState.vue";
 import AppHelpHint from "@/components/ui/AppHelpHint.vue";
 import { useAuthStore } from "@/stores/auth";
 import { useRunStatusStore } from "@/stores/run_status";
+import { useUserSettingsStore } from "@/stores/user_settings";

 const loading = ref(true);
 const errorMessage = ref<string | null>(null);
 const errorRequestId = ref<string | null>(null);
 const successMessage = ref<string | null>(null);
 const snapshot = ref<DashboardSnapshot | null>(null);
+const refreshingAfterCompletion = ref(false);
 const auth = useAuthStore();
 const runStatus = useRunStatusStore();
+const userSettings = useUserSettingsStore();
+
+const isStartBlocked = computed(
+  () =>
+    runStatus.isRunActive ||
+    !userSettings.manualRunAllowed ||
+    runStatus.safetyState.cooldown_active,
+);
+const startCheckDisabledReason = computed(() => {
+  if (!userSettings.manualRunAllowed) {
+    return "Manual checks are disabled by server policy.";
+  }
+  if (runStatus.safetyState.cooldown_active) {
+    return runStatus.safetyState.cooldown_reason_label || "Safety cooldown is active.";
+  }
+  if (runStatus.isRunActive) {
+    return "A check is already in progress.";
+  }
+  return null;
+});

 const startCheckLabel = computed(() => {
+  if (!userSettings.manualRunAllowed) {
+    return "Manual checks disabled";
+  }
+  if (runStatus.safetyState.cooldown_active) {
+    return "Safety cooldown";
+  }
  if (runStatus.isLikelyRunning) {
    return "Check in progress";
  }
@ -32,6 +61,9 @@ const startCheckLabel = computed(() => {
  }
  return "Start check";
 });
+const isStartCheckAnimating = computed(
+  () => runStatus.isSubmitting || runStatus.isLikelyRunning,
+);

 const displayedLatestRun = computed(() => {
  const snapshotLatest = snapshot.value?.latestRun ?? null;
@ -95,6 +127,7 @@ async function loadSnapshot(): Promise<void> {
    const dashboardSnapshot = await fetchDashboardSnapshot();
    snapshot.value = dashboardSnapshot;
    runStatus.setLatestRun(dashboardSnapshot.latestRun);
+    runStatus.setSafetyState(dashboardSnapshot.safetyState);
  } catch (error) {
    snapshot.value = null;
    if (error instanceof ApiRequestError) {
@ -140,6 +173,29 @@ async function onTriggerRun(): Promise<void> {
 onMounted(() => {
  void loadSnapshot();
 });
+
+watch(
+  () => runStatus.latestRun,
+  (nextRun, previousRun) => {
+    if (refreshingAfterCompletion.value) {
+      return;
+    }
+    if (!nextRun || !previousRun) {
+      return;
+    }
+    if (nextRun.id !== previousRun.id) {
+      return;
+    }
+    if (previousRun.status !== "running" || nextRun.status === "running") {
+      return;
+    }
+
+    refreshingAfterCompletion.value = true;
+    void loadSnapshot().finally(() => {
+      refreshingAfterCompletion.value = false;
+    });
+  },
+);
 </script>

 <template>
@ -156,6 +212,10 @@ onMounted(() => {
        error-title="Dashboard request failed"
        @dismiss-success="successMessage = null"
      />
+      <ScrapeSafetyBanner
+        :safety-state="runStatus.safetyState"
+        :manual-run-allowed="userSettings.manualRunAllowed"
+      />

      <div class="h-0 min-h-0 flex-1 xl:overflow-hidden">
        <AsyncStateGate :loading="loading" :loading-lines="6" :show-empty="false">
@ -254,10 +314,20 @@ onMounted(() => {
                    <div class="flex items-center gap-2">
                      <AppButton
                        v-if="auth.isAdmin"
-                        :disabled="runStatus.isRunActive"
+                        :disabled="isStartBlocked"
+                        :title="startCheckDisabledReason || undefined"
+                        :class="isStartCheckAnimating ? 'shadow-[0_0_0_1px_var(--color-state-info-border)]' : ''"
                        @click="onTriggerRun"
                      >
-                        {{ startCheckLabel }}
+                        <span class="inline-flex items-center gap-2">
+                          <span v-if="isStartCheckAnimating" class="relative inline-flex h-2.5 w-2.5">
+                            <span
+                              class="absolute inline-flex h-full w-full animate-ping rounded-full bg-current opacity-60"
+                            />
+                            <span class="relative inline-flex h-2.5 w-2.5 rounded-full bg-current" />
+                          </span>
+                          {{ startCheckLabel }}
+                        </span>
                      </AppButton>
                      <RouterLink
                        v-if="auth.isAdmin"
--- a/frontend/src/pages/PublicationsPage.vue
+++ b/frontend/src/pages/PublicationsPage.vue
@ -28,7 +28,7 @@ type PublicationSortKey = "title" | "scholar" | "year" | "citations" | "status"
 const loading = ref(true);
 const publishingAll = ref(false);
 const publishingSelected = ref(false);
-const mode = ref<PublicationMode>("new");
+const mode = ref<PublicationMode>("all");
 const selectedScholarFilter = ref("");
 const searchQuery = ref("");
 const sortKey = ref<PublicationSortKey>("first_seen");
--- a/frontend/src/pages/RunsPage.vue
+++ b/frontend/src/pages/RunsPage.vue
@ -6,6 +6,7 @@ import AsyncStateGate from "@/components/patterns/AsyncStateGate.vue";
 import QueueHealthBadge from "@/components/patterns/QueueHealthBadge.vue";
 import RequestStateAlerts from "@/components/patterns/RequestStateAlerts.vue";
 import RunStatusBadge from "@/components/patterns/RunStatusBadge.vue";
+import ScrapeSafetyBanner from "@/components/patterns/ScrapeSafetyBanner.vue";
 import AppButton from "@/components/ui/AppButton.vue";
 import AppCard from "@/components/ui/AppCard.vue";
 import AppEmptyState from "@/components/ui/AppEmptyState.vue";
@ -22,6 +23,7 @@ import {
 } from "@/features/runs";
 import { ApiRequestError } from "@/lib/api/errors";
 import { useRunStatusStore } from "@/stores/run_status";
+import { useUserSettingsStore } from "@/stores/user_settings";

 const loading = ref(true);
 const runs = ref<RunListItem[]>([]);
@ -31,6 +33,7 @@ const errorRequestId = ref<string | null>(null);
 const successMessage = ref<string | null>(null);
 const activeQueueItemId = ref<number | null>(null);
 const runStatus = useRunStatusStore();
+const userSettings = useUserSettingsStore();

 function formatDate(value: string | null): string {
  if (!value) {
@ -59,7 +62,31 @@ function queueHealth() {

 const queueCounts = computed(() => queueHealth());
 const activeRunId = computed(() => runStatus.latestRun?.status === "running" ? runStatus.latestRun.id : null);
+const isStartBlocked = computed(
+  () =>
+    runStatus.isRunActive ||
+    !userSettings.manualRunAllowed ||
+    runStatus.safetyState.cooldown_active,
+);
+const startCheckDisabledReason = computed(() => {
+  if (!userSettings.manualRunAllowed) {
+    return "Manual checks are disabled by server policy.";
+  }
+  if (runStatus.safetyState.cooldown_active) {
+    return runStatus.safetyState.cooldown_reason_label || "Safety cooldown is active.";
+  }
+  if (runStatus.isRunActive) {
+    return "A check is already in progress.";
+  }
+  return null;
+});
 const runButtonLabel = computed(() => {
+  if (!userSettings.manualRunAllowed) {
+    return "Manual checks disabled";
+  }
+  if (runStatus.safetyState.cooldown_active) {
+    return "Safety cooldown";
+  }
  if (runStatus.isLikelyRunning) {
    return "Check in progress";
  }
@ -75,10 +102,11 @@ async function loadData(): Promise<void> {
  errorRequestId.value = null;

  try {
-    const [loadedRuns, loadedQueue] = await Promise.all([listRuns({ limit: 100 }), listQueueItems(200)]);
-    runs.value = loadedRuns;
+    const [loadedRunsPayload, loadedQueue] = await Promise.all([listRuns({ limit: 100 }), listQueueItems(200)]);
+    runs.value = loadedRunsPayload.runs;
    queueItems.value = loadedQueue;
-    runStatus.setLatestRun(loadedRuns[0] ?? null);
+    runStatus.setLatestRun(loadedRunsPayload.runs[0] ?? null);
+    runStatus.setSafetyState(loadedRunsPayload.safety_state);
  } catch (error) {
    runs.value = [];
    queueItems.value = [];
@ -174,8 +202,12 @@ onMounted(() => {
          :dropped="queueCounts.dropped"
        />
      </div>
+      <ScrapeSafetyBanner
+        :safety-state="runStatus.safetyState"
+        :manual-run-allowed="userSettings.manualRunAllowed"
+      />
      <div class="flex flex-wrap items-center gap-2">
-        <AppButton :disabled="runStatus.isRunActive" @click="onTriggerManualRun">
+        <AppButton :disabled="isStartBlocked" :title="startCheckDisabledReason || undefined" @click="onTriggerManualRun">
          {{ runButtonLabel }}
        </AppButton>
        <AppButton variant="secondary" :disabled="loading" @click="loadData">
--- a/frontend/src/pages/SettingsPage.vue
+++ b/frontend/src/pages/SettingsPage.vue
@ -14,10 +14,12 @@ import {
  changePassword,
  fetchSettings,
  type UserSettings,
+  type UserSettingsUpdate,
  updateSettings,
 } from "@/features/settings";
 import { ApiRequestError } from "@/lib/api/errors";
 import { useAuthStore } from "@/stores/auth";
+import { useRunStatusStore } from "@/stores/run_status";
 import { normalizeUserNavVisiblePages, useUserSettingsStore } from "@/stores/user_settings";

 interface NavPageOption {
@ -78,6 +80,7 @@ const NAV_PAGE_OPTIONS: NavPageOption[] = [

 const auth = useAuthStore();
 const userSettings = useUserSettingsStore();
+const runStatus = useRunStatusStore();

 const loading = ref(true);
 const saving = ref(false);
@ -98,8 +101,14 @@ const successMessage = ref<string | null>(null);
 const showIngestionModal = ref(false);
 const showPasswordModal = ref(false);
 const showNavigationModal = ref(false);
-const MIN_CHECK_INTERVAL_MINUTES = 15;
-const MIN_REQUEST_DELAY_SECONDS = 2;
+const minCheckIntervalMinutes = ref(15);
+const minRequestDelaySeconds = ref(2);
+const automationAllowed = ref(true);
+const manualRunAllowed = ref(true);
+const blockedFailureThreshold = ref(1);
+const networkFailureThreshold = ref(2);
+const cooldownBlockedSeconds = ref(1800);
+const cooldownNetworkSeconds = ref(900);

 const visibleNavOptions = computed(() =>
  NAV_PAGE_OPTIONS.filter((option) => !option.adminOnly || auth.isAdmin),
@ -111,11 +120,35 @@ const visibleNavLabels = computed(() =>
 );

 function hydrateSettings(settings: UserSettings): void {
-  autoRunEnabled.value = settings.auto_run_enabled;
+  const parsedMinRunInterval = Number(settings.policy?.min_run_interval_minutes);
+  minCheckIntervalMinutes.value = Number.isFinite(parsedMinRunInterval)
+    ? Math.max(15, parsedMinRunInterval)
+    : 15;
+  const parsedMinRequestDelay = Number(settings.policy?.min_request_delay_seconds);
+  minRequestDelaySeconds.value = Number.isFinite(parsedMinRequestDelay)
+    ? Math.max(2, parsedMinRequestDelay)
+    : 2;
+  automationAllowed.value = Boolean(settings.policy?.automation_allowed ?? true);
+  manualRunAllowed.value = Boolean(settings.policy?.manual_run_allowed ?? true);
+  blockedFailureThreshold.value = Number.isFinite(settings.policy?.blocked_failure_threshold)
+    ? Math.max(1, settings.policy.blocked_failure_threshold)
+    : 1;
+  networkFailureThreshold.value = Number.isFinite(settings.policy?.network_failure_threshold)
+    ? Math.max(1, settings.policy.network_failure_threshold)
+    : 2;
+  cooldownBlockedSeconds.value = Number.isFinite(settings.policy?.cooldown_blocked_seconds)
+    ? Math.max(60, settings.policy.cooldown_blocked_seconds)
+    : 1800;
+  cooldownNetworkSeconds.value = Number.isFinite(settings.policy?.cooldown_network_seconds)
+    ? Math.max(60, settings.policy.cooldown_network_seconds)
+    : 900;
+
+  autoRunEnabled.value = Boolean(settings.auto_run_enabled) && automationAllowed.value;
  runIntervalMinutes.value = String(settings.run_interval_minutes);
  requestDelaySeconds.value = String(settings.request_delay_seconds);
  navVisiblePages.value = normalizeUserNavVisiblePages(settings.nav_visible_pages);
  userSettings.applySettings(settings);
+  runStatus.setSafetyState(settings.safety_state);
 }

 function parseBoundedInteger(value: string, label: string, minimum: number): number {
@ -176,17 +209,17 @@ async function onSaveSettings(): Promise<void> {
  successMessage.value = null;

  try {
-    const payload: UserSettings = {
+    const payload: UserSettingsUpdate = {
      auto_run_enabled: autoRunEnabled.value,
      run_interval_minutes: parseBoundedInteger(
        runIntervalMinutes.value,
        "Check interval (minutes)",
-        MIN_CHECK_INTERVAL_MINUTES,
+        minCheckIntervalMinutes.value,
      ),
      request_delay_seconds: parseBoundedInteger(
        requestDelaySeconds.value,
        "Delay between requests (seconds)",
-        MIN_REQUEST_DELAY_SECONDS,
+        minRequestDelaySeconds.value,
      ),
      nav_visible_pages: normalizeUserNavVisiblePages(navVisiblePages.value),
    };
@ -268,29 +301,13 @@ onMounted(() => {
    <AsyncStateGate :loading="loading" :loading-lines="7" :show-empty="false">
      <section class="grid gap-4 xl:grid-cols-3">
        <AppCard class="flex h-full flex-col gap-4">
-          <div class="flex items-center gap-1">
-            <h2 class="text-lg font-semibold text-ink-primary">Automatic Checking</h2>
-            <AppHelpHint text="Controls when Scholarr runs automatic profile checks and how cautiously it scrapes." />
-          </div>
+        <div class="flex items-center gap-1">
+          <h2 class="text-lg font-semibold text-ink-primary">Automatic Checking</h2>
+          <AppHelpHint text="Controls when Scholarr runs automatic profile checks and how cautiously it scrapes." />
+        </div>
          <p class="text-sm text-secondary">
            Configure the background checker that looks for new publications on your tracked profiles.
          </p>
-          <dl class="grid gap-2 text-sm text-secondary">
-            <div class="flex items-center justify-between gap-2">
-              <dt>Automatic checks</dt>
-              <dd class="font-semibold text-ink-primary">
-                {{ autoRunEnabled ? "Enabled" : "Disabled" }}
-              </dd>
-            </div>
-            <div class="flex items-center justify-between gap-2">
-              <dt>Check interval</dt>
-              <dd class="font-semibold text-ink-primary">Every {{ runIntervalMinutes }} min</dd>
-            </div>
-            <div class="flex items-center justify-between gap-2">
-              <dt>Delay between requests</dt>
-              <dd class="font-semibold text-ink-primary">{{ requestDelaySeconds }} sec</dd>
-            </div>
-          </dl>
          <AppButton variant="secondary" class="mt-auto self-start" @click="showIngestionModal = true">
            Edit checking rules
          </AppButton>
@ -330,15 +347,23 @@ onMounted(() => {
      @close="showIngestionModal = false"
    >
      <form class="grid gap-3" @submit.prevent="onSaveSettings">
-        <AppCheckbox id="auto-run-enabled" v-model="autoRunEnabled" label="Enable automatic background checks" />
+        <AppCheckbox
+          id="auto-run-enabled"
+          v-model="autoRunEnabled"
+          :disabled="!automationAllowed"
+          label="Enable automatic background checks"
+        />
+        <p v-if="!automationAllowed" class="text-xs text-secondary">
+          Automatic checks are disabled by server safety policy.
+        </p>

        <label class="grid gap-2 text-sm font-medium text-ink-secondary">
          <span class="inline-flex items-center gap-1">
            Check interval (minutes)
            <AppHelpHint text="How often Scholarr starts a background update check." />
          </span>
-          <AppInput id="run-interval" v-model="runIntervalMinutes" type="number" :min="MIN_CHECK_INTERVAL_MINUTES" />
-          <span class="text-xs text-secondary">Minimum: {{ MIN_CHECK_INTERVAL_MINUTES }} minutes.</span>
+          <AppInput id="run-interval" v-model="runIntervalMinutes" type="number" :min="minCheckIntervalMinutes" />
+          <span class="text-xs text-secondary">Minimum: {{ minCheckIntervalMinutes }} minutes.</span>
        </label>

        <label class="grid gap-2 text-sm font-medium text-ink-secondary">
@ -350,11 +375,18 @@ onMounted(() => {
            id="request-delay"
            v-model="requestDelaySeconds"
            type="number"
-            :min="MIN_REQUEST_DELAY_SECONDS"
+            :min="minRequestDelaySeconds"
          />
-          <span class="text-xs text-secondary">Minimum: {{ MIN_REQUEST_DELAY_SECONDS }} seconds.</span>
+          <span class="text-xs text-secondary">Minimum: {{ minRequestDelaySeconds }} seconds.</span>
        </label>

+        <div class="grid gap-1 rounded-lg border border-stroke-default bg-surface-card-muted px-3 py-2 text-xs text-secondary">
+          <p class="font-medium text-ink-primary">Server-enforced scrape safety policy</p>
+          <p>Blocked failures trigger cooldown at {{ blockedFailureThreshold }} failures.</p>
+          <p>Network failures trigger cooldown at {{ networkFailureThreshold }} failures.</p>
+          <p>Blocked cooldown: {{ cooldownBlockedSeconds }}s. Network cooldown: {{ cooldownNetworkSeconds }}s.</p>
+        </div>
+
        <div class="mt-2 flex flex-wrap justify-end gap-2">
          <AppButton
            variant="ghost"
--- a/frontend/src/stores/run_status.test.ts
+++ b/frontend/src/stores/run_status.test.ts
@ -2,6 +2,7 @@ import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
 import { createPinia, setActivePinia } from "pinia";

 import { ApiRequestError } from "@/lib/api/errors";
+import { createDefaultSafetyState } from "@/features/safety";

 vi.mock("@/features/runs", () => ({
  listRuns: vi.fn(),
@ -40,6 +41,13 @@ function buildRun(overrides: Partial<{
  };
 }

+function buildRunsPayload(runs: ReturnType<typeof buildRun>[]) {
+  return {
+    runs,
+    safety_state: createDefaultSafetyState(),
+  };
+}
+
 describe("run status store", () => {
  const mockedListRuns = vi.mocked(listRuns);
  const mockedTriggerManualRun = vi.mocked(triggerManualRun);
@ -57,7 +65,7 @@ describe("run status store", () => {
  });

  it("bootstraps from latest run and exposes idle start state", async () => {
-    mockedListRuns.mockResolvedValueOnce([buildRun({ id: 11, status: "success" })]);
+    mockedListRuns.mockResolvedValueOnce(buildRunsPayload([buildRun({ id: 11, status: "success" })]));

    const store = useRunStatusStore();
    await store.bootstrap();
@ -80,8 +88,11 @@ describe("run status store", () => {
      new_publication_count: 0,
      reused_existing_run: false,
      idempotency_key: "abc",
+      safety_state: createDefaultSafetyState(),
    });
-    mockedListRuns.mockResolvedValueOnce([buildRun({ id: 25, status: "running", end_dt: null })]);
+    mockedListRuns.mockResolvedValueOnce(
+      buildRunsPayload([buildRun({ id: 25, status: "running", end_dt: null })]),
+    );

    const store = useRunStatusStore();
    const result = await store.startManualCheck();
@ -106,7 +117,9 @@ describe("run status store", () => {
        requestId: "req_123",
      }),
    );
-    mockedListRuns.mockResolvedValueOnce([buildRun({ id: 42, status: "running", end_dt: null })]);
+    mockedListRuns.mockResolvedValueOnce(
+      buildRunsPayload([buildRun({ id: 42, status: "running", end_dt: null })]),
+    );

    const store = useRunStatusStore();
    const result = await store.startManualCheck();
@ -124,8 +137,8 @@ describe("run status store", () => {
  it("polls while a run is active and stops when it completes", async () => {
    vi.useFakeTimers();
    mockedListRuns
-      .mockResolvedValueOnce([buildRun({ id: 90, status: "running", end_dt: null })])
-      .mockResolvedValueOnce([buildRun({ id: 90, status: "success" })]);
+      .mockResolvedValueOnce(buildRunsPayload([buildRun({ id: 90, status: "running", end_dt: null })]))
+      .mockResolvedValueOnce(buildRunsPayload([buildRun({ id: 90, status: "success" })]));

    const store = useRunStatusStore();
    await store.syncLatest();
@ -139,6 +152,42 @@ describe("run status store", () => {
    expect(store.isRunActive).toBe(false);
  });

+  it("stores cooldown safety state when manual start is blocked by policy cooldown", async () => {
+    mockedTriggerManualRun.mockRejectedValueOnce(
+      new ApiRequestError({
+        status: 429,
+        code: "scrape_cooldown_active",
+        message: "Scrape safety cooldown is active; run start is temporarily blocked.",
+        details: {
+          safety_state: {
+            cooldown_active: true,
+            cooldown_reason: "blocked_failure_threshold_exceeded",
+            cooldown_reason_label: "Blocked responses exceeded safety threshold",
+            cooldown_until: "2026-02-19T12:30:00Z",
+            cooldown_remaining_seconds: 600,
+            recommended_action: "Wait for cooldown to expire.",
+            counters: {
+              consecutive_blocked_runs: 1,
+              consecutive_network_runs: 0,
+              cooldown_entry_count: 1,
+              blocked_start_count: 2,
+              last_blocked_failure_count: 1,
+              last_network_failure_count: 0,
+              last_evaluated_run_id: 10,
+            },
+          },
+        },
+      }),
+    );
+
+    const store = useRunStatusStore();
+    const result = await store.startManualCheck();
+
+    expect(result.kind).toBe("error");
+    expect(store.safetyState.cooldown_active).toBe(true);
+    expect(store.canStart).toBe(false);
+  });
+
  it("switches from starting to in-progress when trigger request remains open", async () => {
    vi.useFakeTimers();
    mockedTriggerManualRun.mockImplementation(
@ -155,11 +204,12 @@ describe("run status store", () => {
              new_publication_count: 0,
              reused_existing_run: false,
              idempotency_key: "x",
+              safety_state: createDefaultSafetyState(),
            });
          }, RUN_STATUS_STARTING_PHASE_MS + 500);
        }),
    );
-    mockedListRuns.mockResolvedValueOnce([]);
+    mockedListRuns.mockResolvedValueOnce(buildRunsPayload([]));

    const store = useRunStatusStore();
    const startPromise = store.startManualCheck();
--- a/frontend/src/stores/run_status.ts
+++ b/frontend/src/stores/run_status.ts
@ -1,6 +1,11 @@
 import { defineStore } from "pinia";

 import { listRuns, triggerManualRun, type RunListItem } from "@/features/runs";
+import {
+  createDefaultSafetyState,
+  normalizeSafetyState,
+  type ScrapeSafetyState,
+} from "@/features/safety";
 import { ApiRequestError } from "@/lib/api/errors";

 export const RUN_STATUS_POLL_INTERVAL_MS = 5000;
@ -49,6 +54,19 @@ function extractRunIdFromDetails(details: unknown): number | null {
  return parseRunId(runIdCandidate);
 }

+function extractSafetyStateFromDetails(details: unknown): ScrapeSafetyState | null {
+  if (!details || typeof details !== "object") {
+    return null;
+  }
+
+  const candidate = (details as Record<string, unknown>).safety_state;
+  if (!candidate || typeof candidate !== "object") {
+    return null;
+  }
+
+  return normalizeSafetyState(candidate);
+}
+
 function buildPlaceholderRunningRun(runId: number): RunListItem {
  return {
    id: runId,
@ -66,6 +84,7 @@ function buildPlaceholderRunningRun(runId: number): RunListItem {
 export const useRunStatusStore = defineStore("runStatus", {
  state: () => ({
    latestRun: null as RunListItem | null,
+    safetyState: createDefaultSafetyState() as ScrapeSafetyState,
    isSubmitting: false,
    assumeRunningFromSubmission: false,
    isPolling: false,
@ -81,7 +100,7 @@ export const useRunStatusStore = defineStore("runStatus", {
      return state.latestRun?.status === "running" || state.assumeRunningFromSubmission;
    },
    canStart(): boolean {
-      return !this.isRunActive;
+      return !this.isRunActive && !this.safetyState.cooldown_active;
    },
  },
  actions: {
@ -113,6 +132,9 @@ export const useRunStatusStore = defineStore("runStatus", {
      }
      this.updatePolling();
    },
+    setSafetyState(value: unknown): void {
+      this.safetyState = normalizeSafetyState(value);
+    },
    startPolling(): void {
      if (pollTimer !== null) {
        this.isPolling = true;
@ -146,9 +168,10 @@ export const useRunStatusStore = defineStore("runStatus", {

      syncPromise = (async () => {
        try {
-          const runs = await listRuns({ limit: 1 });
-          const latest = runs[0] ?? null;
+          const payload = await listRuns({ limit: 1 });
+          const latest = payload.runs[0] ?? null;
          this.latestRun = latest;
+          this.safetyState = normalizeSafetyState(payload.safety_state);
          this.lastSyncAt = Date.now();
          this.lastErrorMessage = null;
          this.lastErrorRequestId = null;
@ -184,6 +207,18 @@ export const useRunStatusStore = defineStore("runStatus", {
          requestId: null,
        };
      }
+      if (this.safetyState.cooldown_active) {
+        const message =
+          this.safetyState.recommended_action ||
+          "Scrape safety cooldown is active; run start is temporarily blocked.";
+        this.lastErrorMessage = message;
+        this.lastErrorRequestId = null;
+        return {
+          kind: "error",
+          message,
+          requestId: null,
+        };
+      }

      this.isSubmitting = true;
      this.lastErrorMessage = null;
@ -193,6 +228,7 @@ export const useRunStatusStore = defineStore("runStatus", {

      try {
        const result = await triggerManualRun();
+        this.safetyState = normalizeSafetyState(result.safety_state);
        await this.syncLatest();

        if (!this.latestRun || this.latestRun.id !== result.run_id) {
@ -223,6 +259,20 @@ export const useRunStatusStore = defineStore("runStatus", {
          };
        }

+        if (error instanceof ApiRequestError && error.code === "scrape_cooldown_active") {
+          const safetyState = extractSafetyStateFromDetails(error.details);
+          if (safetyState) {
+            this.safetyState = safetyState;
+          }
+          this.lastErrorMessage = error.message;
+          this.lastErrorRequestId = error.requestId;
+          return {
+            kind: "error",
+            message: error.message,
+            requestId: error.requestId,
+          };
+        }
+
        if (error instanceof ApiRequestError) {
          this.lastErrorMessage = error.message;
          this.lastErrorRequestId = error.requestId;
@ -259,6 +309,7 @@ export const useRunStatusStore = defineStore("runStatus", {
      this.lastErrorMessage = null;
      this.lastErrorRequestId = null;
      this.lastSyncAt = null;
+      this.safetyState = createDefaultSafetyState();
    },
  },
 });
--- a/frontend/src/stores/user_settings.ts
+++ b/frontend/src/stores/user_settings.ts
@ -1,6 +1,7 @@
 import { defineStore } from "pinia";

 import { fetchSettings, type UserSettings } from "@/features/settings";
+import { createDefaultSafetyState, normalizeSafetyState, type ScrapeSafetyState } from "@/features/safety";

 export const REQUIRED_NAV_PAGES = ["dashboard", "scholars", "settings"] as const;
 export const DEFAULT_NAV_VISIBLE_PAGES = [
@ -51,6 +52,15 @@ function normalizeNavVisiblePages(value: unknown): string[] {
 export const useUserSettingsStore = defineStore("userSettings", {
  state: () => ({
    navVisiblePages: [...DEFAULT_NAV_VISIBLE_PAGES] as string[],
+    minRunIntervalMinutes: 15,
+    minRequestDelaySeconds: 2,
+    automationAllowed: true,
+    manualRunAllowed: true,
+    blockedFailureThreshold: 1,
+    networkFailureThreshold: 2,
+    cooldownBlockedSeconds: 1800,
+    cooldownNetworkSeconds: 900,
+    safetyState: createDefaultSafetyState() as ScrapeSafetyState,
  }),
  getters: {
    visiblePageSet: (state) => new Set(state.navVisiblePages),
@ -61,9 +71,39 @@ export const useUserSettingsStore = defineStore("userSettings", {
    },
    applySettings(settings: UserSettings): void {
      this.setNavVisiblePages(settings.nav_visible_pages);
+      this.minRunIntervalMinutes = Number.isFinite(settings.policy?.min_run_interval_minutes)
+        ? Math.max(15, settings.policy.min_run_interval_minutes)
+        : 15;
+      this.minRequestDelaySeconds = Number.isFinite(settings.policy?.min_request_delay_seconds)
+        ? Math.max(2, settings.policy.min_request_delay_seconds)
+        : 2;
+      this.automationAllowed = Boolean(settings.policy?.automation_allowed ?? true);
+      this.manualRunAllowed = Boolean(settings.policy?.manual_run_allowed ?? true);
+      this.blockedFailureThreshold = Number.isFinite(settings.policy?.blocked_failure_threshold)
+        ? Math.max(1, settings.policy.blocked_failure_threshold)
+        : 1;
+      this.networkFailureThreshold = Number.isFinite(settings.policy?.network_failure_threshold)
+        ? Math.max(1, settings.policy.network_failure_threshold)
+        : 1;
+      this.cooldownBlockedSeconds = Number.isFinite(settings.policy?.cooldown_blocked_seconds)
+        ? Math.max(60, settings.policy.cooldown_blocked_seconds)
+        : 1800;
+      this.cooldownNetworkSeconds = Number.isFinite(settings.policy?.cooldown_network_seconds)
+        ? Math.max(60, settings.policy.cooldown_network_seconds)
+        : 900;
+      this.safetyState = normalizeSafetyState(settings.safety_state);
    },
    reset(): void {
      this.navVisiblePages = [...DEFAULT_NAV_VISIBLE_PAGES];
+      this.minRunIntervalMinutes = 15;
+      this.minRequestDelaySeconds = 2;
+      this.automationAllowed = true;
+      this.manualRunAllowed = true;
+      this.blockedFailureThreshold = 1;
+      this.networkFailureThreshold = 2;
+      this.cooldownBlockedSeconds = 1800;
+      this.cooldownNetworkSeconds = 900;
+      this.safetyState = createDefaultSafetyState();
    },
    isPageVisible(pageId: string): boolean {
      if (REQUIRED_NAV_PAGES_SET.has(pageId)) {
--- a/tests/integration/test_api_v1.py
+++ b/tests/integration/test_api_v1.py
@ -1,5 +1,6 @@
 from __future__ import annotations

+from datetime import datetime, timedelta, timezone
 from pathlib import Path

 import pytest
@ -9,10 +10,31 @@ from sqlalchemy.ext.asyncio import AsyncSession

 from app.api.runtime_deps import get_scholar_source
 from app.main import app
+from app.services import user_settings as user_settings_service
 from app.services.scholar_source import FetchResult
 from app.settings import settings
 from tests.integration.helpers import insert_user, login_user

+REGRESSION_FIXTURE_DIR = Path("tests/fixtures/scholar/regression")
+SAFETY_STATE_KEYS = {
+    "cooldown_active",
+    "cooldown_reason",
+    "cooldown_reason_label",
+    "cooldown_until",
+    "cooldown_remaining_seconds",
+    "recommended_action",
+    "counters",
+}
+SAFETY_COUNTER_KEYS = {
+    "consecutive_blocked_runs",
+    "consecutive_network_runs",
+    "cooldown_entry_count",
+    "blocked_start_count",
+    "last_blocked_failure_count",
+    "last_network_failure_count",
+    "last_evaluated_run_id",
+}
+

 def _api_bootstrap_csrf_headers(client: TestClient) -> dict[str, str]:
    bootstrap_response = client.get("/api/v1/auth/csrf")
@ -32,6 +54,17 @@ def _api_csrf_headers(client: TestClient) -> dict[str, str]:
    return {"X-CSRF-Token": token}


+def _regression_fixture(name: str) -> str:
+    return (REGRESSION_FIXTURE_DIR / name).read_text(encoding="utf-8")
+
+
+def _assert_safety_state_contract(payload: dict[str, object]) -> None:
+    assert set(payload.keys()) == SAFETY_STATE_KEYS
+    counters = payload["counters"]
+    assert isinstance(counters, dict)
+    assert set(counters.keys()) == SAFETY_COUNTER_KEYS
+
+
@pytest.mark.integration
@pytest.mark.db
@pytest.mark.asyncio
@ -592,8 +625,35 @@ async def test_api_settings_get_and_update(db_session: AsyncSession) -> None:

    get_response = client.get("/api/v1/settings")
    assert get_response.status_code == 200
-    assert "request_delay_seconds" in get_response.json()["data"]
-    assert "nav_visible_pages" in get_response.json()["data"]
+    settings_payload = get_response.json()["data"]
+    assert "request_delay_seconds" in settings_payload
+    assert "nav_visible_pages" in settings_payload
+    assert settings_payload["policy"]["min_run_interval_minutes"] == user_settings_service.resolve_run_interval_minimum(
+        settings.ingestion_min_run_interval_minutes
+    )
+    assert settings_payload["policy"]["min_request_delay_seconds"] == user_settings_service.resolve_request_delay_minimum(
+        settings.ingestion_min_request_delay_seconds
+    )
+    assert settings_payload["policy"]["automation_allowed"] is settings.ingestion_automation_allowed
+    assert settings_payload["policy"]["manual_run_allowed"] is settings.ingestion_manual_run_allowed
+    assert settings_payload["policy"]["blocked_failure_threshold"] == max(
+        1,
+        int(settings.ingestion_alert_blocked_failure_threshold),
+    )
+    assert settings_payload["policy"]["network_failure_threshold"] == max(
+        1,
+        int(settings.ingestion_alert_network_failure_threshold),
+    )
+    assert settings_payload["policy"]["cooldown_blocked_seconds"] == max(
+        60,
+        int(settings.ingestion_safety_cooldown_blocked_seconds),
+    )
+    assert settings_payload["policy"]["cooldown_network_seconds"] == max(
+        60,
+        int(settings.ingestion_safety_cooldown_network_seconds),
+    )
+    _assert_safety_state_contract(settings_payload["safety_state"])
+    assert settings_payload["safety_state"]["cooldown_active"] is False

    update_response = client.put(
        "/api/v1/settings",
@ -617,6 +677,59 @@ async def test_api_settings_get_and_update(db_session: AsyncSession) -> None:
        "settings",
        "runs",
    ]
+    assert "policy" in updated
+    _assert_safety_state_contract(updated["safety_state"])
+
+
+@pytest.mark.integration
+@pytest.mark.db
+@pytest.mark.asyncio
+async def test_api_settings_enforce_env_minimums(db_session: AsyncSession) -> None:
+    await insert_user(
+        db_session,
+        email="api-settings-policy@example.com",
+        password="api-password",
+    )
+
+    client = TestClient(app)
+    login_user(client, email="api-settings-policy@example.com", password="api-password")
+    headers = _api_csrf_headers(client)
+
+    previous_min_interval = settings.ingestion_min_run_interval_minutes
+    previous_min_delay = settings.ingestion_min_request_delay_seconds
+    object.__setattr__(settings, "ingestion_min_run_interval_minutes", 30)
+    object.__setattr__(settings, "ingestion_min_request_delay_seconds", 8)
+    try:
+        interval_response = client.put(
+            "/api/v1/settings",
+            json={
+                "auto_run_enabled": True,
+                "run_interval_minutes": 29,
+                "request_delay_seconds": 9,
+                "nav_visible_pages": ["dashboard", "scholars", "settings"],
+            },
+            headers=headers,
+        )
+        assert interval_response.status_code == 400
+        assert interval_response.json()["error"]["code"] == "invalid_settings"
+        assert interval_response.json()["error"]["message"] == "Check interval must be at least 30 minutes."
+
+        delay_response = client.put(
+            "/api/v1/settings",
+            json={
+                "auto_run_enabled": True,
+                "run_interval_minutes": 30,
+                "request_delay_seconds": 7,
+                "nav_visible_pages": ["dashboard", "scholars", "settings"],
+            },
+            headers=headers,
+        )
+        assert delay_response.status_code == 400
+        assert delay_response.json()["error"]["code"] == "invalid_settings"
+        assert delay_response.json()["error"]["message"] == "Request delay must be at least 8 seconds."
+    finally:
+        object.__setattr__(settings, "ingestion_min_run_interval_minutes", previous_min_interval)
+        object.__setattr__(settings, "ingestion_min_request_delay_seconds", previous_min_delay)


@pytest.mark.integration
@ -643,6 +756,7 @@ async def test_api_runs_manual_and_queue_actions(db_session: AsyncSession) -> No
    assert run_payload["status"] in {"success", "partial_failure", "failed"}
    assert run_payload["reused_existing_run"] is False
    assert run_payload["idempotency_key"] == "manual-run-0001"
+    _assert_safety_state_contract(run_payload["safety_state"])
    run_id = int(run_payload["run_id"])

    stored_key = await db_session.execute(
@ -659,16 +773,19 @@ async def test_api_runs_manual_and_queue_actions(db_session: AsyncSession) -> No
    replay_payload = replay_response.json()["data"]
    assert replay_payload["run_id"] == run_payload["run_id"]
    assert replay_payload["reused_existing_run"] is True
+    _assert_safety_state_contract(replay_payload["safety_state"])

    runs_response = client.get("/api/v1/runs")
    assert runs_response.status_code == 200
    assert len(runs_response.json()["data"]["runs"]) >= 1
+    _assert_safety_state_contract(runs_response.json()["data"]["safety_state"])

    run_detail_response = client.get(f"/api/v1/runs/{run_id}")
    assert run_detail_response.status_code == 200
    detail_payload = run_detail_response.json()["data"]
    assert "summary" in detail_payload
    assert isinstance(detail_payload["scholar_results"], list)
+    _assert_safety_state_contract(detail_payload["safety_state"])

    scholar_result = await db_session.execute(
        text(
@ -750,6 +867,278 @@ async def test_api_runs_manual_and_queue_actions(db_session: AsyncSession) -> No
    assert clear_response.json()["data"]["message"] == "Queue item cleared."


+@pytest.mark.integration
+@pytest.mark.db
+@pytest.mark.asyncio
+async def test_api_manual_run_can_be_disabled_by_policy(db_session: AsyncSession) -> None:
+    await insert_user(
+        db_session,
+        email="api-runs-policy@example.com",
+        password="api-password",
+    )
+    client = TestClient(app)
+    login_user(client, email="api-runs-policy@example.com", password="api-password")
+    headers = _api_csrf_headers(client)
+
+    previous_manual_allowed = settings.ingestion_manual_run_allowed
+    object.__setattr__(settings, "ingestion_manual_run_allowed", False)
+    try:
+        response = client.post("/api/v1/runs/manual", headers=headers)
+        assert response.status_code == 403
+        payload = response.json()
+        assert payload["error"]["code"] == "manual_runs_disabled"
+        assert payload["error"]["details"]["policy"]["manual_run_allowed"] is False
+        assert payload["error"]["details"]["safety_state"]["cooldown_active"] is False
+    finally:
+        object.__setattr__(settings, "ingestion_manual_run_allowed", previous_manual_allowed)
+
+
+@pytest.mark.integration
+@pytest.mark.db
+@pytest.mark.asyncio
+async def test_api_manual_run_enforces_scrape_safety_cooldown(db_session: AsyncSession) -> None:
+    user_id = await insert_user(
+        db_session,
+        email="api-runs-safety@example.com",
+        password="api-password",
+    )
+    await db_session.execute(
+        text(
+            """
+            INSERT INTO scholar_profiles (user_id, scholar_id, display_name, is_enabled)
+            VALUES (:user_id, :scholar_id, :display_name, true)
+            """
+        ),
+        {
+            "user_id": user_id,
+            "scholar_id": "A1B2C3D4E5F6",
+            "display_name": "Safety Probe",
+        },
+    )
+    await db_session.commit()
+
+    blocked_fixture = _regression_fixture("profile_AAAAAAAAAAAA.html")
+
+    class BlockedScholarSource:
+        async def fetch_profile_page_html(
+            self,
+            scholar_id: str,
+            *,
+            cstart: int,
+            pagesize: int,
+        ) -> FetchResult:
+            _ = (scholar_id, cstart, pagesize)
+            return FetchResult(
+                requested_url="https://scholar.google.com/citations?hl=en&user=A1B2C3D4E5F6",
+                status_code=200,
+                final_url=(
+                    "https://accounts.google.com/v3/signin/identifier"
+                    "?continue=https%3A%2F%2Fscholar.google.com%2Fcitations"
+                ),
+                body=blocked_fixture,
+                error=None,
+            )
+
+        async def fetch_profile_html(self, scholar_id: str) -> FetchResult:
+            _ = scholar_id
+            return await self.fetch_profile_page_html(
+                "A1B2C3D4E5F6",
+                cstart=0,
+                pagesize=settings.ingestion_page_size,
+            )
+
+    app.dependency_overrides[get_scholar_source] = lambda: BlockedScholarSource()
+
+    previous_blocked_threshold = settings.ingestion_alert_blocked_failure_threshold
+    previous_blocked_cooldown_seconds = settings.ingestion_safety_cooldown_blocked_seconds
+    object.__setattr__(settings, "ingestion_alert_blocked_failure_threshold", 1)
+    object.__setattr__(settings, "ingestion_safety_cooldown_blocked_seconds", 600)
+
+    try:
+        client = TestClient(app)
+        login_user(client, email="api-runs-safety@example.com", password="api-password")
+        headers = _api_csrf_headers(client)
+
+        first_run_response = client.post(
+            "/api/v1/runs/manual",
+            headers={**headers, "Idempotency-Key": "safety-cooldown-run-1"},
+        )
+        assert first_run_response.status_code == 200
+
+        settings_response = client.get("/api/v1/settings")
+        assert settings_response.status_code == 200
+        safety_state = settings_response.json()["data"]["safety_state"]
+        _assert_safety_state_contract(safety_state)
+        assert safety_state["cooldown_active"] is True
+        assert safety_state["cooldown_reason"] == "blocked_failure_threshold_exceeded"
+        assert int(safety_state["cooldown_remaining_seconds"]) > 0
+        assert int(safety_state["counters"]["cooldown_entry_count"]) >= 1
+        assert int(safety_state["counters"]["last_blocked_failure_count"]) >= 1
+
+        blocked_start_response = client.post(
+            "/api/v1/runs/manual",
+            headers={**headers, "Idempotency-Key": "safety-cooldown-run-2"},
+        )
+        assert blocked_start_response.status_code == 429
+        blocked_payload = blocked_start_response.json()
+        assert blocked_payload["error"]["code"] == "scrape_cooldown_active"
+        blocked_state = blocked_payload["error"]["details"]["safety_state"]
+        _assert_safety_state_contract(blocked_state)
+        assert blocked_state["cooldown_active"] is True
+        assert blocked_state["cooldown_reason"] == "blocked_failure_threshold_exceeded"
+        assert int(blocked_state["counters"]["blocked_start_count"]) >= 1
+
+        runs_response = client.get("/api/v1/runs")
+        assert runs_response.status_code == 200
+        _assert_safety_state_contract(runs_response.json()["data"]["safety_state"])
+        assert runs_response.json()["data"]["safety_state"]["cooldown_active"] is True
+    finally:
+        object.__setattr__(settings, "ingestion_alert_blocked_failure_threshold", previous_blocked_threshold)
+        object.__setattr__(settings, "ingestion_safety_cooldown_blocked_seconds", previous_blocked_cooldown_seconds)
+        app.dependency_overrides.pop(get_scholar_source, None)
+
+
+@pytest.mark.integration
+@pytest.mark.db
+@pytest.mark.asyncio
+async def test_api_manual_run_enforces_network_failure_safety_cooldown(
+    db_session: AsyncSession,
+) -> None:
+    user_id = await insert_user(
+        db_session,
+        email="api-runs-safety-network@example.com",
+        password="api-password",
+    )
+    await db_session.execute(
+        text(
+            """
+            INSERT INTO scholar_profiles (user_id, scholar_id, display_name, is_enabled)
+            VALUES (:user_id, :scholar_id, :display_name, true)
+            """
+        ),
+        {
+            "user_id": user_id,
+            "scholar_id": "NNN111NNN111",
+            "display_name": "Network Safety Probe",
+        },
+    )
+    await db_session.commit()
+
+    class NetworkFailureScholarSource:
+        async def fetch_profile_page_html(
+            self,
+            scholar_id: str,
+            *,
+            cstart: int,
+            pagesize: int,
+        ) -> FetchResult:
+            _ = (scholar_id, cstart, pagesize)
+            return FetchResult(
+                requested_url="https://scholar.google.com/citations?hl=en&user=NNN111NNN111",
+                status_code=None,
+                final_url=None,
+                body="",
+                error="timed out",
+            )
+
+        async def fetch_profile_html(self, scholar_id: str) -> FetchResult:
+            _ = scholar_id
+            return await self.fetch_profile_page_html(
+                "NNN111NNN111",
+                cstart=0,
+                pagesize=settings.ingestion_page_size,
+            )
+
+    app.dependency_overrides[get_scholar_source] = lambda: NetworkFailureScholarSource()
+
+    previous_network_threshold = settings.ingestion_alert_network_failure_threshold
+    previous_network_cooldown_seconds = settings.ingestion_safety_cooldown_network_seconds
+    previous_network_retries = settings.ingestion_network_error_retries
+    previous_retry_backoff = settings.ingestion_retry_backoff_seconds
+    object.__setattr__(settings, "ingestion_alert_network_failure_threshold", 1)
+    object.__setattr__(settings, "ingestion_safety_cooldown_network_seconds", 600)
+    object.__setattr__(settings, "ingestion_network_error_retries", 0)
+    object.__setattr__(settings, "ingestion_retry_backoff_seconds", 0.0)
+
+    try:
+        client = TestClient(app)
+        login_user(client, email="api-runs-safety-network@example.com", password="api-password")
+        headers = _api_csrf_headers(client)
+
+        first_run_response = client.post(
+            "/api/v1/runs/manual",
+            headers={**headers, "Idempotency-Key": "safety-network-cooldown-run-1"},
+        )
+        assert first_run_response.status_code == 200
+
+        settings_response = client.get("/api/v1/settings")
+        assert settings_response.status_code == 200
+        safety_state = settings_response.json()["data"]["safety_state"]
+        _assert_safety_state_contract(safety_state)
+        assert safety_state["cooldown_active"] is True
+        assert safety_state["cooldown_reason"] == "network_failure_threshold_exceeded"
+        assert int(safety_state["cooldown_remaining_seconds"]) > 0
+        assert int(safety_state["counters"]["last_network_failure_count"]) >= 1
+
+        blocked_start_response = client.post(
+            "/api/v1/runs/manual",
+            headers={**headers, "Idempotency-Key": "safety-network-cooldown-run-2"},
+        )
+        assert blocked_start_response.status_code == 429
+        blocked_payload = blocked_start_response.json()
+        assert blocked_payload["error"]["code"] == "scrape_cooldown_active"
+        blocked_state = blocked_payload["error"]["details"]["safety_state"]
+        _assert_safety_state_contract(blocked_state)
+        assert blocked_state["cooldown_active"] is True
+        assert blocked_state["cooldown_reason"] == "network_failure_threshold_exceeded"
+        assert int(blocked_state["counters"]["blocked_start_count"]) >= 1
+    finally:
+        object.__setattr__(settings, "ingestion_alert_network_failure_threshold", previous_network_threshold)
+        object.__setattr__(settings, "ingestion_safety_cooldown_network_seconds", previous_network_cooldown_seconds)
+        object.__setattr__(settings, "ingestion_network_error_retries", previous_network_retries)
+        object.__setattr__(settings, "ingestion_retry_backoff_seconds", previous_retry_backoff)
+        app.dependency_overrides.pop(get_scholar_source, None)
+
+
+@pytest.mark.integration
+@pytest.mark.db
+@pytest.mark.asyncio
+async def test_api_settings_clears_expired_scrape_safety_cooldown(
+    db_session: AsyncSession,
+) -> None:
+    user_id = await insert_user(
+        db_session,
+        email="api-settings-safety-expired@example.com",
+        password="api-password",
+    )
+    user_settings = await user_settings_service.get_or_create_settings(
+        db_session,
+        user_id=user_id,
+    )
+    user_settings.scrape_safety_state = {"blocked_start_count": 2}
+    user_settings.scrape_cooldown_reason = "blocked_failure_threshold_exceeded"
+    user_settings.scrape_cooldown_until = datetime.now(timezone.utc) - timedelta(seconds=15)
+    await db_session.commit()
+
+    client = TestClient(app)
+    login_user(client, email="api-settings-safety-expired@example.com", password="api-password")
+
+    settings_response = client.get("/api/v1/settings")
+    assert settings_response.status_code == 200
+    settings_safety_state = settings_response.json()["data"]["safety_state"]
+    _assert_safety_state_contract(settings_safety_state)
+    assert settings_safety_state["cooldown_active"] is False
+    assert settings_safety_state["cooldown_reason"] is None
+    assert int(settings_safety_state["counters"]["blocked_start_count"]) == 2
+
+    runs_response = client.get("/api/v1/runs")
+    assert runs_response.status_code == 200
+    runs_safety_state = runs_response.json()["data"]["safety_state"]
+    _assert_safety_state_contract(runs_safety_state)
+    assert runs_safety_state["cooldown_active"] is False
+    assert runs_safety_state["cooldown_reason"] is None
+
+
@pytest.mark.integration
@pytest.mark.db
@pytest.mark.asyncio
--- a/tests/integration/test_migrations.py
+++ b/tests/integration/test_migrations.py
@ -14,7 +14,7 @@ EXPECTED_TABLES = {
 }

 EXPECTED_ENUMS = {"run_status", "run_trigger_type"}
-EXPECTED_REVISION = "20260219_0008"
+EXPECTED_REVISION = "20260219_0009"


@pytest.mark.integration
@ -199,3 +199,22 @@ async def test_user_settings_has_nav_visible_pages_column(db_session: AsyncSessi
        )
    )
    assert result.scalar_one() == 1
+
+
+@pytest.mark.integration
+@pytest.mark.db
+@pytest.mark.migrations
+@pytest.mark.asyncio
+async def test_user_settings_has_scrape_safety_columns(db_session: AsyncSession) -> None:
+    result = await db_session.execute(
+        text(
+            """
+            SELECT column_name
+            FROM information_schema.columns
+            WHERE table_name = 'user_settings'
+              AND column_name IN ('scrape_safety_state', 'scrape_cooldown_until', 'scrape_cooldown_reason')
+            """
+        )
+    )
+    columns = {row[0] for row in result}
+    assert columns == {"scrape_safety_state", "scrape_cooldown_until", "scrape_cooldown_reason"}
--- a/tests/unit/test_run_safety.py
+++ b/tests/unit/test_run_safety.py
@ -0,0 +1,107 @@
+from __future__ import annotations
+
+from datetime import datetime, timedelta, timezone
+
+from app.db.models import UserSetting
+from app.services import run_safety
+
+
+def test_apply_run_safety_outcome_triggers_blocked_cooldown() -> None:
+    settings = UserSetting(user_id=1, scrape_safety_state={})
+    now = datetime(2026, 2, 19, 20, 0, tzinfo=timezone.utc)
+
+    safety_state, reason = run_safety.apply_run_safety_outcome(
+        settings,
+        run_id=12,
+        blocked_failure_count=2,
+        network_failure_count=0,
+        blocked_failure_threshold=1,
+        network_failure_threshold=2,
+        blocked_cooldown_seconds=600,
+        network_cooldown_seconds=300,
+        now_utc=now,
+    )
+
+    assert reason == run_safety.COOLDOWN_REASON_BLOCKED_FAILURE_THRESHOLD
+    assert safety_state["cooldown_active"] is True
+    assert safety_state["cooldown_reason"] == run_safety.COOLDOWN_REASON_BLOCKED_FAILURE_THRESHOLD
+    assert safety_state["cooldown_remaining_seconds"] == 600
+    assert safety_state["counters"]["cooldown_entry_count"] == 1
+    assert safety_state["counters"]["last_blocked_failure_count"] == 2
+    assert safety_state["counters"]["last_evaluated_run_id"] == 12
+
+
+def test_clear_expired_cooldown() -> None:
+    now = datetime(2026, 2, 19, 20, 0, tzinfo=timezone.utc)
+    settings = UserSetting(
+        user_id=1,
+        scrape_safety_state={"blocked_start_count": 3},
+        scrape_cooldown_until=now - timedelta(seconds=5),
+        scrape_cooldown_reason=run_safety.COOLDOWN_REASON_NETWORK_FAILURE_THRESHOLD,
+    )
+
+    changed = run_safety.clear_expired_cooldown(settings, now_utc=now)
+    safety_state = run_safety.get_safety_state_payload(settings, now_utc=now)
+
+    assert changed is True
+    assert safety_state["cooldown_active"] is False
+    assert safety_state["cooldown_reason"] is None
+    assert safety_state["counters"]["blocked_start_count"] == 3
+
+
+def test_apply_run_safety_outcome_triggers_network_cooldown() -> None:
+    settings = UserSetting(user_id=1, scrape_safety_state={})
+    now = datetime(2026, 2, 19, 20, 0, tzinfo=timezone.utc)
+
+    safety_state, reason = run_safety.apply_run_safety_outcome(
+        settings,
+        run_id=21,
+        blocked_failure_count=0,
+        network_failure_count=3,
+        blocked_failure_threshold=2,
+        network_failure_threshold=1,
+        blocked_cooldown_seconds=900,
+        network_cooldown_seconds=300,
+        now_utc=now,
+    )
+
+    assert reason == run_safety.COOLDOWN_REASON_NETWORK_FAILURE_THRESHOLD
+    assert safety_state["cooldown_active"] is True
+    assert safety_state["cooldown_reason"] == run_safety.COOLDOWN_REASON_NETWORK_FAILURE_THRESHOLD
+    assert safety_state["cooldown_remaining_seconds"] == 300
+    assert safety_state["counters"]["cooldown_entry_count"] == 1
+    assert safety_state["counters"]["last_network_failure_count"] == 3
+    assert safety_state["counters"]["last_evaluated_run_id"] == 21
+
+
+def test_register_cooldown_blocked_start_increments_counter() -> None:
+    now = datetime(2026, 2, 19, 20, 0, tzinfo=timezone.utc)
+    settings = UserSetting(
+        user_id=1,
+        scrape_safety_state={"blocked_start_count": 1},
+        scrape_cooldown_until=now + timedelta(minutes=5),
+        scrape_cooldown_reason=run_safety.COOLDOWN_REASON_BLOCKED_FAILURE_THRESHOLD,
+    )
+
+    safety_state = run_safety.register_cooldown_blocked_start(settings, now_utc=now)
+
+    assert safety_state["cooldown_active"] is True
+    assert safety_state["cooldown_reason"] == run_safety.COOLDOWN_REASON_BLOCKED_FAILURE_THRESHOLD
+    assert safety_state["counters"]["blocked_start_count"] == 2
+
+
+def test_get_safety_event_context_contains_structured_fields() -> None:
+    now = datetime(2026, 2, 19, 20, 0, tzinfo=timezone.utc)
+    settings = UserSetting(
+        user_id=1,
+        scrape_safety_state={"cooldown_entry_count": 4},
+        scrape_cooldown_until=now + timedelta(minutes=10),
+        scrape_cooldown_reason=run_safety.COOLDOWN_REASON_BLOCKED_FAILURE_THRESHOLD,
+    )
+
+    event_context = run_safety.get_safety_event_context(settings, now_utc=now)
+
+    assert event_context["cooldown_active"] is True
+    assert event_context["cooldown_reason"] == run_safety.COOLDOWN_REASON_BLOCKED_FAILURE_THRESHOLD
+    assert int(event_context["cooldown_remaining_seconds"]) == 600
+    assert event_context["safety_counters"]["cooldown_entry_count"] == 4
--- a/tests/unit/test_user_settings.py
+++ b/tests/unit/test_user_settings.py
@ -4,10 +4,14 @@ import pytest

 from app.services.user_settings import (
    DEFAULT_NAV_VISIBLE_PAGES,
+    HARD_MIN_REQUEST_DELAY_SECONDS,
+    HARD_MIN_RUN_INTERVAL_MINUTES,
    UserSettingsServiceError,
    parse_nav_visible_pages,
    parse_request_delay_seconds,
    parse_run_interval_minutes,
+    resolve_request_delay_minimum,
+    resolve_run_interval_minimum,
 )


@ -35,6 +39,27 @@ def test_parse_request_delay_seconds_rejects_below_minimum() -> None:
        parse_request_delay_seconds("1")


+def test_parse_run_interval_minutes_rejects_below_configured_minimum() -> None:
+    with pytest.raises(
+        UserSettingsServiceError,
+        match="Check interval must be at least 30 minutes.",
+    ):
+        parse_run_interval_minutes("29", minimum=30)
+
+
+def test_parse_request_delay_seconds_rejects_below_configured_minimum() -> None:
+    with pytest.raises(
+        UserSettingsServiceError,
+        match="Request delay must be at least 8 seconds.",
+    ):
+        parse_request_delay_seconds("7", minimum=8)
+
+
+def test_resolve_minimums_keep_hard_floors() -> None:
+    assert resolve_run_interval_minimum(1) == HARD_MIN_RUN_INTERVAL_MINUTES
+    assert resolve_request_delay_minimum(0) == HARD_MIN_REQUEST_DELAY_SECONDS
+
+
 def test_parse_nav_visible_pages_accepts_valid_pages() -> None:
    parsed = parse_nav_visible_pages(
        [