feat: security version check

VERSION

@@ -0,0 +1 @@
+0.16.1

application/alembic/versions/0002_app_metadata.py

@@ -0,0 +1,37 @@
+"""0002 app_metadata — singleton key/value table for instance-wide state.
+
+Used by the startup version-check client to persist the anonymous
+instance UUID and a one-shot "notice shown" flag. Both values are tiny
+plain-text strings; this is a deliberate generic-config table rather
+than dedicated columns so future one-off settings (telemetry opt-in
+timestamps, feature-flag overrides, etc.) don't each need their own
+migration.
+
+Revision ID: 0002_app_metadata
+Revises: 0001_initial
+"""
+
+from typing import Sequence, Union
+
+from alembic import op
+
+
+revision: str = "0002_app_metadata"
+down_revision: Union[str, None] = "0001_initial"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.execute(
+        """
+        CREATE TABLE app_metadata (
+            key   TEXT PRIMARY KEY,
+            value TEXT NOT NULL
+        );
+        """
+    )
+
+
+def downgrade() -> None:
+    op.execute("DROP TABLE IF EXISTS app_metadata;")

application/celery_init.py

@@ -1,6 +1,8 @@
+import threading
+
 from celery import Celery
 from application.core.settings import settings
-from celery.signals import setup_logging, worker_process_init
+from celery.signals import setup_logging, worker_process_init, worker_ready
 
 
 def make_celery(app_name=__name__):
@@ -39,5 +41,25 @@ def _dispose_db_engine_on_fork(*args, **kwargs):
     dispose_engine()
 
 
+@worker_ready.connect
+def _run_version_check(*args, **kwargs):
+    """Kick off the anonymous version check on worker startup.
+
+    Runs in a daemon thread so a slow endpoint or bad DNS never holds
+    up the worker becoming ready for tasks. The check itself is
+    fail-silent (see ``application.updates.version_check.run_check``);
+    this handler's only job is to launch it and get out of the way.
+
+    Import is lazy so the symbol resolution never fires at module
+    import time — consistent with the ``_dispose_db_engine_on_fork``
+    pattern above.
+    """
+    try:
+        from application.updates.version_check import run_check
+    except Exception:
+        return
+    threading.Thread(target=run_check, name="version-check", daemon=True).start()
+
+
 celery = make_celery()
 celery.config_from_object("application.celeryconfig")

application/core/settings.py

@@ -149,6 +149,9 @@ class Settings(BaseSettings):
 
     FLASK_DEBUG_MODE: bool = False
     STORAGE_TYPE: str = "local"  # local or s3
+
+    # Anonymous startup version check for security issues.
+    VERSION_CHECK: bool = True
     URL_STRATEGY: str = "backend"  # backend or s3
 
     JWT_SECRET_KEY: str = ""

application/storage/db/models.py

@@ -117,6 +117,16 @@ stack_logs_table = Table(
     Column("timestamp", DateTime(timezone=True), nullable=False, server_default=func.now()),
 )
 
+# Singleton key/value table for instance-wide state (e.g. anonymous
+# instance UUID, one-shot notice flags). Added in migration
+# ``0002_app_metadata``.
+app_metadata_table = Table(
+    "app_metadata",
+    metadata,
+    Column("key", Text, primary_key=True),
+    Column("value", Text, nullable=False),
+)
+
 
 # --- Phase 2, Tier 2 --------------------------------------------------------
 

application/storage/db/repositories/app_metadata.py

@@ -0,0 +1,60 @@
+"""Repository for the ``app_metadata`` singleton key/value table.
+
+Owns the instance-wide state the version-check client needs:
+``instance_id`` (anonymous UUID sent with each check) and
+``version_check_notice_shown`` (one-shot flag for the first-run
+telemetry notice). Kept deliberately generic so future one-off config
+values can piggyback without a new migration each time.
+"""
+
+from __future__ import annotations
+
+import uuid
+from typing import Optional
+
+from sqlalchemy import Connection, text
+
+
+class AppMetadataRepository:
+    """Postgres-backed ``app_metadata`` store. Tiny by design."""
+
+    def __init__(self, conn: Connection) -> None:
+        self._conn = conn
+
+    def get(self, key: str) -> Optional[str]:
+        row = self._conn.execute(
+            text("SELECT value FROM app_metadata WHERE key = :key"),
+            {"key": key},
+        ).fetchone()
+        return row[0] if row is not None else None
+
+    def set(self, key: str, value: str) -> None:
+        self._conn.execute(
+            text(
+                "INSERT INTO app_metadata (key, value) VALUES (:key, :value) "
+                "ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value"
+            ),
+            {"key": key, "value": value},
+        )
+
+    def get_or_create_instance_id(self) -> str:
+        """Return the anonymous instance UUID, generating one if absent.
+
+        Uses ``INSERT ... ON CONFLICT DO NOTHING`` + re-read so two
+        workers racing on the very first startup converge on a single
+        UUID instead of each persisting their own.
+        """
+        existing = self.get("instance_id")
+        if existing:
+            return existing
+        candidate = str(uuid.uuid4())
+        self._conn.execute(
+            text(
+                "INSERT INTO app_metadata (key, value) VALUES ('instance_id', :value) "
+                "ON CONFLICT (key) DO NOTHING"
+            ),
+            {"value": candidate},
+        )
+        # Re-read: if another worker won the race, their UUID is now authoritative.
+        winner = self.get("instance_id")
+        return winner or candidate

application/updates/version_check.py

@@ -0,0 +1,288 @@
+"""Anonymous startup version-check client.
+
+Called once per Celery worker boot (see ``application/celery_init.py``
+``worker_ready`` handler). Posts the running version + anonymous
+instance UUID to ``gptcloud.arc53.com/api/check``, caches the response
+in Redis, and surfaces any advisories to stdout + logs.
+
+Design invariants — all enforced by a broad ``try/except`` at the top
+of :func:`run_check`:
+
+* Never blocks worker startup (fired from a daemon thread).
+* Never raises to the caller (every failure is swallowed + logged at
+  ``DEBUG``).
+* Opt-out via ``VERSION_CHECK=0`` short-circuits before any Postgres
+  write, Redis access, or outbound request.
+* Redis coordinates multi-worker and multi-replica deployments — the
+  first worker to acquire ``docsgpt:version_check:lock`` fetches, the
+  rest read from the cached response on the next cycle.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import os
+import platform
+import socket
+import sys
+from typing import Any, Dict, Optional
+
+import requests
+
+from application.cache import get_redis_instance
+from application.core.settings import settings
+from application.storage.db.repositories.app_metadata import AppMetadataRepository
+from application.storage.db.session import db_session
+from application.version import get_version
+
+logger = logging.getLogger(__name__)
+
+ENDPOINT_URL = "https://gptcloud.arc53.com/api/check"
+CLIENT_NAME = "docsgpt-backend"
+REQUEST_TIMEOUT_SECONDS = 5
+
+CACHE_KEY = "docsgpt:version_check:response"
+LOCK_KEY = "docsgpt:version_check:lock"
+CACHE_TTL_SECONDS = 6 * 3600  # 6h default; shortened by response `next_check_after`.
+LOCK_TTL_SECONDS = 60
+
+NOTICE_KEY = "version_check_notice_shown"
+INSTANCE_ID_KEY = "instance_id"
+
+_HIGH_SEVERITIES = {"high", "critical"}
+
+_ANSI_RESET = "\033[0m"
+_ANSI_RED = "\033[31m"
+_ANSI_YELLOW = "\033[33m"
+
+
+def run_check() -> None:
+    """Entry point for the worker-startup daemon thread.
+
+    Safe to call unconditionally: the opt-out, Redis-outage, and
+    Postgres-outage paths all return silently. No exception propagates.
+    """
+    try:
+        _run_check_inner()
+    except Exception as exc:  # noqa: BLE001 — belt-and-braces; nothing escapes.
+        logger.debug("version check crashed: %s", exc, exc_info=True)
+
+
+def _run_check_inner() -> None:
+    if not settings.VERSION_CHECK:
+        return
+
+    instance_id = _resolve_instance_id_and_notice()
+    if instance_id is None:
+        # Postgres unavailable — per spec we skip the check entirely
+        # rather than phone home with a synthetic/ephemeral UUID.
+        return
+
+    redis_client = get_redis_instance()
+
+    cached = _read_cache(redis_client)
+    if cached is not None:
+        _render_advisories(cached)
+        return
+
+    # Cache miss. Try to win the lock; if another worker has it, skip.
+    # ``redis_client is None`` here means Redis is unreachable — per the
+    # spec we still proceed uncached (acceptable duplicate calls in
+    # multi-worker Redis-less deploys).
+    if redis_client is not None and not _acquire_lock(redis_client):
+        return
+
+    response = _fetch(instance_id)
+    if response is None:
+        if redis_client is not None:
+            _release_lock(redis_client)
+        return
+
+    _write_cache(redis_client, response)
+    _render_advisories(response)
+    if redis_client is not None:
+        _release_lock(redis_client)
+
+
+def _resolve_instance_id_and_notice() -> Optional[str]:
+    """Load (or create) the instance UUID and emit the first-run notice.
+
+    The notice is printed at most once across the lifetime of the
+    installation — tracked via the ``version_check_notice_shown`` row
+    in ``app_metadata``. Both reads and the write happen inside one
+    short transaction so two racing workers can't each emit the notice.
+    """
+    try:
+        with db_session() as conn:
+            repo = AppMetadataRepository(conn)
+            instance_id = repo.get_or_create_instance_id()
+            if repo.get(NOTICE_KEY) is None:
+                _print_first_run_notice()
+                repo.set(NOTICE_KEY, "1")
+            return instance_id
+    except Exception as exc:  # noqa: BLE001 — Postgres down, bad URI, etc.
+        logger.debug("version check: Postgres unavailable (%s)", exc, exc_info=True)
+        return None
+
+
+def _print_first_run_notice() -> None:
+    message = (
+        "Anonymous version check enabled — sends version to "
+        "gptcloud.arc53.com.\nDisable with VERSION_CHECK=0."
+    )
+    print(message, flush=True)
+    logger.info("version check: first-run notice shown")
+
+
+def _read_cache(redis_client) -> Optional[Dict[str, Any]]:
+    if redis_client is None:
+        return None
+    try:
+        raw = redis_client.get(CACHE_KEY)
+    except Exception as exc:  # noqa: BLE001 — Redis transient errors.
+        logger.debug("version check: cache GET failed (%s)", exc, exc_info=True)
+        return None
+    if raw is None:
+        return None
+    try:
+        return json.loads(raw.decode("utf-8") if isinstance(raw, bytes) else raw)
+    except (ValueError, AttributeError) as exc:
+        logger.debug("version check: cache decode failed (%s)", exc, exc_info=True)
+        return None
+
+
+def _write_cache(redis_client, response: Dict[str, Any]) -> None:
+    if redis_client is None:
+        return
+    ttl = _compute_ttl(response)
+    try:
+        redis_client.setex(CACHE_KEY, ttl, json.dumps(response))
+    except Exception as exc:  # noqa: BLE001
+        logger.debug("version check: cache SETEX failed (%s)", exc, exc_info=True)
+
+
+def _compute_ttl(response: Dict[str, Any]) -> int:
+    """Cap the cache at 6h but honor a shorter server-specified window."""
+    next_after = response.get("next_check_after")
+    if isinstance(next_after, (int, float)) and next_after > 0:
+        return max(1, min(CACHE_TTL_SECONDS, int(next_after)))
+    return CACHE_TTL_SECONDS
+
+
+def _acquire_lock(redis_client) -> bool:
+    try:
+        owner = f"{socket.gethostname()}:{os.getpid()}"
+        return bool(
+            redis_client.set(LOCK_KEY, owner, nx=True, ex=LOCK_TTL_SECONDS)
+        )
+    except Exception as exc:  # noqa: BLE001
+        # Treat a failing Redis the same as "no lock infra" — skip rather
+        # than fire without coordination, because Redis outage is
+        # usually transient and one missed cycle is harmless.
+        logger.debug("version check: lock acquire failed (%s)", exc, exc_info=True)
+        return False
+
+
+def _release_lock(redis_client) -> None:
+    try:
+        redis_client.delete(LOCK_KEY)
+    except Exception as exc:  # noqa: BLE001
+        logger.debug("version check: lock release failed (%s)", exc, exc_info=True)
+
+
+def _fetch(instance_id: str) -> Optional[Dict[str, Any]]:
+    payload = {
+        "version": get_version(),
+        "instance_id": instance_id,
+        "python_version": platform.python_version(),
+        "platform": sys.platform,
+        "client": CLIENT_NAME,
+    }
+    try:
+        resp = requests.post(
+            ENDPOINT_URL,
+            json=payload,
+            timeout=REQUEST_TIMEOUT_SECONDS,
+        )
+    except requests.RequestException as exc:
+        logger.debug("version check: request failed (%s)", exc, exc_info=True)
+        return None
+    if resp.status_code >= 400:
+        logger.debug("version check: non-2xx response %s", resp.status_code)
+        return None
+    try:
+        return resp.json()
+    except ValueError as exc:
+        logger.debug("version check: response decode failed (%s)", exc, exc_info=True)
+        return None
+
+
+def _render_advisories(response: Dict[str, Any]) -> None:
+    advisories = response.get("advisories") or []
+    if not isinstance(advisories, list):
+        return
+    current_version = get_version()
+    for advisory in advisories:
+        if not isinstance(advisory, dict):
+            continue
+        severity = str(advisory.get("severity", "")).lower()
+        advisory_id = advisory.get("id", "UNKNOWN")
+        title = advisory.get("title", "")
+        url = advisory.get("url", "")
+        fixed_in = advisory.get("fixed_in")
+        summary = advisory.get(
+            "summary",
+            f"Your DocsGPT version {current_version} is vulnerable.",
+        )
+
+        logger.warning(
+            "security advisory %s (severity=%s) affects version %s: %s%s%s",
+            advisory_id,
+            severity or "unknown",
+            current_version,
+            title or summary,
+            f" — fixed in {fixed_in}" if fixed_in else "",
+            f" — {url}" if url else "",
+        )
+
+        if severity in _HIGH_SEVERITIES:
+            _print_console_advisory(
+                advisory_id=advisory_id,
+                title=title,
+                severity=severity,
+                summary=summary,
+                fixed_in=fixed_in,
+                url=url,
+            )
+
+
+def _print_console_advisory(
+    *,
+    advisory_id: str,
+    title: str,
+    severity: str,
+    summary: str,
+    fixed_in: Optional[str],
+    url: str,
+) -> None:
+    color = _ANSI_RED if severity == "critical" else _ANSI_YELLOW
+    bar = "=" * 60
+    upgrade_line = ""
+    if fixed_in and url:
+        upgrade_line = f"   Upgrade to {fixed_in}+ — {url}"
+    elif fixed_in:
+        upgrade_line = f"   Upgrade to {fixed_in}+"
+    elif url:
+        upgrade_line = f"   {url}"
+
+    lines = [
+        bar,
+        f"\u26a0  SECURITY ADVISORY: {advisory_id}",
+        f"   {summary}",
+        f"   {title} (severity: {severity})" if title else f"   severity: {severity}",
+    ]
+    if upgrade_line:
+        lines.append(upgrade_line)
+    lines.append(bar)
+    print(f"{color}{chr(10).join(lines)}{_ANSI_RESET}", flush=True)

application/version.py

@@ -0,0 +1,34 @@
+"""DocsGPT backend version string.
+
+Read from the top-level ``VERSION`` file so release tooling can bump the
+version without touching Python code. Cached after first read.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Optional
+
+_UNKNOWN = "unknown"
+_VERSION_PATH = Path(__file__).resolve().parent.parent / "VERSION"
+_cached: Optional[str] = None
+
+
+def get_version() -> str:
+    """Return the DocsGPT backend version, or ``"unknown"`` if absent.
+
+    A missing ``VERSION`` file is treated as a soft failure — callers
+    (notably the startup version check) must never crash because the
+    file wasn't shipped with the source checkout.
+    """
+    global _cached
+    if _cached is not None:
+        return _cached
+    try:
+        _cached = _VERSION_PATH.read_text(encoding="utf-8").strip() or _UNKNOWN
+    except (FileNotFoundError, OSError):
+        _cached = _UNKNOWN
+    return _cached
+
+
+__version__ = get_version()

tests/test_version_check.py

@@ -0,0 +1,377 @@
+"""Unit tests for the anonymous startup version-check client.
+
+All external dependencies (Postgres, Redis, HTTP) are mocked so the
+suite runs in pure-Python isolation. The focus is on the branching
+behavior described in the spec: opt-out, cache-hit, cache-miss,
+lock-denied, and the various failure paths that must never propagate.
+"""
+
+from __future__ import annotations
+
+import json
+from contextlib import contextmanager
+from unittest.mock import MagicMock, patch
+
+import pytest
+import requests
+
+from application.updates import version_check as vc_module
+
+
+class _FakeRepo:
+    """Stand-in for AppMetadataRepository backed by a plain dict."""
+
+    def __init__(self, store: dict | None = None, *, raise_on_get_instance: bool = False):
+        self._store: dict[str, str] = dict(store) if store else {}
+        self._raise = raise_on_get_instance
+
+    def get(self, key: str):
+        return self._store.get(key)
+
+    def set(self, key: str, value: str) -> None:
+        self._store[key] = value
+
+    def get_or_create_instance_id(self) -> str:
+        if self._raise:
+            raise RuntimeError("simulated Postgres outage")
+        existing = self._store.get("instance_id")
+        if existing:
+            return existing
+        self._store["instance_id"] = "11111111-2222-3333-4444-555555555555"
+        return self._store["instance_id"]
+
+
+@contextmanager
+def _fake_db_session():
+    """Stand-in for ``db_session()`` — yields ``None`` because the fake
+    repository ignores its connection argument."""
+    yield None
+
+
+def _install_repo(monkeypatch, repo: _FakeRepo):
+    """Patch the repo constructor so ``AppMetadataRepository(conn)`` → ``repo``."""
+    monkeypatch.setattr(
+        vc_module, "AppMetadataRepository", lambda conn: repo
+    )
+
+
+def _install_db_session(monkeypatch, *, raise_exc: Exception | None = None):
+    if raise_exc is not None:
+
+        @contextmanager
+        def boom():
+            raise raise_exc
+            yield  # pragma: no cover - unreachable
+
+        monkeypatch.setattr(vc_module, "db_session", boom)
+    else:
+        monkeypatch.setattr(vc_module, "db_session", _fake_db_session)
+
+
+def _make_redis_mock(*, get_return=None, set_return=True):
+    client = MagicMock()
+    client.get.return_value = get_return
+    client.set.return_value = set_return
+    client.setex.return_value = True
+    client.delete.return_value = 1
+    return client
+
+
+@pytest.fixture
+def enable_check(monkeypatch):
+    monkeypatch.setattr(vc_module.settings, "VERSION_CHECK", True)
+
+
+@pytest.mark.unit
+def test_opt_out_short_circuits(monkeypatch):
+    """VERSION_CHECK=0 → no Postgres, no Redis, no network."""
+    monkeypatch.setattr(vc_module.settings, "VERSION_CHECK", False)
+    db_spy = MagicMock()
+    redis_spy = MagicMock()
+    post_spy = MagicMock()
+    monkeypatch.setattr(vc_module, "db_session", db_spy)
+    monkeypatch.setattr(vc_module, "get_redis_instance", redis_spy)
+    monkeypatch.setattr(vc_module.requests, "post", post_spy)
+
+    vc_module.run_check()
+
+    db_spy.assert_not_called()
+    redis_spy.assert_not_called()
+    post_spy.assert_not_called()
+
+
+@pytest.mark.unit
+def test_cache_hit_renders_without_lock_or_network(monkeypatch, enable_check, capsys):
+    repo = _FakeRepo({"version_check_notice_shown": "1"})
+    _install_repo(monkeypatch, repo)
+    _install_db_session(monkeypatch)
+
+    cached = {
+        "advisories": [
+            {
+                "id": "DOCSGPT-TEST-1",
+                "title": "Example",
+                "severity": "high",
+                "fixed_in": "0.17.0",
+                "url": "https://example.test/a",
+                "summary": "Upgrade required.",
+            }
+        ]
+    }
+    redis_client = _make_redis_mock(get_return=json.dumps(cached).encode("utf-8"))
+    monkeypatch.setattr(vc_module, "get_redis_instance", lambda: redis_client)
+
+    post_spy = MagicMock()
+    monkeypatch.setattr(vc_module.requests, "post", post_spy)
+
+    vc_module.run_check()
+
+    redis_client.get.assert_called_once_with(vc_module.CACHE_KEY)
+    redis_client.set.assert_not_called()
+    redis_client.setex.assert_not_called()
+    post_spy.assert_not_called()
+    assert "SECURITY ADVISORY: DOCSGPT-TEST-1" in capsys.readouterr().out
+
+
+@pytest.mark.unit
+def test_cache_miss_lock_acquired_fetches_and_caches(monkeypatch, enable_check):
+    repo = _FakeRepo({"version_check_notice_shown": "1"})
+    _install_repo(monkeypatch, repo)
+    _install_db_session(monkeypatch)
+
+    redis_client = _make_redis_mock(get_return=None, set_return=True)
+    monkeypatch.setattr(vc_module, "get_redis_instance", lambda: redis_client)
+
+    response_body = {
+        "advisories": [
+            {
+                "id": "DOCSGPT-LOW-1",
+                "title": "Minor",
+                "severity": "low",
+                "fixed_in": "0.17.0",
+                "url": "https://example.test/low",
+            }
+        ],
+        "next_check_after": 1800,
+    }
+    post_response = MagicMock()
+    post_response.status_code = 200
+    post_response.json.return_value = response_body
+    post_spy = MagicMock(return_value=post_response)
+    monkeypatch.setattr(vc_module.requests, "post", post_spy)
+
+    vc_module.run_check()
+
+    post_spy.assert_called_once()
+    call_kwargs = post_spy.call_args
+    assert call_kwargs.args[0] == vc_module.ENDPOINT_URL
+    payload = call_kwargs.kwargs["json"]
+    assert payload["client"] == "docsgpt-backend"
+    assert payload["instance_id"] == "11111111-2222-3333-4444-555555555555"
+    assert "version" in payload and "python_version" in payload
+
+    # Lock acquired with NX EX, cache written with server-specified TTL,
+    # lock released.
+    redis_client.set.assert_called_once()
+    set_kwargs = redis_client.set.call_args.kwargs
+    assert set_kwargs == {"nx": True, "ex": vc_module.LOCK_TTL_SECONDS}
+    redis_client.setex.assert_called_once()
+    setex_args = redis_client.setex.call_args.args
+    assert setex_args[0] == vc_module.CACHE_KEY
+    assert setex_args[1] == 1800  # server override under 6h
+    redis_client.delete.assert_called_once_with(vc_module.LOCK_KEY)
+
+
+@pytest.mark.unit
+def test_cache_miss_lock_denied_skips_silently(monkeypatch, enable_check):
+    repo = _FakeRepo({"version_check_notice_shown": "1"})
+    _install_repo(monkeypatch, repo)
+    _install_db_session(monkeypatch)
+
+    redis_client = _make_redis_mock(get_return=None, set_return=False)  # lock not acquired
+    monkeypatch.setattr(vc_module, "get_redis_instance", lambda: redis_client)
+
+    post_spy = MagicMock()
+    monkeypatch.setattr(vc_module.requests, "post", post_spy)
+
+    vc_module.run_check()
+
+    post_spy.assert_not_called()
+    redis_client.setex.assert_not_called()
+    redis_client.delete.assert_not_called()
+
+
+@pytest.mark.unit
+def test_instance_id_persisted_across_runs(monkeypatch, enable_check):
+    repo = _FakeRepo({"version_check_notice_shown": "1"})
+    _install_repo(monkeypatch, repo)
+    _install_db_session(monkeypatch)
+
+    redis_client = _make_redis_mock(get_return=None, set_return=True)
+    monkeypatch.setattr(vc_module, "get_redis_instance", lambda: redis_client)
+
+    post_response = MagicMock()
+    post_response.status_code = 200
+    post_response.json.return_value = {}
+    monkeypatch.setattr(
+        vc_module.requests, "post", MagicMock(return_value=post_response)
+    )
+
+    vc_module.run_check()
+    first_id = repo.get("instance_id")
+    vc_module.run_check()
+    second_id = repo.get("instance_id")
+
+    assert first_id is not None
+    assert first_id == second_id
+
+
+@pytest.mark.unit
+def test_first_run_notice_emitted_once(monkeypatch, enable_check, capsys):
+    repo = _FakeRepo()  # empty — notice not shown yet
+    _install_repo(monkeypatch, repo)
+    _install_db_session(monkeypatch)
+
+    # Cache hit so we don't need to mock HTTP. Notice logic runs before cache.
+    redis_client = _make_redis_mock(get_return=json.dumps({}).encode("utf-8"))
+    monkeypatch.setattr(vc_module, "get_redis_instance", lambda: redis_client)
+
+    vc_module.run_check()
+    first_out = capsys.readouterr().out
+    assert "Anonymous version check enabled" in first_out
+    assert repo.get("version_check_notice_shown") == "1"
+
+    vc_module.run_check()
+    second_out = capsys.readouterr().out
+    assert "Anonymous version check enabled" not in second_out
+
+
+@pytest.mark.unit
+def test_postgres_unavailable_skips_silently(monkeypatch, enable_check):
+    _install_db_session(monkeypatch, raise_exc=RuntimeError("db down"))
+    redis_spy = MagicMock()
+    post_spy = MagicMock()
+    monkeypatch.setattr(vc_module, "get_redis_instance", redis_spy)
+    monkeypatch.setattr(vc_module.requests, "post", post_spy)
+
+    vc_module.run_check()
+
+    redis_spy.assert_not_called()
+    post_spy.assert_not_called()
+
+
+@pytest.mark.unit
+def test_postgres_repo_raises_skips_silently(monkeypatch, enable_check):
+    repo = _FakeRepo(raise_on_get_instance=True)
+    _install_repo(monkeypatch, repo)
+    _install_db_session(monkeypatch)
+    redis_spy = MagicMock()
+    post_spy = MagicMock()
+    monkeypatch.setattr(vc_module, "get_redis_instance", redis_spy)
+    monkeypatch.setattr(vc_module.requests, "post", post_spy)
+
+    vc_module.run_check()
+
+    redis_spy.assert_not_called()
+    post_spy.assert_not_called()
+
+
+@pytest.mark.unit
+def test_redis_unavailable_proceeds_uncached(monkeypatch, enable_check):
+    """``get_redis_instance()`` → None should not abort the check."""
+    repo = _FakeRepo({"version_check_notice_shown": "1"})
+    _install_repo(monkeypatch, repo)
+    _install_db_session(monkeypatch)
+    monkeypatch.setattr(vc_module, "get_redis_instance", lambda: None)
+
+    post_response = MagicMock()
+    post_response.status_code = 200
+    post_response.json.return_value = {"advisories": []}
+    post_spy = MagicMock(return_value=post_response)
+    monkeypatch.setattr(vc_module.requests, "post", post_spy)
+
+    vc_module.run_check()
+
+    post_spy.assert_called_once()
+
+
+@pytest.mark.unit
+def test_http_5xx_swallowed(monkeypatch, enable_check):
+    repo = _FakeRepo({"version_check_notice_shown": "1"})
+    _install_repo(monkeypatch, repo)
+    _install_db_session(monkeypatch)
+    redis_client = _make_redis_mock(get_return=None, set_return=True)
+    monkeypatch.setattr(vc_module, "get_redis_instance", lambda: redis_client)
+
+    post_response = MagicMock()
+    post_response.status_code = 503
+    post_response.json.return_value = {}
+    monkeypatch.setattr(
+        vc_module.requests, "post", MagicMock(return_value=post_response)
+    )
+
+    vc_module.run_check()
+
+    redis_client.setex.assert_not_called()
+    # Lock still released so the next cycle can retry.
+    redis_client.delete.assert_called_once_with(vc_module.LOCK_KEY)
+
+
+@pytest.mark.unit
+def test_http_timeout_swallowed(monkeypatch, enable_check):
+    repo = _FakeRepo({"version_check_notice_shown": "1"})
+    _install_repo(monkeypatch, repo)
+    _install_db_session(monkeypatch)
+    redis_client = _make_redis_mock(get_return=None, set_return=True)
+    monkeypatch.setattr(vc_module, "get_redis_instance", lambda: redis_client)
+    monkeypatch.setattr(
+        vc_module.requests,
+        "post",
+        MagicMock(side_effect=requests.Timeout("boom")),
+    )
+
+    # Must not raise.
+    vc_module.run_check()
+
+    redis_client.setex.assert_not_called()
+    redis_client.delete.assert_called_once_with(vc_module.LOCK_KEY)
+
+
+@pytest.mark.unit
+def test_compute_ttl_honors_server_override():
+    assert vc_module._compute_ttl({"next_check_after": 300}) == 300
+    assert vc_module._compute_ttl({"next_check_after": 60000}) == vc_module.CACHE_TTL_SECONDS
+    assert vc_module._compute_ttl({}) == vc_module.CACHE_TTL_SECONDS
+    assert vc_module._compute_ttl({"next_check_after": "bad"}) == vc_module.CACHE_TTL_SECONDS
+    # Zero/negative overrides fall back to the 6h default.
+    assert vc_module._compute_ttl({"next_check_after": 0}) == vc_module.CACHE_TTL_SECONDS
+
+
+@pytest.mark.unit
+def test_render_advisories_logs_warning_and_prints_banner(monkeypatch, capsys):
+    with patch.object(vc_module, "logger") as mock_logger:
+        vc_module._render_advisories(
+            {
+                "advisories": [
+                    {
+                        "id": "DOCSGPT-2025-001",
+                        "title": "SSRF",
+                        "severity": "critical",
+                        "fixed_in": "0.17.0",
+                        "url": "https://example.test/a",
+                        "summary": "Your DocsGPT is vulnerable.",
+                    },
+                    {
+                        "id": "DOCSGPT-2025-002",
+                        "title": "Low-sev",
+                        "severity": "low",
+                    },
+                ]
+            }
+        )
+    # Both advisories logged as warnings.
+    assert mock_logger.warning.call_count == 2
+    out = capsys.readouterr().out
+    # Only the high/critical one gets the console banner.
+    assert "DOCSGPT-2025-001" in out
+    assert "DOCSGPT-2025-002" not in out