deploy: backend from 0d25f4f

main.py

@@ -60,7 +60,7 @@ except Exception:
 
 # STARTUP VALIDATION - Run before anything else to prevent restart loops
 try:
-    from startup_validation import run_all_validations
+    from startup_validation import run_all_validations, validate_deepseek_auth
     run_all_validations()  # Exits with error if any critical check fails
 except ImportError as e:
     # If startup_validation module is not found, log warning but continue
@@ -565,8 +565,17 @@ async def app_lifespan(_app: FastAPI) -> AsyncIterator[None]:
         except Exception as exc:
             logger.error("RAG vectorstore warm-up failed: %s", exc)
 
+    async def _warmup_deepseek_auth() -> None:
+        auth_state = await asyncio.to_thread(validate_deepseek_auth, 5.0)
+        logger.warning(
+            "DeepSeek startup auth state: status=%s key=%s",
+            auth_state["status"],
+            auth_state["key_suffix"],
+        )
+
     _warmup_inference_task = asyncio.create_task(_warmup_inference_client())
     _warmup_vectorstore_task = asyncio.create_task(_warmup_vectorstore())
+    _warmup_deepseek_auth_task = asyncio.create_task(_warmup_deepseek_auth())
 
     # FIX(502): Set a readiness flag so /health reports the true state without
     # triggering heavy init on every health-check ping from HF Spaces' proxy.
@@ -2325,12 +2334,27 @@ class TestingResetResponse(BaseModel):
 # to decide whether to route traffic; a slow response or crash here causes 502.
 @app.get("/health")
 async def health_check():
+    # Cached DeepSeek auth state is read dynamically (never imported by name):
+    # the startup probe rebinds the holder, and this handler must see the
+    # current object. Missing module or holder degrades to "unchecked".
+    try:
+        import startup_validation as _startup_validation_module
+        cached_auth = (
+            getattr(_startup_validation_module, "cached_deepseek_auth_state", None) or {}
+        )
+    except Exception:
+        cached_auth = {}
     return {
         "status": "ready" if _backend_ready else "starting",
         "space": "mathpulse-ai",
         "firebase": _firebase_ready,
         "chat_model": CHAT_MODEL,
         "risk_model": RISK_MODEL,
+        "deepseek": {
+            "status": cached_auth.get("status", "unchecked"),
+            "key_suffix": cached_auth.get("key_suffix", "****"),
+            "checked_at": cached_auth.get("checked_at"),
+        },
     }
 
 

startup_validation.py

@@ -11,7 +11,12 @@ that's visible in HF Space logs.
 import os
 import sys
 import logging
+from concurrent.futures import ThreadPoolExecutor, TimeoutError as FutureTimeoutError
+from datetime import datetime, timezone
 from pathlib import Path
+from typing import Literal, Optional, TypedDict
+
+from openai import APIConnectionError, APIError, APITimeoutError
 
 try:
     from dotenv import load_dotenv
@@ -28,12 +33,104 @@ except Exception:
 
 logger = logging.getLogger("mathpulse.startup")
 
+AuthProbeStatus = Literal["ok", "auth_failed", "connection_failed", "unchecked"]
+
+
+class DeepSeekAuthState(TypedDict):
+    status: AuthProbeStatus
+    key_suffix: str
+    checked_at: Optional[str]
+
+
+cached_deepseek_auth_state: DeepSeekAuthState = {
+    "status": "unchecked",
+    "key_suffix": "****",
+    "checked_at": None,
+}
+
 
 class StartupError(Exception):
     """Critical error during startup validation."""
     pass
 
 
+def compute_key_fingerprint(api_key: Optional[str] = None) -> str:
+    configured_key = api_key if api_key is not None else os.getenv("DEEPSEEK_API_KEY", "")
+    return f"****{configured_key[-4:]}" if configured_key else "****"
+
+
+def _record_deepseek_auth_state(status: AuthProbeStatus, key_suffix: str) -> DeepSeekAuthState:
+    global cached_deepseek_auth_state
+    cached_deepseek_auth_state = {
+        "status": status,
+        "key_suffix": key_suffix,
+        "checked_at": datetime.now(timezone.utc).isoformat(),
+    }
+    return cached_deepseek_auth_state
+
+
+def validate_deepseek_auth(timeout_s: float = 5.0) -> DeepSeekAuthState:
+    api_key = os.getenv("DEEPSEEK_API_KEY", "")
+    key_suffix = compute_key_fingerprint(api_key)
+    if not api_key:
+        logger.warning("DeepSeek startup auth probe: status=unchecked key=%s", key_suffix)
+        return _record_deepseek_auth_state("unchecked", key_suffix)
+
+    try:
+        from services import inference_client
+
+        executor = ThreadPoolExecutor(max_workers=1)
+        probe_future = executor.submit(
+            lambda: inference_client.get_deepseek_client().models.list()
+        )
+        try:
+            probe_future.result(timeout=max(timeout_s, 0.0))
+        finally:
+            executor.shutdown(wait=False, cancel_futures=True)
+    except APIError as exc:
+        if getattr(exc, "status_code", None) in (401, 403):
+            logger.warning(
+                "DeepSeek startup auth probe: status=auth_failed key=%s",
+                key_suffix,
+            )
+            return _record_deepseek_auth_state("auth_failed", key_suffix)
+        logger.warning(
+            "DeepSeek startup auth probe: status=connection_failed key=%s",
+            key_suffix,
+        )
+        return _record_deepseek_auth_state("connection_failed", key_suffix)
+    except (APIConnectionError, APITimeoutError, FutureTimeoutError, TimeoutError, OSError):
+        logger.warning(
+            "DeepSeek startup auth probe: status=connection_failed key=%s",
+            key_suffix,
+        )
+        return _record_deepseek_auth_state("connection_failed", key_suffix)
+    except ValueError:
+        logger.warning("DeepSeek startup auth probe: status=unchecked key=%s", key_suffix)
+        return _record_deepseek_auth_state("unchecked", key_suffix)
+    except Exception as exc:
+        if getattr(exc, "status_code", None) in (401, 403) or type(exc).__name__ == "InferenceAuthError":
+            logger.warning(
+                "DeepSeek startup auth probe: status=auth_failed key=%s",
+                key_suffix,
+            )
+            return _record_deepseek_auth_state("auth_failed", key_suffix)
+        if type(exc).__name__ == "InferenceConnectionError":
+            logger.warning(
+                "DeepSeek startup auth probe: status=connection_failed key=%s",
+                key_suffix,
+            )
+            return _record_deepseek_auth_state("connection_failed", key_suffix)
+        logger.warning(
+            "DeepSeek startup auth probe: status=connection_failed key=%s",
+            key_suffix,
+        )
+        return _record_deepseek_auth_state("connection_failed", key_suffix)
+
+    logger.info("DeepSeek startup auth probe: status=ok key=%s", key_suffix)
+    return _record_deepseek_auth_state("ok", key_suffix)
+
+
 def validate_imports() -> None:
     """Verify all critical imports work. Use absolute imports."""
     logger.info("🔍 Validating Python imports...")

tests/test_startup_auth_visibility.py

@@ -0,0 +1,170 @@
+"""Tests for DeepSeek startup authentication validation and health visibility.
+
+These tests define the contract for startup credential validation and health visibility
+before production implementation exists (RED state).
+
+Contract under test:
+(a) validate_deepseek_auth(timeout_s=5.0) when DeepSeek returns HTTP 401:
+    - returns a status representation with status == 'auth_failed'
+    - logs/captures key suffix in '****last4' format
+    - ensures the full fake key string never appears in logs or output
+(b) GET /health when injected with cached auth state:
+    - returns additive dictionary under 'deepseek' key: {status, key_suffix, checked_at}
+    - contains only masked key suffix (e.g. '****1234')
+    - ensures full fake key string never appears in response body
+(c) validate_deepseek_auth(timeout_s=5.0) when DeepSeek returns HTTP 200 / successful probe:
+    - returns status == 'ok'
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+from unittest.mock import MagicMock, patch
+
+import httpx
+import openai
+import pytest
+from fastapi.testclient import TestClient
+
+# Ensure test env vars match existing test harness conventions
+os.environ["DEEPSEEK_API_KEY"] = "mock-key-for-testing"
+os.environ["INFERENCE_MAX_RETRIES"] = "1"
+os.environ["INFERENCE_BACKGROUND_MAX_RETRIES"] = "1"
+os.environ["INFERENCE_INTERACTIVE_MAX_RETRIES"] = "1"
+os.environ["INFERENCE_BACKOFF_SEC"] = "0"
+os.environ["INFERENCE_BACKGROUND_BACKOFF_SEC"] = "0"
+
+import main as main_module
+from main import app
+
+# Test client configured following existing backend test patterns
+client = TestClient(app, headers={"Authorization": "Bearer mock_token_teacher_T"})
+
+FAKE_DEEPSEEK_KEY = "sk-fake-test-key-1234"
+EXPECTED_MASKED_SUFFIX = "****1234"
+
+
+@pytest.fixture(autouse=True)
+def _setup_common_mocks():
+    """Bypass Firebase token verification, time.sleep backoff, and curriculum retrieval."""
+    with patch.object(
+        main_module.firebase_auth,
+        "verify_id_token",
+        return_value={"uid": "teacher_T", "role": "teacher"},
+    ), patch(
+        "routes.rag_routes.retrieve_lesson_pdf_context",
+        return_value=([], "mock_file"),
+    ), patch(
+        "time.sleep",
+        return_value=None,
+    ):
+        yield
+
+
+class TestStartupAuthVisibility:
+    """Contract tests for DeepSeek auth-visibility at startup and /health."""
+
+    def test_mocked_401_startup_auth_check_yields_auth_failed_and_redacts_key(
+        self, caplog: pytest.LogCaptureFixture
+    ):
+        """(a) Mocked-401 startup auth check yields 'auth_failed' with masked suffix and no raw key."""
+        from startup_validation import (
+            compute_key_fingerprint,
+            validate_deepseek_auth,
+        )
+
+        req = httpx.Request("POST", "https://api.deepseek.com")
+        api_error_401 = openai.APIError(
+            f"Authentication failed for key {EXPECTED_MASKED_SUFFIX}",
+            request=req,
+            body={"error": {"message": f"Invalid API key {EXPECTED_MASKED_SUFFIX}"}},
+        )
+        api_error_401.status_code = 401
+
+        mock_client = MagicMock()
+        mock_client.models.list.side_effect = api_error_401
+        mock_client.chat.completions.create.side_effect = api_error_401
+
+        fingerprint = compute_key_fingerprint(FAKE_DEEPSEEK_KEY)
+        assert EXPECTED_MASKED_SUFFIX in fingerprint
+        assert FAKE_DEEPSEEK_KEY not in fingerprint
+
+        caplog.clear()
+        with caplog.at_level(logging.DEBUG), patch.dict(
+            os.environ, {"DEEPSEEK_API_KEY": FAKE_DEEPSEEK_KEY}
+        ), patch(
+            "services.inference_client.get_deepseek_client",
+            return_value=mock_client,
+        ):
+            auth_status = validate_deepseek_auth(timeout_s=5.0)
+
+        status_value = (
+            auth_status.get("status")
+            if isinstance(auth_status, dict)
+            else getattr(auth_status, "status", None)
+        )
+        assert status_value == "auth_failed"
+
+        captured_logs = caplog.text
+        assert EXPECTED_MASKED_SUFFIX in captured_logs
+        assert FAKE_DEEPSEEK_KEY not in captured_logs
+
+    def test_health_endpoint_returns_additive_cached_deepseek_auth_state(self):
+        """(b) GET /health returns additive deepseek: {status, key_suffix, checked_at} with masked key."""
+        import startup_validation
+
+        cached_state_holder = getattr(
+            startup_validation, "cached_deepseek_auth_state"
+        )
+
+        injected_state = {
+            "status": "auth_failed",
+            "key_suffix": EXPECTED_MASKED_SUFFIX,
+            "checked_at": "2026-09-22T00:00:00Z",
+        }
+
+        with patch.object(
+            startup_validation,
+            "cached_deepseek_auth_state",
+            injected_state,
+        ), patch.dict(
+            os.environ, {"DEEPSEEK_API_KEY": FAKE_DEEPSEEK_KEY}
+        ):
+            response = client.get("/health")
+
+        assert response.status_code == 200
+        response_body = response.json()
+
+        assert "deepseek" in response_body
+        deepseek_info = response_body["deepseek"]
+        assert deepseek_info.get("status") == "auth_failed"
+        assert deepseek_info.get("key_suffix") == EXPECTED_MASKED_SUFFIX
+        assert "checked_at" in deepseek_info
+
+        raw_response_text = response.text
+        assert EXPECTED_MASKED_SUFFIX in raw_response_text
+        assert FAKE_DEEPSEEK_KEY not in raw_response_text
+
+    def test_mocked_ok_startup_auth_check_yields_ok_status(self):
+        """(c) Mocked-OK startup auth check yields status 'ok'."""
+        from startup_validation import validate_deepseek_auth
+
+        mock_client = MagicMock()
+        mock_client.models.list.return_value = MagicMock()
+        mock_client.chat.completions.create.return_value = MagicMock()
+
+        with patch.dict(
+            os.environ, {"DEEPSEEK_API_KEY": FAKE_DEEPSEEK_KEY}
+        ), patch(
+            "services.inference_client.get_deepseek_client",
+            return_value=mock_client,
+        ):
+            auth_status = validate_deepseek_auth(timeout_s=5.0)
+
+        status_value = (
+            auth_status.get("status")
+            if isinstance(auth_status, dict)
+            else getattr(auth_status, "status", None)
+        )
+        assert status_value == "ok"