def load_model(repo_id: str) -> SentenceTransformerType:
    """Load (and cache on disk) a SentenceTransformer for the given repo id.

    The repo_id is validated against the canonical HuggingFace format
    before being passed to ``snapshot_download``/``SentenceTransformer``;
    this prevents path-traversal style inputs and local filesystem paths.
    ``trust_remote_code`` is forced off so a malicious model card cannot
    execute arbitrary code on load.
    """
    _validate_repo_id(repo_id)
    CACHE_DIR.mkdir(parents=True, exist_ok=True)

    snapshot = _load_snapshot_download()
    st_cls = _load_sentence_transformer_cls()

    try:
        model_path = snapshot(
            repo_id=repo_id,
            cache_dir=CACHE_DIR,
            local_files_only=False,  # auto-download first time
            etag_timeout=_ETAG_TIMEOUT,  # HTTP HEAD timeout
        )
    except Exception as exc:
        # Try local-only as fallback (model may already be cached)
        _logger.warning(
            "Remote download failed for %s, trying local cache: %s", repo_id, exc
        )
        model_path = snapshot(
            repo_id=repo_id,
            cache_dir=CACHE_DIR,
            local_files_only=True,
        )

    # Use PyTorch backend by default (most compatible). Force
    # trust_remote_code=False so a model's config.json cannot trigger
    # execution of arbitrary downloaded Python on load.
    model = st_cls(
        model_path,
        tokenizer_kwargs={"padding": True, "truncation": True, "max_length": 512},
        backend="torch",
        trust_remote_code=False,
    )

    return model

def get_embedder(model_id: str | None = None) -> SentenceTransformerType:
    """Return a cached embedder for the requested model (defaults to config value)."""
    repo_id = model_id or DEFAULT_MODEL
    with _model_lock:
        model = _loaded_models.get(repo_id)
        if model is None:
            model = load_model(repo_id)
            _loaded_models[repo_id] = model
    return model

def embed_texts(
    texts: list[str], *, model_id: str | None = None, batch_size: int | None = None
) -> list[list[float]]:
    """
    Embed a list of texts using the default model.

    Args:
        texts: List of strings to embed.
        model_id: Optional repo id / alias override.
        batch_size: Optional override for encode batch size.

    Returns:
        List of embedding vectors (list of floats).
    """
    if not texts:
        return []

    if len(texts) > _MAX_ENCODE_BATCH:
        raise ValueError(
            f"Batch too large ({len(texts)} texts). "
            f"Maximum is {_MAX_ENCODE_BATCH}. Split into smaller batches."
        )

    model = get_embedder(model_id)
    effective_batch_size = batch_size or config.EMBEDDING_BATCH_SIZE
    embeddings = model.encode(
        texts,
        normalize_embeddings=True,
        batch_size=effective_batch_size,
        show_progress_bar=False,
    )
    return embeddings.tolist()

@app.get("/health")
async def health_check():
    """Health check endpoint."""
    return {"status": "ok"}

def authenticate_request(
    credentials: HTTPAuthorizationCredentials | None = Security(auth_scheme),
    api_key_header: str | None = Header(default=None, alias="X-API-Key"),
) -> str:
    """Validate API key if auth is enabled; otherwise return anonymous identity."""
    allowed_keys = config.EMBEDDING_SERVER_API_KEYS
    if not allowed_keys:
        return "anonymous"

    token = api_key_header or (credentials.credentials if credentials else None)
    if not token:
        raise HTTPException(status_code=401, detail="Missing API key")
    # Constant-time comparison so the response time does not leak prefixes of
    # valid API keys to an attacker probing the endpoint. ``in`` on a set
    # short-circuits on the first differing character.
    if not any(hmac.compare_digest(token, k) for k in allowed_keys):
        raise HTTPException(status_code=403, detail="Invalid API key")
    return token

@app.post("/v1/embeddings")
async def create_embeddings(
    request: EmbeddingRequest,
    raw_request: Request,
    api_identity: str = Depends(authenticate_request),
) -> EmbeddingResponse:
    """
    Create embeddings for the input text(s).

    Args:
        request: EmbeddingRequest containing input text and model.

    Returns:
        EmbeddingResponse with vector data.
    """
    # Rate limit by IP or API key
    rate_key = (
        api_identity
        if api_identity != "anonymous"
        else (raw_request.client.host if raw_request.client else "unknown")
    )
    if not rate_limiter.is_allowed(rate_key):
        raise HTTPException(
            status_code=429, detail="Rate limit exceeded. Try again later."
        )

    # (#8) Properly normalize all input formats including nested token arrays
    texts = _normalize_input(request.input)

    # (#5) Validate individual texts
    if texts:
        _validate_texts(texts)

    if len(texts) > config.EMBEDDING_SERVER_MAX_REQUEST_ITEMS:
        raise HTTPException(
            status_code=413,
            detail=(
                "Batch size "
                f"{len(texts)} exceeds EMBEDDING_SERVER_MAX_REQUEST_ITEMS="
                f"{config.EMBEDDING_SERVER_MAX_REQUEST_ITEMS}"
            ),
        )

    resolved_model_name, repo_id = registry.resolve(request.model)

    if not texts:
        embeddings: list[list[float]] = []
    else:
        try:
            effective_batch = min(
                config.EMBEDDING_BATCH_SIZE,
                config.EMBEDDING_SERVER_MAX_REQUEST_ITEMS,
            )
            # (#2) Run embedding in executor to avoid blocking the event loop
            loop = asyncio.get_running_loop()
            embeddings = await loop.run_in_executor(
                None,
                lambda: embed_texts(
                    texts, model_id=repo_id, batch_size=effective_batch
                ),
            )
        except Exception as e:
            # Log the full error internally but return generic message
            _logger.exception("Embedding failed: %s", e)
            raise HTTPException(
                status_code=500,
                detail="Embedding operation failed. Check server logs for details.",
            )

    # Fake token usage (optional – some tools expect it)
    total_tokens = sum(len(t.split()) for t in texts)
    usage_meter.record(api_identity, total_tokens)

    return EmbeddingResponse(
        data=[
            EmbeddingData(embedding=emb, index=i) for i, emb in enumerate(embeddings)
        ],
        model=resolved_model_name or repo_id,
        usage={"prompt_tokens": total_tokens, "total_tokens": total_tokens},
    )

@app.get("/v1/models")
async def list_models(
    api_identity: str = Depends(authenticate_request),
) -> dict[str, Any]:
    """List available embedding models (requires auth when configured)."""
    _ = api_identity  # dependency enforces auth when enabled
    return {"data": registry.list_models(), "object": "list"}

@app.get("/v1/usage")
async def usage(api_identity: str = Depends(authenticate_request)) -> dict[str, Any]:
    """Return aggregate or per-key usage statistics.

    When auth is configured, return only the caller's bucket. When auth is
    disabled, return a single aggregated total — the per-identity buckets
    are keyed by client IP and exposing the full list to anyone is an
    information leak.
    """
    if config.EMBEDDING_SERVER_API_KEYS:
        return {"object": "usage", "data": usage_meter.snapshot(api_identity)}
    return {"object": "usage", "data": usage_meter.snapshot(aggregate=True)}

def run_server(host: str | None = None, port: int | None = None) -> None:
    """Run the embedding server.

    Can be called programmatically or via the ``simplevecdb-server`` CLI.

    Examples
    --------
    Run with default settings::

        $ simplevecdb-server

    Override host and port::

        $ simplevecdb-server --host 0.0.0.0 --port 9000

    Skip model warm-up::

        $ simplevecdb-server --no-warmup

    Args:
        host: Server host (defaults to config.SERVER_HOST).
        port: Server port (defaults to config.SERVER_PORT).
    """
    # (#6) Only parse CLI args when invoked as entry point (not programmatically)
    skip_warmup = False
    if host is None and port is None:
        parser = _build_cli_parser()
        args = parser.parse_args()
        host = args.host
        port = args.port
        skip_warmup = args.no_warmup

    host = host or config.SERVER_HOST
    port = port or config.SERVER_PORT

    # (#7) Startup banner with config summary
    auth_status = (
        f"{len(config.EMBEDDING_SERVER_API_KEYS)} key(s)"
        if config.EMBEDDING_SERVER_API_KEYS
        else "DISABLED"
    )
    _logger.info(
        "\n"
        "┌─────────────────────────────────────────────┐\n"
        "│       SimpleVecDB Embeddings Server         │\n"
        "├─────────────────────────────────────────────┤\n"
        "│  Host:       %-30s│\n"
        "│  Port:       %-30s│\n"
        "│  Model:      %-30s│\n"
        "│  Auth:       %-30s│\n"
        "│  Rate limit: %-30s│\n"
        "│  Version:    %-30s│\n"
        "└─────────────────────────────────────────────┘",
        host,
        port,
        config.EMBEDDING_MODEL,
        auth_status,
        "100 req/min, burst 20",
        _server_version,
    )

    # Security warnings
    if not config.EMBEDDING_SERVER_API_KEYS:
        _logger.warning(
            "No API keys configured (EMBEDDING_SERVER_API_KEYS is empty). "
            "Server is running without authentication. "
            "Set EMBEDDING_SERVER_API_KEYS for production use."
        )
    # Coordinate the per-request item cap with the encode-call cap. The
    # validator also runs at module import (see _validate_request_item_cap)
    # so non-CLI ASGI deployments are covered too; calling it here keeps
    # CLI startup fail-fast behaviour identical.
    _validate_request_item_cap()

    if host == "0.0.0.0":
        _logger.warning(
            "Server binding to all interfaces (0.0.0.0). "
            "This exposes the server to the network. "
            "Use 127.0.0.1 for local-only access."
        )

    # (#3) Model warm-up — pre-load default model before accepting traffic
    if not skip_warmup:
        _logger.info("Warming up default model: %s ...", config.EMBEDDING_MODEL)
        try:
            get_embedder(config.EMBEDDING_MODEL)
            _logger.info("Model warm-up complete.")
        except Exception:
            _logger.warning(
                "Model warm-up failed (will retry on first request).", exc_info=True
            )

    # (#1) Graceful shutdown with in-flight request draining
    uvi_config = uvicorn.Config(
        app, host=host, port=port, log_level="info", timeout_graceful_shutdown=10
    )
    server = uvicorn.Server(uvi_config)

    # Install signal handlers that tell uvicorn to drain gracefully
    original_sigint = signal.getsignal(signal.SIGINT)
    original_sigterm = signal.getsignal(signal.SIGTERM)

    def _graceful_shutdown(signum: int, frame: Any) -> None:
        sig_name = "SIGINT" if signum == signal.SIGINT else "SIGTERM"
        _logger.info("Received %s — draining in-flight requests...", sig_name)
        server.should_exit = True

    signal.signal(signal.SIGINT, _graceful_shutdown)
    signal.signal(signal.SIGTERM, _graceful_shutdown)

    try:
        server.run()
    finally:
        signal.signal(signal.SIGINT, original_sigint)
        signal.signal(signal.SIGTERM, original_sigterm)