Add secret selection prompt for Fireworks upload in CLI

- Introduced `_prompt_select_secrets` function to allow users to select environment variables for upload as secrets.
- Implemented fallback selection method for non-interactive environments.
- Updated `upload_command` to utilize the new secret selection logic.
- Enhanced user experience with improved prompts and error handling.
- Added `_get_questionary_style` function for consistent CLI styling.

Author: Dylan Huang

eval_protocol/cli_commands/upload.py

@@ -1,4 +1,5 @@
 import argparse
+from eval_protocol.cli_commands.utils import DiscoveredTest
 import importlib.util
 import os
 import re
@@ -14,10 +15,9 @@
     _build_entry_point,
     _build_evaluator_dashboard_url,
     _discover_and_select_tests,
-    _discover_tests,
     _ensure_account_id,
+    _get_questionary_style,
     _normalize_evaluator_id,
-    _prompt_select,
 )
 
 
@@ -170,6 +170,109 @@ def _mask_secret_value(value: str) -> str:
         return "<masked>"
 
 
+def _prompt_select_secrets(
+    secrets: Dict[str, str],
+    secrets_from_env_file: Dict[str, str],
+    non_interactive: bool,
+) -> Dict[str, str]:
+    """
+    Prompt user to select which environment variables to upload as secrets.
+    Returns the selected secrets.
+    """
+    if not secrets:
+        return {}
+
+    if non_interactive:
+        return secrets
+
+    # Check if running in a non-TTY environment (e.g., CI/CD)
+    if not sys.stdin.isatty():
+        return secrets
+
+    try:
+        import questionary
+
+        custom_style = _get_questionary_style()
+
+        # Build choices with source info and masked values
+        choices = []
+        for key, value in secrets.items():
+            source = ".env" if key in secrets_from_env_file else "env"
+            masked = _mask_secret_value(value)
+            label = f"{key} ({source}: {masked})"
+            choices.append(questionary.Choice(title=label, value=key, checked=True))
+
+        if len(choices) == 0:
+            return {}
+
+        print("\nFound environment variables to upload as Fireworks secrets:")
+        selected_keys = questionary.checkbox(
+            "Select secrets to upload:",
+            choices=choices,
+            style=custom_style,
+            pointer=">",
+            instruction="(↑↓ move, space select, enter confirm)",
+        ).ask()
+
+        if selected_keys is None:
+            # User cancelled with Ctrl+C
+            print("\nSecret upload cancelled.")
+            return {}
+
+        return {k: v for k, v in secrets.items() if k in selected_keys}
+
+    except ImportError:
+        # Fallback to simple text-based selection
+        return _prompt_select_secrets_fallback(secrets, secrets_from_env_file)
+    except KeyboardInterrupt:
+        print("\n\nSecret upload cancelled.")
+        return {}
+
+
+def _prompt_select_secrets_fallback(
+    secrets: Dict[str, str],
+    secrets_from_env_file: Dict[str, str],
+) -> Dict[str, str]:
+    """Fallback prompt selection for when questionary is not available."""
+    print("\n" + "=" * 60)
+    print("Found environment variables to upload as Fireworks secrets:")
+    print("=" * 60)
+    print("\nTip: Install questionary for better UX: pip install questionary\n")
+
+    secret_list = list(secrets.items())
+    for idx, (key, value) in enumerate(secret_list, 1):
+        source = ".env" if key in secrets_from_env_file else "env"
+        masked = _mask_secret_value(value)
+        print(f"  [{idx}] {key} ({source}: {masked})")
+
+    print("\n" + "=" * 60)
+    print("Enter numbers to select (comma-separated), 'all' for all, or 'none' to skip:")
+
+    try:
+        choice = input("Selection: ").strip().lower()
+    except KeyboardInterrupt:
+        print("\nSecret upload cancelled.")
+        return {}
+
+    if not choice or choice == "none":
+        return {}
+
+    if choice == "all":
+        return secrets
+
+    try:
+        indices = [int(x.strip()) for x in choice.split(",")]
+        selected = {}
+        for idx in indices:
+            if 1 <= idx <= len(secret_list):
+                key, value = secret_list[idx - 1]
+                selected[key] = value
+        return selected
+    except ValueError:
+        print("Invalid input. Skipping secret upload.")
+        return {}
+
+
 def upload_command(args: argparse.Namespace) -> int:
     root = os.path.abspath(getattr(args, "path", "."))
     entries_arg = getattr(args, "entry", None)
@@ -181,7 +284,7 @@ def upload_command(args: argparse.Namespace) -> int:
             qualname, resolved_path = _resolve_entry_to_qual_and_source(e, root)
             selected_specs.append((qualname, resolved_path))
     else:
-        selected_tests = _discover_and_select_tests(root, non_interactive=non_interactive)
+        selected_tests: list[DiscoveredTest] | None = _discover_and_select_tests(root, non_interactive=non_interactive)
         if not selected_tests:
             return 1
         selected_specs = [(t.qualname, t.file_path) for t in selected_tests]
@@ -212,24 +315,34 @@ def upload_command(args: argparse.Namespace) -> int:
             secrets_from_file["FIREWORKS_API_KEY"] = fw_api_key_value
 
         if fw_account_id and secrets_from_file:
-            print(f"Found {len(secrets_from_file)} API keys to upload as Fireworks secrets...")
             if secrets_from_env_file and os.path.exists(env_file_path):
                 print(f"Loading secrets from: {env_file_path}")
 
-            for secret_name, secret_value in secrets_from_file.items():
-                source = ".env" if secret_name in secrets_from_env_file else "environment"
-                print(
-                    f"Ensuring {secret_name} is registered as a secret on Fireworks for rollout... "
-                    f"({source}: {_mask_secret_value(secret_value)})"
-                )
-                if create_or_update_fireworks_secret(
-                    account_id=fw_account_id,
-                    key_name=secret_name,
-                    secret_value=secret_value,
-                ):
-                    print(f"✓ {secret_name} secret created/updated on Fireworks.")
-                else:
-                    print(f"Warning: Failed to create/update {secret_name} secret on Fireworks.")
+            # Prompt user to select which secrets to upload
+            selected_secrets = _prompt_select_secrets(
+                secrets_from_file,
+                secrets_from_env_file,
+                non_interactive,
+            )
+
+            if selected_secrets:
+                print(f"\nUploading {len(selected_secrets)} selected secret(s) to Fireworks...")
+                for secret_name, secret_value in selected_secrets.items():
+                    source = ".env" if secret_name in secrets_from_env_file else "environment"
+                    print(
+                        f"Ensuring {secret_name} is registered as a secret on Fireworks for rollout... "
+                        f"({source}: {_mask_secret_value(secret_value)})"
+                    )
+                    if create_or_update_fireworks_secret(
+                        account_id=fw_account_id,
+                        key_name=secret_name,
+                        secret_value=secret_value,
+                    ):
+                        print(f"✓ {secret_name} secret created/updated on Fireworks.")
+                    else:
+                        print(f"Warning: Failed to create/update {secret_name} secret on Fireworks.")
+            else:
+                print("No secrets selected for upload.")
         else:
             if not fw_account_id:
                 print(

eval_protocol/cli_commands/utils.py

@@ -23,6 +23,28 @@
 from ..fireworks_rft import _map_api_host_to_app_host
 
 
+def _get_questionary_style():
+    """Get the shared questionary style for CLI prompts - minimal and clean."""
+    try:
+        from questionary import Style
+
+        return Style(
+            [
+                ("qmark", "fg:#888888"),
+                ("question", "bold"),
+                ("answer", "noinherit"),
+                ("pointer", "noinherit"),
+                ("highlighted", "noinherit"),
+                ("selected", "noinherit"),
+                ("separator", "noinherit"),
+                ("instruction", "noinherit fg:#888888"),
+                ("text", "noinherit"),
+            ]
+        )
+    except ImportError:
+        return None
+
+
 @dataclass
 class DiscoveredTest:
     module_path: str
@@ -233,22 +255,8 @@ def _prompt_select_interactive(tests: list[DiscoveredTest]) -> list[DiscoveredTe
     """Interactive selection with arrow keys using questionary."""
     try:
         import questionary
-        from questionary import Style
 
-        # Custom style similar to Vercel CLI
-        custom_style = Style(
-            [
-                ("qmark", "fg:#673ab7 bold"),
-                ("question", "bold"),
-                ("answer", "fg:#2196f3 bold"),
-                ("pointer", "fg:#673ab7 bold"),
-                ("highlighted", "fg:#673ab7 bold"),
-                ("selected", "fg:#cc5454"),
-                ("separator", "fg:#cc5454"),
-                ("instruction", ""),
-                ("text", ""),
-            ]
-        )
+        custom_style = _get_questionary_style()
 
         # Check if only one test - auto-select it
         if len(tests) == 1:
@@ -259,25 +267,31 @@ def _prompt_select_interactive(tests: list[DiscoveredTest]) -> list[DiscoveredTe
             else:
                 return []
 
-        # Single-select UX
-        print("\n")
-        print("Tip: Use ↑/↓ arrows to navigate and press ENTER to select.\n")
-
+        # Build checkbox choices
         choices = []
         for idx, t in enumerate(tests, 1):
             choice_text = _format_test_choice(t, idx)
-            choices.append({"name": choice_text, "value": idx - 1})
+            choices.append(questionary.Choice(title=choice_text, value=idx - 1, checked=False))
 
-        selected = questionary.select(
-            "Select an evaluation test to upload:", choices=choices, style=custom_style
+        print()
+        selected_indices = questionary.checkbox(
+            "Select evaluation tests to upload:",
+            choices=choices,
+            style=custom_style,
+            pointer=">",
+            instruction="(↑↓ move, space select, enter confirm)",
         ).ask()
 
-        if selected is None:  # Ctrl+C
+        if selected_indices is None:  # Ctrl+C
             print("\nUpload cancelled.")
             return []
 
-        print("\n✓ Selected 1 test")
-        return [tests[selected]]
+        if not selected_indices:
+            return []
+
+        selected_tests = [tests[i] for i in selected_indices]
+        print(f"\n✓ Selected {len(selected_tests)} test(s)")
+        return selected_tests
 
     except ImportError:
         # Fallback to simpler implementation
@@ -355,8 +369,9 @@ def _discover_and_select_tests(project_root: str, non_interactive: bool) -> Opti
 
     try:
         selected_tests = _prompt_select(tests, non_interactive=non_interactive)
-    except Exception:
-        print("Error: Failed to open selector UI. Please pass --evaluator or --entry explicitly.")
+    except Exception as e:
+        print(f"Error: Failed to open selector UI: {e}")
+        print("Please pass --evaluator or --entry explicitly.")
         return None
 
     if not selected_tests: