# Copyright (c) Microsoft Corporation.
# Licensed under the MIT license.

"""
Lightweight shared CLI argument definitions for PyRIT frontends.

This module contains constants, validators, help text, and argument parsers
that are shared between ``pyrit_shell``, ``pyrit_scan``, and other CLI entry
points.  It intentionally avoids heavy imports (no ``pyrit.scenario``,
``pyrit.registry``, ``pyrit.setup``, etc.) so it can be loaded quickly for
argument parsing before the full runtime is initialised.
"""

from __future__ import annotations

import argparse
import dataclasses
import inspect
import json
import logging
import shlex
from enum import Enum
from pathlib import Path
from typing import TYPE_CHECKING, Any, get_args, get_origin

from pyrit.common.cli_helpers import (
    CONFIG_FILE_HELP,
    validate_log_level,
    validate_log_level_argparse,
)

if TYPE_CHECKING:
    from collections.abc import Callable

    from pyrit.models.parameter import Parameter

# ---------------------------------------------------------------------------
# Database type constants
# ---------------------------------------------------------------------------
IN_MEMORY = "InMemory"
SQLITE = "SQLite"
AZURE_SQL = "AzureSQL"


# ---------------------------------------------------------------------------
# Scenario-results views
# ---------------------------------------------------------------------------


class ScenarioResultView(str, Enum):
    """
    Granularity of a ``scenario-results`` render.

    Defined here (a lightweight, parse-time-safe module) rather than alongside
    the Pydantic payload models in ``pyrit.cli._results`` so the argument parsers
    can reference it without importing ``pydantic`` on the ``--help`` path.
    Inherits from ``str`` so the value round-trips cleanly through argparse.
    """

    #: Scenario-level aggregate (totals + per-group success rates).
    OVERVIEW = "overview"
    #: One row per individual attack result.
    ATTACKS = "attacks"
    #: Individual messages for each attack result conversation.
    CONVERSATIONS = "conversations"
    #: All of the above.
    FULL = "full"


def parse_scenario_result_view(raw: str) -> ScenarioResultView:
    """
    Parse a ``--view`` token into a ``ScenarioResultView``.

    Used as an argparse ``type=`` so an invalid value produces an error that
    lists the valid view names (``ScenarioResultView(raw)`` alone would raise a
    bare ``ValueError`` argparse renders without the choices).

    Args:
        raw (str): The raw ``--view`` token.

    Returns:
        ScenarioResultView: The matching view.

    Raises:
        argparse.ArgumentTypeError: If *raw* is not a valid view name.
    """
    try:
        return ScenarioResultView(raw)
    except ValueError:
        valid = ", ".join(view.value for view in ScenarioResultView)
        raise argparse.ArgumentTypeError(f"invalid view '{raw}' (choose from {valid})") from None


# ---------------------------------------------------------------------------
# Pure validators
# ---------------------------------------------------------------------------


def validate_database(*, database: str) -> str:
    """
    Validate database type.

    Args:
        database: Database type string.

    Returns:
        Validated database type.

    Raises:
        ValueError: If database type is invalid.
    """
    valid_databases = [IN_MEMORY, SQLITE, AZURE_SQL]
    if database not in valid_databases:
        raise ValueError(f"Invalid database type: {database}. Must be one of: {', '.join(valid_databases)}")
    return database


def validate_integer(value: str, *, name: str = "value", min_value: int | None = None) -> int:
    """
    Validate and parse an integer value.

    Note: The 'value' parameter is positional (not keyword-only) to allow use with
    argparse lambdas like: lambda v: validate_integer(v, min_value=1).
    This is an exception to the PyRIT style guide for argparse compatibility.

    Args:
        value: String value to parse.
        name: Parameter name for error messages. Defaults to "value".
        min_value: Optional minimum value constraint.

    Returns:
        Parsed integer.

    Raises:
        ValueError: If value is not a valid integer or violates constraints.
    """
    # Reject boolean types explicitly (int(True) == 1, int(False) == 0)
    if isinstance(value, bool):
        raise ValueError(f"{name} must be an integer string, got boolean: {value}")

    # Ensure value is a string
    if not isinstance(value, str):
        raise ValueError(f"{name} must be a string, got {type(value).__name__}: {value}")

    # Strip whitespace and validate it looks like an integer
    value = value.strip()
    if not value:
        raise ValueError(f"{name} cannot be empty")

    try:
        int_value = int(value)
    except (ValueError, TypeError) as e:
        raise ValueError(f"{name} must be an integer, got: {value}") from e

    if min_value is not None and int_value < min_value:
        raise ValueError(f"{name} must be at least {min_value}, got: {int_value}")

    return int_value


# ---------------------------------------------------------------------------
# Argparse adapter
# ---------------------------------------------------------------------------


def _argparse_validator(validator_func: Callable[..., Any]) -> Callable[[Any], Any]:
    """
    Adapt a validator to argparse by converting ValueError to ArgumentTypeError.

    This decorator adapts our keyword-only validators for use with argparse's type= parameter.
    It handles two challenges:

    1. Exception Translation: argparse expects ArgumentTypeError, but our validators raise
       ValueError. This decorator catches ValueError and re-raises as ArgumentTypeError.

    2. Keyword-Only Parameters: PyRIT validators use keyword-only parameters (e.g.,
       validate_database(*, database: str)), but argparse's type= passes a positional argument.
       This decorator inspects the function signature and calls the validator with the correct
       keyword argument name.

    This pattern allows us to:
    - Keep validators as pure functions with proper type hints
    - Follow PyRIT style guide (keyword-only parameters)
    - Reuse the same validation logic in both argparse and non-argparse contexts

    Args:
        validator_func: Function that raises ValueError on invalid input.
            Must have at least one parameter (can be keyword-only).

    Returns:
        Wrapped function that:
        - Accepts a single positional argument (for argparse compatibility)
        - Calls validator_func with the correct keyword argument
        - Raises ArgumentTypeError instead of ValueError

    Raises:
        ValueError: If validator_func has no parameters.
    """
    # Get the first parameter name from the function signature
    sig = inspect.signature(validator_func)
    params = list(sig.parameters.keys())
    if not params:
        raise ValueError(f"Validator function {validator_func.__name__} must have at least one parameter")  # type: ignore[ty:unresolved-attribute]
    first_param = params[0]

    def wrapper(value: Any) -> Any:
        try:
            # Call with keyword argument to support keyword-only parameters
            return validator_func(**{first_param: value})
        except ValueError as e:
            raise argparse.ArgumentTypeError(str(e)) from e

    # Preserve function metadata for better debugging
    wrapper.__name__ = getattr(validator_func, "__name__", "argparse_validator")
    wrapper.__doc__ = getattr(validator_func, "__doc__", None)
    return wrapper


# ---------------------------------------------------------------------------
# Path / env-file helpers
# ---------------------------------------------------------------------------


def resolve_env_files(*, env_file_paths: list[str]) -> list[Path]:
    """
    Resolve environment file paths to absolute Path objects.

    Args:
        env_file_paths: List of environment file path strings.

    Returns:
        List of resolved Path objects.

    Raises:
        ValueError: If any path does not exist.
    """
    resolved_paths = []
    for path_str in env_file_paths:
        path = Path(path_str).resolve()
        if not path.exists():
            raise ValueError(f"Environment file not found: {path}")
        resolved_paths.append(path)
    return resolved_paths


# ---------------------------------------------------------------------------
# Argparse-compatible validators
#
# These wrappers adapt our core validators (which use keyword-only parameters and raise
# ValueError) for use with argparse's type= parameter (which passes positional arguments
# and expects ArgumentTypeError).
#
# Pattern:
#   - Use core validators (validate_database, validate_log_level, etc.) in regular code
#   - Use these _argparse versions ONLY in parser.add_argument(..., type=...)
#
# The lambda wrappers for validate_integer are necessary because we need to partially
# apply the min_value parameter while still allowing the decorator to work correctly.
# ---------------------------------------------------------------------------
validate_database_argparse = _argparse_validator(validate_database)
positive_int = _argparse_validator(lambda v: validate_integer(v, min_value=1))
non_negative_int = _argparse_validator(lambda v: validate_integer(v, min_value=0))
resolve_env_files_argparse = _argparse_validator(resolve_env_files)


# ---------------------------------------------------------------------------
# Memory label / argument parsing
# ---------------------------------------------------------------------------


def parse_memory_labels(json_string: str) -> dict[str, str]:
    """
    Parse memory labels from a JSON string.

    Args:
        json_string: JSON string containing label key-value pairs.

    Returns:
        Dictionary of labels.

    Raises:
        ValueError: If JSON is invalid or contains non-string values.
    """
    try:
        labels = json.loads(json_string)
    except json.JSONDecodeError as e:
        raise ValueError(f"Invalid JSON for memory labels: {e}") from e

    if not isinstance(labels, dict):
        raise ValueError("Memory labels must be a JSON object (dictionary)")

    # Validate all keys and values are strings and build a precisely typed result
    validated_labels: dict[str, str] = {}
    for key, value in labels.items():
        if not isinstance(key, str) or not isinstance(value, str):
            raise ValueError(f"All label keys and values must be strings. Got: {key}={value}")
        validated_labels[key] = value

    return validated_labels


def parse_dataset_filter(arg: str) -> tuple[str, str]:
    """
    Parse a single ``KEY=VALUE`` dataset-filter token from the CLI.

    Note: The ``arg`` parameter is positional (not keyword-only) so it can be used directly
    as an argparse ``type=`` callable and an ``_ArgSpec`` parser. This mirrors
    ``_parse_initializer_arg`` and is an intentional exception to the keyword-only style rule
    for argparse compatibility.

    Args:
        arg (str): The raw ``KEY=VALUE`` token.

    Returns:
        tuple[str, str]: The (key, value) pair. The value keeps its raw string form so the
            server can coerce and validate it.

    Raises:
        ValueError: If the token is not in ``KEY=VALUE`` form or the key is empty. Argparse
            converts this into a clean CLI error; the shell catches it directly.
    """
    if "=" not in arg:
        raise ValueError(f"Dataset filter must be in KEY=VALUE form, got: {arg!r}")
    key, _, value = arg.partition("=")
    key = key.strip()
    if not key:
        raise ValueError(f"Dataset filter key cannot be empty in: {arg!r}")
    return key, value


def collapse_dataset_filters(tokens: list[tuple[str, str]]) -> dict[str, list[str]]:
    """
    Fold parsed ``KEY=VALUE`` dataset-filter tokens into list-valued filters, rejecting duplicates.

    Repeating a key (e.g. ``harm_categories=cyber harm_categories=violence``) would otherwise be
    silently collapsed by ``dict(...)`` to the last value, dropping earlier constraints. Since
    list-valued filters accept comma-separated values, a repeated key is almost certainly a
    mistake, so this fails loud instead. Each value is coerced into a list of tokens (comma
    splitting is CLI input parsing); the server-side request model validates the keys.

    Args:
        tokens (list[tuple[str, str]]): The parsed ``(key, value)`` pairs.

    Returns:
        dict[str, list[str]]: The collapsed, list-valued filter mapping.

    Raises:
        ValueError: If any key appears more than once.
    """
    filters: dict[str, list[str]] = {}
    for key, value in tokens:
        if key in filters:
            raise ValueError(
                f"Duplicate dataset filter '{key}'; combine values with commas: {key}={','.join(filters[key])},{value}"
            )
        filters[key] = _coerce_filter_values(value)
    return filters


def _coerce_filter_values(value: str) -> list[str]:
    """
    Split a raw dataset-filter value string into a list of non-empty tokens.

    Comma-splitting raw CLI input is a frontend concern, so it lives here next to
    ``parse_dataset_filter`` rather than in the dataset-config module that consumes filters.

    Args:
        value (str): The raw comma-separated value string.

    Returns:
        list[str]: The cleaned list of values.
    """
    return [part.strip() for part in value.split(",") if part.strip()]


# ---------------------------------------------------------------------------
# Shared argument help text
# ---------------------------------------------------------------------------

# Per-key help for --dataset-filters. Kept in sync with
# pyrit.models.catalog.scenario.DATASET_FILTERS (this module deliberately avoids importing
# pyrit.models to keep argument-parsing startup fast); tests/unit/cli/test_pyrit_scan.py pins
# the key sets equal so the two cannot drift.
#
# Each value states this key's comma-list semantics, because those differ per field (the actual
# behavior lives in get_seeds, in pyrit.memory). When you add a key here to match a new
# DATASET_FILTERS entry, the surrounding entries model the expected one-line semantics note --
# write one for the new key too.
_DATASET_FILTER_HELP: dict[str, str] = {
    "harm_categories": "matches seeds tagged with ALL given values (AND, substring match), "
    "so harm_categories=cyber,violence is an intersection",
    "data_types": "matches seeds of ANY given value (OR, exact match), so data_types=text,image_path is a union",
}

ARG_HELP = {
    "config_file": CONFIG_FILE_HELP,
    "initializers": (
        "Built-in initializer names to run before the scenario. "
        "Supports optional params with name:key=val syntax "
        "(e.g., target:tags=default,scorer dataset:mode=strict)"
    ),
    "initialization_scripts": "Paths to custom Python initialization scripts to run before the scenario",
    "env_files": "Paths to environment files to load in order (e.g., .env.production .env.local). Later files "
    "override earlier ones.",
    "scenario_techniques": "List of technique names to run (e.g., base64 rot13). Append one or more "
    "registered converters to a technique with ':converter.<name>' (repeatable), e.g. "
    "role_play_movie_script:converter.translation_spanish:converter.leetspeak. The converter is appended on top of "
    "the technique's built-in converters. Use --list-converters to see registered converter names",
    "max_concurrency": "Maximum number of concurrent attack executions (must be >= 1)",
    "max_retries": "Maximum number of automatic retries on exception (must be >= 0)",
    "memory_labels": 'Additional labels as JSON string (e.g., \'{"experiment": "test1"}\')',
    "database": "Database type to use for memory storage",
    "log_level": "Logging level",
    "dataset_names": "List of dataset names to use instead of scenario defaults (e.g., harmbench advbench). "
    "Creates a new dataset config; fetches all items unless --max-dataset-size is also specified",
    "max_dataset_size": "Maximum number of items to use from the dataset (must be >= 1). "
    "Limits new datasets if --dataset-names provided, otherwise overrides scenario's default limit",
    "dataset_filters": "Dataset seed filters as KEY=VALUE tokens "
    "(e.g., harm_categories=cyber data_types=text). Keys filter seeds before sizing. "
    "List values may be comma-separated, but semantics differ per key: "
    + "; ".join(f"{key} {semantics}" for key, semantics in _DATASET_FILTER_HELP.items())
    + ".",
    "target": "Name of a registered target from the TargetRegistry to use as the objective target. "
    "Targets are registered by initializers (e.g., 'target' initializer). "
    "Use --list-targets to see available target names after initializers have run",
}


def add_results_arguments(*, parser: argparse.ArgumentParser) -> None:
    """
    Add the shared ``scenario-results`` selection flags to *parser*.

    Registers ``--view``, ``--attack-result-ids``, and ``--limit`` in a
    ``scenario results`` group so that ``pyrit_scan`` (its ``scenario-results``
    sub-parser) and ``pyrit_shell`` (``build_scenario_results_parser``) expose an
    identical results interface. Both surfaces take the scenario-result id as a
    positional, added by the caller.

    ``--view`` defaults to ``None`` (not ``OVERVIEW``) so callers can tell an
    explicit ``--view`` apart from an omitted one; resolve it with
    ``pyrit.cli._results.resolve_view``.

    Args:
        parser (argparse.ArgumentParser): The parser to extend.
    """
    group = parser.add_argument_group("scenario results")
    group.add_argument(
        "--view",
        type=parse_scenario_result_view,
        default=None,
        metavar="{" + ",".join(view.value for view in ScenarioResultView) + "}",
        help="Result granularity: 'overview' (aggregate, default), 'attacks' (per-attack table), "
        "'conversations' (per-attack message transcripts), or 'full' (attacks + conversations)",
    )
    group.add_argument(
        "--attack-result-ids",
        nargs="+",
        metavar="ID",
        help="Restrict the view to these attack result ids (default: all attacks in the run)",
    )
    group.add_argument(
        "--limit",
        type=positive_int,
        metavar="N",
        help="Show at most N attacks (ignored for --view overview; defaults to 5 for "
        "--view conversations/full when no --attack-result-ids is given)",
    )


def build_scenario_results_parser() -> argparse.ArgumentParser:
    """
    Build the ``pyrit_shell`` parser for ``scenario-results <id> [flags]``.

    The shell takes the scenario-result id positionally (scan takes it as the
    ``--scenario-results`` value), then shares the remaining selection flags via
    ``add_results_arguments``. ``add_help`` is disabled so a bad line raises
    ``SystemExit`` for the caller to catch instead of printing argparse's help.

    Returns:
        argparse.ArgumentParser: The configured parser.
    """
    parser = argparse.ArgumentParser(prog="scenario-results", add_help=False)
    parser.add_argument("scenario_result_id", help="Scenario result id to inspect")
    add_results_arguments(parser=parser)
    return parser


# ---------------------------------------------------------------------------
# Initializer argument parsing
# ---------------------------------------------------------------------------


def _parse_initializer_arg(arg: str) -> str | dict[str, Any]:
    """
    Parse an initializer CLI argument into a string or dict for ConfigurationLoader.

    Supports two formats:
    - Simple name: "simple" → "simple"
    - Name with params: "target:tags=default,scorer" → {"name": "target", "args": {"tags": ["default", "scorer"]}}

    For multiple params on one initializer, separate with semicolons: "name:key1=val1;key2=val2"
    For multiple initializers with params, space-separate them: "target:tags=a,b dataset:mode=strict"

    Args:
        arg: The CLI argument string.

    Returns:
        str | dict[str, Any]: A plain name string, or a dict with 'name' and 'args' keys.

    Raises:
        ValueError: If the argument format is invalid.
    """
    if ":" not in arg:
        return arg

    name, params_str = arg.split(":", 1)
    if not name:
        raise ValueError(f"Invalid initializer argument '{arg}': missing name before ':'")

    args: dict[str, list[str]] = {}
    for pair in params_str.split(";"):
        pair = pair.strip()
        if not pair:
            continue
        if "=" not in pair:
            raise ValueError(f"Invalid initializer parameter '{pair}' in '{arg}': expected key=value format")
        key, value = pair.split("=", 1)
        key = key.strip()
        if not key:
            raise ValueError(f"Invalid initializer parameter in '{arg}': empty key")
        args[key] = [v.strip() for v in value.split(",")]

    if args:
        return {"name": name, "args": args}
    return name


# ---------------------------------------------------------------------------
# Shell argument specification
# ---------------------------------------------------------------------------


@dataclasses.dataclass(frozen=True)
class _ArgSpec:
    """
    Declarative specification for a single shell-mode CLI argument.

    Each instance describes one CLI flag (or set of aliases) and how its
    value(s) should be collected and validated. A list of ``_ArgSpec`` objects
    is passed to ``_parse_shell_arguments`` which handles the actual parsing
    loop. Adding a new flag only requires defining a new ``_ArgSpec``
    constant, not editing any parsing logic.

    Attributes:
        flags: CLI flag strings that trigger this argument (e.g., ``["--techniques", "-s"]``).
        result_key: Key name in the returned dict (e.g., ``"scenario_techniques"``).
        multi_value: If True, collect values until the next flag.
            If False, consume exactly one value.
        parser: Optional callable to transform each raw string value.
            Applied per-item for multi-value args, or to the single value otherwise.
    """

    flags: list[str]
    result_key: str
    multi_value: bool = False
    parser: Callable[[str], Any] | None = None


_INITIALIZERS_ARG = _ArgSpec(
    flags=["--initializers"],
    result_key="initializers",
    multi_value=True,
    parser=_parse_initializer_arg,
)
_INIT_SCRIPTS_ARG = _ArgSpec(
    flags=["--initialization-scripts"],
    result_key="initialization_scripts",
    multi_value=True,
)

_TECHNIQUES_ARG = _ArgSpec(
    flags=["--techniques", "-t"],
    result_key="scenario_techniques",
    multi_value=True,
)
_MAX_CONCURRENCY_ARG = _ArgSpec(
    flags=["--max-concurrency"],
    result_key="max_concurrency",
    parser=lambda v: validate_integer(v, name="--max-concurrency", min_value=1),
)
_MAX_RETRIES_ARG = _ArgSpec(
    flags=["--max-retries"],
    result_key="max_retries",
    parser=lambda v: validate_integer(v, name="--max-retries", min_value=0),
)
_MEMORY_LABELS_ARG = _ArgSpec(
    flags=["--memory-labels"],
    result_key="memory_labels",
    parser=parse_memory_labels,
)
_LOG_LEVEL_ARG = _ArgSpec(
    flags=["--log-level"],
    result_key="log_level",
    parser=lambda v: validate_log_level(log_level=v),
)
_DATASET_NAMES_ARG = _ArgSpec(
    flags=["--dataset-names"],
    result_key="dataset_names",
    multi_value=True,
)
_MAX_DATASET_SIZE_ARG = _ArgSpec(
    flags=["--max-dataset-size"],
    result_key="max_dataset_size",
    parser=lambda v: validate_integer(v, name="--max-dataset-size", min_value=1),
)
_DATASET_FILTERS_ARG = _ArgSpec(
    flags=["--dataset-filters"],
    result_key="dataset_filters",
    multi_value=True,
    parser=parse_dataset_filter,
)
_TARGET_ARG = _ArgSpec(
    flags=["--target"],
    result_key="target",
)

_RUN_ARG_SPECS: list[_ArgSpec] = [
    _INITIALIZERS_ARG,
    _TECHNIQUES_ARG,
    _MAX_CONCURRENCY_ARG,
    _MAX_RETRIES_ARG,
    _MEMORY_LABELS_ARG,
    _DATASET_NAMES_ARG,
    _MAX_DATASET_SIZE_ARG,
    _DATASET_FILTERS_ARG,
    _TARGET_ARG,
]

_LIST_TARGETS_ARG_SPECS: list[_ArgSpec] = [
    _INITIALIZERS_ARG,
    _INIT_SCRIPTS_ARG,
]


# ---------------------------------------------------------------------------
# Generic shell argument parser
# ---------------------------------------------------------------------------


def _parse_shell_arguments(*, parts: list[str], arg_specs: list[_ArgSpec]) -> dict[str, Any]:
    """
    Parse a list of shell tokens against a set of argument specifications.

    Each ``_ArgSpec`` in *arg_specs* declares how its flag(s) should be handled
    (multi-value collection vs. single-value consumption) and what validation
    or transformation to apply.

    Args:
        parts: Token list (already split on whitespace, positional args removed).
        arg_specs: Argument specifications that this command accepts.

    Returns:
        Dictionary mapping each spec's ``result_key`` to its parsed value,
        defaulting to ``None`` for arguments not present in *parts*.

    Raises:
        ValueError: On unknown flags or missing values.
    """
    # Build lookup: flag string → spec
    flag_to_spec: dict[str, _ArgSpec] = {}
    for spec in arg_specs:
        for flag in spec.flags:
            flag_to_spec[flag] = spec

    # Initialise result with None defaults
    result: dict[str, Any] = {spec.result_key: None for spec in arg_specs}

    i = 0
    while i < len(parts):
        token = parts[i]
        matched_spec: _ArgSpec | None = flag_to_spec.get(token)

        if matched_spec is None:
            valid = sorted(flag_to_spec.keys())
            raise ValueError(f"Unknown argument: {token}. Valid arguments: {', '.join(valid)}")

        i += 1

        if matched_spec.multi_value:
            values: list[Any] = []
            # Collect values until the next flag (whether valid or invalid)
            while i < len(parts) and not (parts[i].startswith("--") or parts[i] in flag_to_spec):
                item = matched_spec.parser(parts[i]) if matched_spec.parser else parts[i]
                values.append(item)
                i += 1
            if len(values) == 0:
                raise ValueError(f"{matched_spec.flags[0]} requires at least one value")
            result[matched_spec.result_key] = values
        else:
            if i >= len(parts):
                raise ValueError(f"{matched_spec.flags[0]} requires a value")
            raw = parts[i]
            result[matched_spec.result_key] = matched_spec.parser(raw) if matched_spec.parser else raw
            i += 1

    return result


def parse_run_arguments(*, args_string: str, declared_params: list[Parameter] | None = None) -> dict[str, Any]:
    """
    Parse run command arguments from a string (for shell mode).

    Args:
        args_string: Space-separated argument string.
        declared_params: Optional scenario-declared parameters. When supplied,
            adds ``--kebab-case`` flags for each, namespaced under
            ``scenario__<name>`` in the result dict.

    Returns:
        Dictionary mapping built-in result_keys (and ``scenario__*`` keys for
        any declared params) to their parsed values. ``scenario_name`` is
        always populated from the first positional token.

    Raises:
        ValueError: Empty input or shell parser failure.
    """
    parts = shlex.split(args_string)

    if not parts:
        raise ValueError("No scenario name provided")

    augmented_specs: list[_ArgSpec] = list(_RUN_ARG_SPECS)
    if declared_params:
        scenario_specs = [_arg_spec_from_parameter(param=p) for p in declared_params]
        scenario_specs = _resolve_scenario_flag_collisions(scenario_specs=scenario_specs, base_specs=_RUN_ARG_SPECS)
        augmented_specs.extend(scenario_specs)

    result = _parse_shell_arguments(parts=parts[1:], arg_specs=augmented_specs)
    result["scenario_name"] = parts[0]
    return result


def parse_list_targets_arguments(*, args_string: str) -> dict[str, Any]:
    """
    Parse list-targets command arguments from a string (for shell mode).

    Args:
        args_string: Space-separated argument string (e.g., "--initializers target").

    Returns:
        Dictionary with parsed arguments:
            - initializers: list[str | dict[str, Any]] | None
            - initialization_scripts: list[str] | None

    Raises:
        ValueError: If parsing or validation fails.
    """
    parts = shlex.split(args_string)
    return _parse_shell_arguments(parts=parts, arg_specs=_LIST_TARGETS_ARG_SPECS)


# ---------------------------------------------------------------------------
# Scenario-declared parameter support (Stage 2b)
# ---------------------------------------------------------------------------

# Namespacing prefix for scenario-declared params on the parsed result dict.
# Mirrors the convention in pyrit_scan.py so both parsers extract scenario
# args the same way.
_SCENARIO_RESULT_KEY_PREFIX = "scenario__"


def _normalize_scenario_flag(*, name: str) -> str:
    """Return the kebab-cased CLI flag for a scenario parameter name."""
    return f"--{name.replace('_', '-')}"


def _arg_spec_from_parameter(*, param: Parameter) -> _ArgSpec:
    """
    Build a shell ``_ArgSpec`` from a scenario ``Parameter`` declaration.

    Args:
        param (Parameter): Scenario-declared parameter.

    Returns:
        _ArgSpec: Spec with ``scenario__<name>`` result key and a parser
            that routes through ``pyrit.models.parameter.coerce_value``.
    """
    from pyrit.models.parameter import Parameter

    multi = get_origin(param.param_type) is list
    parser: Callable[[str], Any] | None
    if multi:
        # Per-element coercion via a temporary scalar-typed Parameter.
        type_args = get_args(param.param_type)
        element_type = type_args[0] if type_args else str
        element_param = Parameter(
            name=param.name,
            description=param.description,
            param_type=element_type,
        )

        def parser(raw: str) -> Any:
            return element_param.coerce_value(raw)

    elif param.param_type is None or param.param_type is str:
        # No coercion needed (plain str / untyped passthrough).
        parser = None
    else:
        # Coerce + validate (handles ints/floats/bools AND Literal/Enum membership).
        def parser(raw: str) -> Any:
            return param.coerce_value(raw)

    return _ArgSpec(
        flags=[_normalize_scenario_flag(name=param.name)],
        result_key=f"{_SCENARIO_RESULT_KEY_PREFIX}{param.name}",
        multi_value=multi,
        parser=parser,
    )


def _resolve_scenario_flag_collisions(*, scenario_specs: list[_ArgSpec], base_specs: list[_ArgSpec]) -> list[_ArgSpec]:
    """
    Drop scenario specs whose flag is already taken; first declaration wins.

    A scenario's ``supported_parameters`` (fetched from the API) include the framework's common
    parameters — e.g. ``memory_labels``, ``max_concurrency``, ``max_retries`` — which normalize to
    flags already provided as built-ins. A scenario could also declare two params that normalize to
    the same flag. In both cases the colliding spec is silently dropped and the earlier owner (the
    built-in, or the first-declared scenario param) keeps the flag. This mirrors
    ``pyrit_scan._add_scenario_params_from_api``, which skips any flag already registered on the
    parser, so the two entry points accept the same inputs.

    Args:
        scenario_specs: Specs built from scenario-declared parameters.
        base_specs: Built-in run argument specs.

    Returns:
        list[_ArgSpec]: The scenario specs whose flags do not collide with a built-in flag or an
            earlier scenario spec.
    """
    seen: set[str] = {flag for spec in base_specs for flag in spec.flags}
    resolved: list[_ArgSpec] = []
    for spec in scenario_specs:
        if any(flag in seen for flag in spec.flags):
            # Flag already owned by a built-in or an earlier scenario param; first wins.
            continue
        seen.update(spec.flags)
        resolved.append(spec)
    return resolved


def extract_scenario_args(*, parsed: dict[str, Any]) -> dict[str, Any]:
    """
    Pull scenario-declared parameter values out of a parsed shell-args dict.

    Drops keys whose value is ``None`` so absent flags don't reach
    ``Scenario.set_params_from_args`` as explicit ``None`` (the shell parser
    initializes unsupplied keys to ``None``, unlike argparse's ``SUPPRESS``).

    Args:
        parsed (dict[str, Any]): Result from ``parse_run_arguments``.

    Returns:
        dict[str, Any]: Map of original parameter name to supplied value.
    """
    return {
        key.removeprefix(_SCENARIO_RESULT_KEY_PREFIX): value
        for key, value in parsed.items()
        if key.startswith(_SCENARIO_RESULT_KEY_PREFIX) and value is not None
    }


# ---------------------------------------------------------------------------
# Shared argparse builder
# ---------------------------------------------------------------------------


def build_parameters_from_api(*, api_params: list[Parameter]) -> list[Parameter] | None:
    """
    Return a scenario catalog's ``supported_parameters`` as coercion-ready ``Parameter`` objects.

    The REST client deserializes catalog payloads directly into ``Parameter``,
    which reconstructs each parameter's live ``param_type`` from its serialized
    display fields (``type_name`` / ``choices`` / ``is_list``). The parameters are
    therefore already coercion-ready, so the shell parser can apply per-element
    coercion and treat list parameters as ``multi_value`` without further mapping.

    Args:
        api_params: Scenario-declared parameters from ``GET /api/scenarios/catalog/{name}``.

    Returns:
        list[Parameter] | None: Parameter list when ``api_params`` is non-empty, else ``None``.
    """
    return list(api_params) or None


def add_common_arguments(parser: argparse.ArgumentParser) -> None:
    """Add arguments shared between pyrit_shell and pyrit_scan."""
    parser.add_argument("--config-file", type=Path, help=ARG_HELP["config_file"])
    parser.add_argument(
        "--log-level",
        type=validate_log_level_argparse,
        default=logging.WARNING,
        help="Logging level (DEBUG, INFO, WARNING, ERROR, CRITICAL) (default: WARNING)",
    )