# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
# SPDX-License-Identifier: Apache-2.0

import argparse
import json
import logging
import os
import socket
import warnings
from typing import Any, Dict, Optional

from vllm.distributed.kv_events import KVEventsConfig
from vllm.engine.arg_utils import AsyncEngineArgs

try:
    from vllm.utils import FlexibleArgumentParser
except ImportError:
    from vllm.utils.argparse_utils import FlexibleArgumentParser

from dynamo.common.config_dump import register_encoder
from dynamo.common.configuration.groups.runtime_args import (
    DynamoRuntimeArgGroup,
    DynamoRuntimeConfig,
)
from dynamo.common.utils.runtime import parse_endpoint
from dynamo.vllm.backend_args import DynamoVllmArgGroup, DynamoVllmConfig
from dynamo.vllm.constants import DisaggregationMode

from . import envs

logger = logging.getLogger(__name__)

DEFAULT_MODEL = "Qwen/Qwen3-0.6B"


class Config(DynamoRuntimeConfig, DynamoVllmConfig):
    component: str
    custom_jinja_template: Optional[str] = None
    discovery_backend: str
    request_plane: str
    event_plane: str
    enable_local_indexer: bool = True
    use_kv_events: bool

    # mirror vLLM
    model: str
    served_model_name: Optional[str] = None

    # rest vLLM args
    engine_args: AsyncEngineArgs

    def validate(self) -> None:
        DynamoRuntimeConfig.validate(self)
        DynamoVllmConfig.validate(self)


@register_encoder(Config)
def _preprocess_for_encode_config(config: Config) -> Dict[str, Any]:
    """Convert Config object to dictionary for encoding."""
    return config.__dict__


def parse_args() -> Config:
    """Parse command-line arguments for the vLLM backend.

    Returns:
        Config: Parsed configuration object.
    """
    dynamo_runtime_argspec = DynamoRuntimeArgGroup()
    dynamo_vllm_argspec = DynamoVllmArgGroup()

    parser = argparse.ArgumentParser(
        description="Dynamo vLLM worker configuration",
        formatter_class=argparse.RawTextHelpFormatter,
        allow_abbrev=False,
    )

    # Build argument parser
    dynamo_runtime_argspec.add_arguments(parser)
    dynamo_vllm_argspec.add_arguments(parser)

    # trick to add vllm engine flags to a specific group without breaking the Dynamo groups.
    vg = parser.add_argument_group(
        "vLLM Engine Options. Please refer to vLLM documentation for more details."
    )
    vllm_parser = FlexibleArgumentParser(add_help=False)
    AsyncEngineArgs.add_cli_args(vllm_parser, async_args_only=False)

    for action in vllm_parser._actions:
        if not action.option_strings:
            continue
        vg._group_actions.append(action)

    args, unknown = parser.parse_known_args()
    dynamo_config = Config.from_cli_args(args)

    # Validate arguments
    dynamo_config.validate()

    vllm_args = vllm_parser.parse_args(unknown)
    # Set the model name from the command line arguments
    # model is defined in AsyncEngineArgs, but when AsyncEngineArgs.from_cli_args is called,
    # vllm will update the model name to the full path of the model, which will break the dynamo logic,
    # as we use the model name as served_model_name (if served_model_name is not set)
    dynamo_config.model = vllm_args.model

    engine_config = AsyncEngineArgs.from_cli_args(vllm_args)

    cross_validate_config(dynamo_config, engine_config)
    update_dynamo_config_with_engine(dynamo_config, engine_config)
    update_engine_config_with_dynamo(dynamo_config, engine_config)

    dynamo_config.engine_args = engine_config
    return dynamo_config


def cross_validate_config(
    dynamo_config: Config, engine_config: AsyncEngineArgs
) -> None:
    """Validate dynamo and engine config together. This should not modify the configs."""

    if hasattr(engine_config, "stream_interval") and engine_config.stream_interval != 1:
        logger.warning(
            "--stream-interval is currently not respected in Dynamo. "
            "Dynamo uses its own post-processing implementation on the frontend, "
            "bypassing vLLM's OutputProcessor buffering."
        )


def update_dynamo_config_with_engine(
    dynamo_config: Config, engine_config: AsyncEngineArgs
) -> None:
    """Update dynamo_config fields from engine_config and worker flags."""

    if getattr(engine_config, "served_model_name", None) is not None:
        served = engine_config.served_model_name
        if len(served) > 1:
            raise ValueError("We do not support multiple model names.")
        dynamo_config.served_model_name = served[0]
    else:
        dynamo_config.served_model_name = None

    # Capture user-provided --endpoint before defaults overwrite it
    user_endpoint = dynamo_config.endpoint

    if dynamo_config.route_to_encoder:
        dynamo_config.component = "processor"
        dynamo_config.endpoint = "generate"
    elif dynamo_config.multimodal_encode_worker:
        dynamo_config.component = "encoder"
        dynamo_config.endpoint = "generate"
    elif dynamo_config.multimodal_decode_worker:
        dynamo_config.component = "decoder"
        dynamo_config.endpoint = "generate"
    elif (
        dynamo_config.multimodal_worker
        and dynamo_config.disaggregation_mode == DisaggregationMode.PREFILL
    ):
        dynamo_config.component = "backend"
        dynamo_config.endpoint = "generate"
    elif dynamo_config.disaggregation_mode == DisaggregationMode.PREFILL:
        dynamo_config.component = "prefill"
        dynamo_config.endpoint = "generate"
    else:
        dynamo_config.component = "backend"
        dynamo_config.endpoint = "generate"

    # If user provided --endpoint, override namespace/component/endpoint
    if user_endpoint is not None:
        parsed_ns, parsed_comp, parsed_ep = parse_endpoint(user_endpoint)
        dynamo_config.namespace = parsed_ns
        dynamo_config.component = parsed_comp
        dynamo_config.endpoint = parsed_ep

    if dynamo_config.custom_jinja_template is not None:
        expanded_template_path = os.path.expanduser(
            os.path.expandvars(dynamo_config.custom_jinja_template)
        )
        dynamo_config.custom_jinja_template = expanded_template_path
        if not os.path.isfile(expanded_template_path):
            raise FileNotFoundError(
                f"Custom Jinja template file not found: {expanded_template_path}. "
                "Please ensure the file exists and the path is correct."
            )

    # --connector is no longer supported for vLLM. Raise hard error if explicitly set.
    _reject_connector_flag(dynamo_config)

    # If disaggregation mode is prefill, require explicit --kv-transfer-config
    has_kv_transfer_config = (
        hasattr(engine_config, "kv_transfer_config")
        and engine_config.kv_transfer_config is not None
    )
    if (
        dynamo_config.disaggregation_mode == DisaggregationMode.PREFILL
        and not has_kv_transfer_config
    ):
        raise ValueError(
            "--connector is deprecated and the default is no longer nixl. "
            "When using --disaggregation-mode prefill, you must explicitly "
            "provide --kv-transfer-config. Example:\n"
            "  --kv-transfer-config "
            '\'{"kv_connector":"NixlConnector","kv_role":"kv_both"}\''
        )

    # Clear connector list (no longer used for vLLM)
    dynamo_config.connector = []  # type: ignore[assignment]

    # Validate ModelExpress P2P server URL
    if getattr(engine_config, "load_format", None) in ("mx-source", "mx-target"):
        if not dynamo_config.model_express_url:
            raise ValueError(
                f"--model-express-url or MODEL_EXPRESS_URL env var is required "
                f"when using --load-format={engine_config.load_format}"
            )


def update_engine_config_with_dynamo(
    dynamo_config: Config, engine_config: AsyncEngineArgs
) -> None:
    """Update engine config based on Dynamo config."""
    if engine_config.enable_prefix_caching is None:
        logger.debug(
            "--enable-prefix-caching or --no-enable-prefix-caching not specified. "
            "Defaulting to True (vLLM v1 default behavior)"
        )
        engine_config.enable_prefix_caching = True

    if getattr(engine_config, "block_size", None) is None:
        engine_config.block_size = 16
        logger.debug(
            f"Setting reasonable default of {engine_config.block_size} for block_size"
        )

    if _uses_nixl_connector(engine_config):
        ensure_side_channel_host()

    defaults = {
        # vLLM 0.13+ renamed 'task' to 'runner'
        "runner": "generate",
        # As of vLLM >=0.10.0 the engine unconditionally calls
        # `sampling_params.update_from_tokenizer(...)`, so we can no longer
        # skip tokenizer initialisation.  Setting this to **False** avoids
        # a NoneType error when the processor accesses the tokenizer.
        "skip_tokenizer_init": False,
        "enable_log_requests": False,
        "disable_log_stats": False,
    }

    kv_cfg = create_kv_events_config(dynamo_config, engine_config)
    defaults["kv_events_config"] = kv_cfg
    dynamo_config.use_kv_events = kv_cfg is not None and kv_cfg.enable_kv_cache_events

    logger.info(
        f"Using kv_events_config for publishing vLLM kv events over zmq: {kv_cfg} "
        f"(use_kv_events={dynamo_config.use_kv_events})"
    )

    if envs.is_set("DYN_FORWARDPASS_METRIC_PORT"):
        existing_cls = getattr(engine_config, "scheduler_cls", None)
        if existing_cls is None:
            defaults[
                "scheduler_cls"
            ] = "dynamo.vllm.instrumented_scheduler.InstrumentedScheduler"
            logger.info(
                "Forward pass metrics enabled: scheduler_cls set to InstrumentedScheduler "
                f"(port={envs.DYN_FORWARDPASS_METRIC_PORT})"
            )
        else:
            logger.warning(
                f"DYN_FORWARDPASS_METRIC_PORT is set but scheduler_cls "
                f"is already '{existing_cls}'. InstrumentedScheduler will NOT "
                f"be injected. To use forward pass metrics, either remove "
                f"--scheduler-cls or subclass InstrumentedScheduler."
            )

    logger.debug("Setting Dynamo defaults for vLLM")
    for key, value in defaults.items():
        if hasattr(engine_config, key):
            setattr(engine_config, key, value)
            logger.debug(f" engine_args.{key} = {value}")
        else:
            logger.debug(
                f" Skipping engine_args.{key} (not available in this vLLM version)"
            )


def create_kv_events_config(
    dynamo_config: Config, engine_config: AsyncEngineArgs
) -> Optional[KVEventsConfig]:
    """Create KVEventsConfig for prefix caching if needed."""
    if dynamo_config.disaggregation_mode == DisaggregationMode.DECODE:
        logger.info(
            "Decode worker detected (disaggregation_mode=decode): "
            "kv_events_config disabled (decode workers don't publish KV events)"
        )
        return None

    # If prefix caching is not enabled, no events config needed
    if not engine_config.enable_prefix_caching:
        logger.info("No kv_events_config required: prefix caching is disabled")
        return None

    # If user provided their own config, use that
    if c := getattr(engine_config, "kv_events_config"):
        if not c.enable_kv_cache_events:
            logger.warning(
                "User provided --kv_events_config which set enable_kv_cache_events to False (default). "
                "To publish events, explicitly set enable_kv_cache_events to True."
            )
        logger.info(f"Using user-provided kv_events_config {c}")
        return c

    # Create default events config for prefix caching
    # TODO: move this to configuration system.
    port = envs.DYN_VLLM_KV_EVENT_PORT
    warnings.warn(
        "Automatic KV events configuration is deprecated and will be removed in "
        "the next release. After that, KV events will be disabled by default "
        "(matching upstream vLLM). To preserve current behavior, pass "
        "--kv-events-config explicitly. For example:\n"
        f'  --kv-events-config \'{{"enable_kv_cache_events":true,"publisher":"zmq","endpoint":"tcp://*:{port}"}}\'\n'
        "See docs/backends/vllm/README.md for details.",
        FutureWarning,
        stacklevel=2,
    )
    logger.info(
        f"Using env-var DYN_VLLM_KV_EVENT_PORT={port} to create kv_events_config"
    )
    dp_rank = engine_config.data_parallel_rank or 0
    return KVEventsConfig(
        enable_kv_cache_events=True,
        publisher="zmq",
        endpoint=f"tcp://*:{port - dp_rank}",  # vLLM will iterate dp_rank for us, so we need to subtract it out TODO: fix in vLLM
    )


def _uses_nixl_connector(engine_config: AsyncEngineArgs) -> bool:
    """Check if the user-provided --kv-transfer-config uses NixlConnector.

    Handles both direct usage (kv_connector="NixlConnector") and nested usage
    inside PdConnector (kv_connector_extra_config.connectors contains
    "NixlConnector").
    """
    kv_cfg = getattr(engine_config, "kv_transfer_config", None)
    if kv_cfg is None:
        return False
    if kv_cfg.kv_connector == "NixlConnector":
        return True
    # PdConnector wraps multiple connectors in kv_connector_extra_config.
    # Each entry is a dict like {"kv_connector": "NixlConnector", ...}.
    if kv_cfg.kv_connector == "PdConnector":
        extra = kv_cfg.kv_connector_extra_config or {}
        for entry in extra.get("connectors", []):
            if isinstance(entry, dict) and entry.get("kv_connector") == "NixlConnector":
                return True
    return False


def _uses_dynamo_connector(engine_config: AsyncEngineArgs) -> bool:
    """Check if the user-provided --kv-transfer-config uses DynamoConnector (KVBM).

    Handles both direct usage and nested usage inside PdConnector.
    """
    kv_cfg = getattr(engine_config, "kv_transfer_config", None)
    if kv_cfg is None:
        return False
    if kv_cfg.kv_connector == "DynamoConnector":
        return True
    if kv_cfg.kv_connector == "PdConnector":
        extra = kv_cfg.kv_connector_extra_config or {}
        for entry in extra.get("connectors", []):
            if (
                isinstance(entry, dict)
                and entry.get("kv_connector") == "DynamoConnector"
            ):
                return True
    return False


def _connector_to_kv_transfer_json(connectors: list[str]) -> str:
    """Convert a legacy --connector list to the equivalent --kv-transfer-config JSON.

    Used in error messages to help users migrate.
    """
    multi_connectors = []
    for conn in connectors:
        c = conn.lower()
        if c == "lmcache":
            multi_connectors.append(
                {"kv_connector": "LMCacheConnectorV1", "kv_role": "kv_both"}
            )
        elif c == "flexkv":
            multi_connectors.append(
                {"kv_connector": "FlexKVConnectorV1", "kv_role": "kv_both"}
            )
        elif c == "nixl":
            multi_connectors.append(
                {"kv_connector": "NixlConnector", "kv_role": "kv_both"}
            )
        elif c == "kvbm":
            multi_connectors.append(
                {
                    "kv_connector": "DynamoConnector",
                    "kv_connector_module_path": "kvbm.vllm_integration.connector",
                    "kv_role": "kv_both",
                }
            )

    if len(multi_connectors) == 1:
        return json.dumps(multi_connectors[0])

    return json.dumps(
        {
            "kv_connector": "PdConnector",
            "kv_role": "kv_both",
            "kv_connector_extra_config": {"connectors": multi_connectors},
            "kv_connector_module_path": "kvbm.vllm_integration.connector",
        }
    )


def _reject_connector_flag(dynamo_config: Config) -> None:
    """Raise ValueError if --connector was explicitly set (CLI or DYN_CONNECTOR env var).

    The --connector flag is no longer supported for the vLLM backend.
    Users must use --kv-transfer-config instead.
    """
    connector_list = dynamo_config.connector or []

    # Check if --connector was explicitly provided via CLI or DYN_CONNECTOR env var
    env_connector = os.environ.get("DYN_CONNECTOR")
    explicitly_set = bool(connector_list) or (env_connector is not None)

    if not explicitly_set:
        return

    # Normalize: "none"/"null" means no connector
    normalized = [c.lower() for c in connector_list]
    if normalized and all(c in ("none", "null") for c in normalized):
        # --connector none/null: tell user it's no longer needed
        raise ValueError(
            "--connector is no longer supported for the vLLM backend. "
            "'--connector none' is no longer needed — the default is already "
            "no connector. Simply remove the --connector flag."
        )

    # Active connectors: show migration path
    if normalized:
        equiv = _connector_to_kv_transfer_json(normalized)
        raise ValueError(
            "--connector is no longer supported for the vLLM backend. "
            "Use --kv-transfer-config instead.\n"
            f"  Equivalent: --kv-transfer-config '{equiv}'"
        )

    # DYN_CONNECTOR env var set but parsed to empty list
    if env_connector is not None:
        env_values = [v.strip().lower() for v in env_connector.split() if v.strip()]
        if env_values and not all(v in ("none", "null") for v in env_values):
            equiv = _connector_to_kv_transfer_json(env_values)
            raise ValueError(
                "The DYN_CONNECTOR environment variable is no longer supported "
                "for the vLLM backend. Use --kv-transfer-config instead.\n"
                f"  Equivalent: --kv-transfer-config '{equiv}'"
            )
        raise ValueError(
            "The DYN_CONNECTOR environment variable is no longer supported "
            "for the vLLM backend. Use --kv-transfer-config instead."
        )


def get_host_ip() -> str:
    """Get the IP address of the host for side-channel coordination."""
    try:
        host_name = socket.gethostname()
    except socket.error as exc:
        logger.warning("Failed to get hostname: %s, falling back to 127.0.0.1", exc)
        return "127.0.0.1"

    try:
        host_ip = socket.gethostbyname(host_name)
        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as test_socket:
            test_socket.bind((host_ip, 0))
        return host_ip
    except socket.gaierror as exc:
        logger.warning(
            "Hostname %s cannot be resolved: %s, falling back to 127.0.0.1",
            host_name,
            exc,
        )
        return "127.0.0.1"
    except socket.error as exc:
        logger.warning(
            "Hostname %s is not usable for binding: %s, falling back to 127.0.0.1",
            host_name,
            exc,
        )
        return "127.0.0.1"


def ensure_side_channel_host():
    """Ensure the NIXL side-channel host is available without overriding user settings."""

    existing_host = os.getenv("VLLM_NIXL_SIDE_CHANNEL_HOST")
    if existing_host:
        logger.debug(
            "Preserving existing VLLM_NIXL_SIDE_CHANNEL_HOST=%s", existing_host
        )
        return

    host_ip = get_host_ip()
    os.environ["VLLM_NIXL_SIDE_CHANNEL_HOST"] = host_ip
    logger.debug("Set VLLM_NIXL_SIDE_CHANNEL_HOST to %s", host_ip)
