refactor: remove dead code across the codebase (#5779)

## Summary

- Remove 6 unused modules (`cli/utils.py`, `core/utils/image_types.py`,
`core/utils/serialize.py`, `providers/utils/pagination.py`,
`providers/utils/scheduler.py`) and the scheduler test file
- Remove unused classes, functions, and re-exports from 15 files
including `configure_api_providers`, `resolve_remote_stack_impls`,
`build_hf_repo_model_entry`, `get_shield_registry`,
`formulate_run_args`, `run_command`, and others
- Net removal of **936 lines** of dead code identified via static
analysis

## Test plan

- [ ] Pre-commit checks pass (`uv run pre-commit run --all-files`)
- [ ] Unit tests pass (`uv run pytest tests/unit/ -x --tb=short`)
- [ ] No new import errors — removed symbols had zero callers
- [ ] API spec unchanged — no public API surface affected

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Signed-off-by: Sébastien Han <seb@redhat.com>
Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: leseb

src/ogx/cli/stack/utils.py

@@ -7,7 +7,6 @@
 from __future__ import annotations
 
 import argparse
-from enum import Enum
 from pathlib import Path
 from typing import Any
 
@@ -18,13 +17,6 @@
 TEMPLATES_PATH = Path(__file__).parent.parent.parent / "distributions"
 
 
-class ImageType(Enum):
-    """Supported image types for building OGX distributions."""
-
-    CONTAINER = "container"
-    VENV = "venv"
-
-
 def print_subcommand_description(parser: argparse.ArgumentParser, subparsers: argparse._SubParsersAction[Any]) -> None:
     """Print descriptions of subcommands."""
     description_text = ""

src/ogx/cli/utils.py

@@ -6,7 +6,6 @@
 
 import sys
 
-from pydantic import BaseModel
 from termcolor import cprint
 
 from ogx.core.datatypes import StackConfig
@@ -30,13 +29,6 @@
 ]
 
 
-class ApiInput(BaseModel):
-    """Input specification pairing an API with its provider type."""
-
-    api: Api
-    provider: str
-
-
 def get_provider_dependencies(
     config: StackConfig,
 ) -> tuple[list[str], list[str], list[str]]:

src/ogx/core/build.py

@@ -3,24 +3,18 @@
 #
 # This source code is licensed under the terms described in the LICENSE file in
 # the root directory of this source tree.
-import textwrap
 from typing import Any
 
 from ogx.core.datatypes import (
     OGX_RUN_CONFIG_VERSION,
-    DistributionSpec,
     Provider,
     StackConfig,
 )
-from ogx.core.distribution import (
-    builtin_automatically_routed_apis,
-    get_provider_registry,
-)
 from ogx.core.stack import cast_distro_name_to_string, replace_env_vars
 from ogx.core.utils.dynamic import instantiate_class_type
 from ogx.core.utils.prompt_for_config import prompt_for_config
 from ogx.log import get_logger
-from ogx_api import Api, ProviderSpec
+from ogx_api import ProviderSpec
 
 logger = get_logger(name=__name__, category="core")
 
@@ -53,89 +47,6 @@ def configure_single_provider(registry: dict[str, ProviderSpec], provider: Provi
     )
 
 
-def configure_api_providers(config: StackConfig, build_spec: DistributionSpec) -> StackConfig:
-    """Interactively configure providers for all APIs in the stack configuration.
-
-    Args:
-        config: The current stack configuration to update.
-        build_spec: The distribution specification defining available providers.
-
-    Returns:
-        The updated StackConfig with user-configured providers.
-    """
-    is_nux = len(config.providers) == 0
-
-    if is_nux:
-        logger.info(
-            textwrap.dedent(
-                """
-        OGX is composed of several APIs working together. For each API served by the Stack,
-        we need to configure the providers (implementations) you want to use for these APIs.
-"""
-            )
-        )
-
-    provider_registry = get_provider_registry()
-    builtin_apis = [a.routing_table_api for a in builtin_automatically_routed_apis()]
-
-    if config.apis:
-        apis_to_serve = config.apis
-    else:
-        apis_to_serve = [a.value for a in Api if a not in (Api.inspect, Api.providers, Api.admin)]
-
-    for api_str in apis_to_serve:
-        api = Api(api_str)
-        if api in builtin_apis:
-            continue
-        if api not in provider_registry:
-            raise ValueError(f"Unknown API `{api_str}`")
-
-        existing_providers = config.providers.get(api_str, [])
-        if existing_providers:
-            logger.info("Re-configuring existing providers for API", api=api_str)
-            updated_providers = []
-            for p in existing_providers:
-                logger.info("Configuring provider", provider_type=p.provider_type)
-                updated_providers.append(configure_single_provider(provider_registry[api], p))
-                logger.info("")
-        else:
-            # we are newly configuring this API
-            plist = build_spec.providers.get(api_str, [])
-            plist = plist if isinstance(plist, list) else [plist]
-
-            if not plist:
-                raise ValueError(f"No provider configured for API {api_str}?")
-
-            logger.info("Configuring API", api=api_str)
-            updated_providers = []
-            for i, provider in enumerate(plist):
-                if i >= 1:
-                    others = ", ".join(p.provider_type for p in plist[i:])
-                    logger.info(
-                        "Not configuring other providers () interactively. Please edit the resulting YAML directly.\n",
-                        others=others,
-                    )
-                    break
-
-                logger.info("Configuring provider", provider_type=provider.provider_type)
-                pid = provider.provider_type.split("::")[-1]
-                updated_providers.append(
-                    configure_single_provider(
-                        provider_registry[api],
-                        Provider(
-                            provider_id=(f"{pid}-{i:02d}" if len(plist) > 1 else pid),
-                            provider_type=provider.provider_type,
-                            config={},
-                        ),
-                    )
-                )
-                logger.info("")
-
-        config.providers[api_str] = updated_providers
-
-    return config
-
-
 def upgrade_from_routing_table(
     config_dict: dict[str, Any],
 ) -> dict[str, Any]:

src/ogx/core/configure.py

@@ -9,7 +9,6 @@
 import inspect
 from typing import Any
 
-from ogx.core.client import get_client_impl
 from ogx.core.datatypes import (
     AccessRule,
     AutoRoutedProviderSpec,
@@ -41,7 +40,6 @@
     ModelsProtocolPrivate,
     Prompts,
     ProviderSpec,
-    RemoteProviderConfig,
     RemoteProviderSpec,
     Responses,
     Safety,
@@ -519,38 +517,3 @@ def check_protocol_compliance(obj: Any, protocol: Any) -> None:
         raise ValueError(
             f"Provider `{obj.__provider_id__} ({obj.__provider_spec__.api})` does not implement the following methods:\n{missing_methods}"
         )
-
-
-async def resolve_remote_stack_impls(
-    config: RemoteProviderConfig,
-    apis: list[str],
-) -> dict[Api, Any]:
-    """Resolve provider implementations for a remote stack by creating API clients.
-
-    Args:
-        config: Remote provider configuration containing the connection URL.
-        apis: List of API names to resolve.
-
-    Returns:
-        Dictionary mapping APIs to their remote client implementations.
-    """
-    protocols = api_protocol_map()
-    additional_protocols = additional_protocols_map()
-
-    impls = {}
-    for api_str in apis:
-        api = Api(api_str)
-        impls[api] = await get_client_impl(
-            protocols[api],
-            config,
-            {},
-        )
-        if api in additional_protocols:
-            _, additional_protocol, additional_api = additional_protocols[api]
-            impls[additional_api] = await get_client_impl(
-                additional_protocol,
-                config,
-                {},
-            )
-
-    return impls

src/ogx/core/resolver.py

@@ -4,51 +4,6 @@
 # This source code is licensed under the terms described in the LICENSE file in
 # the root directory of this source tree.
 
-import importlib
-import os
-import signal
-import subprocess
-import sys
-
-from termcolor import cprint
-
-from ogx.log import get_logger
-
-log = get_logger(name=__name__, category="core")
-
-
-def formulate_run_args(image_type: str, distro_name: str) -> list:
-    """Build the command-line arguments for starting a OGX server.
-
-    Args:
-        image_type: The image type (e.g., 'venv', 'container').
-        distro_name: The distribution or virtual environment name.
-
-    Returns:
-        A list of command-line arguments, or an empty list if no environment is detected.
-    """
-    # Only venv is supported now
-    current_venv = os.environ.get("VIRTUAL_ENV")
-    env_name = distro_name or current_venv
-    if not env_name:
-        cprint(
-            "No current virtual environment detected, please specify a virtual environment name with --image-name",
-            color="red",
-            file=sys.stderr,
-        )
-        return []
-
-    cprint(f"Using virtual environment: {env_name}", file=sys.stderr)
-
-    script = importlib.resources.files("ogx") / "core/start_stack.sh"
-    run_args = [
-        script,
-        image_type,
-        env_name,
-    ]
-
-    return run_args
-
 
 def in_notebook():
     """Detect whether the current code is running inside a Jupyter notebook.
@@ -67,53 +22,3 @@ def in_notebook():
     except AttributeError:
         return False
     return True
-
-
-def run_command(command: list[str]) -> int:
-    """
-    Run a command with interrupt handling and output capture.
-    Uses subprocess.run with direct stream piping for better performance.
-
-    Args:
-        command (list): The command to run.
-
-    Returns:
-        int: The return code of the command.
-    """
-    original_sigint = signal.getsignal(signal.SIGINT)
-    ctrl_c_pressed = False
-
-    def sigint_handler(signum, frame):
-        nonlocal ctrl_c_pressed
-        ctrl_c_pressed = True
-        log.info("\nCtrl-C detected. Aborting...")
-
-    try:
-        # Set up the signal handler
-        signal.signal(signal.SIGINT, sigint_handler)
-
-        # Run the command with stdout/stderr piped directly to system streams
-        result = subprocess.run(
-            command,
-            text=True,
-            check=False,
-        )
-
-        # Print stdout and stderr if command failed
-        if result.returncode != 0:
-            log.error(f"Command {' '.join(command)} failed with returncode {result.returncode}")
-            if result.stdout:
-                log.error(f"STDOUT: {result.stdout}")
-            if result.stderr:
-                log.error(f"STDERR: {result.stderr}")
-
-        return result.returncode
-    except subprocess.SubprocessError as e:
-        log.error(f"Subprocess error: {e}")
-        return 1
-    except Exception as e:
-        log.exception(f"Unexpected error: {e}")
-        return 1
-    finally:
-        # Restore the original signal handler
-        signal.signal(signal.SIGINT, original_sigint)