ccproxy.services.cli_detection

ccproxy.services.cli_detection

Centralized CLI detection service for all plugins.

This module provides a unified interface for detecting CLI binaries, checking versions, and managing CLI-related state across all plugins. It eliminates duplicate CLI detection logic by consolidating common patterns.

CLIDetectionResult

Bases: NamedTuple

Result of CLI detection for a specific binary.

CLIDetectionService

CLIDetectionService(settings, binary_resolver=None)

Centralized service for CLI detection across all plugins.

This service provides: - Unified binary detection using BinaryResolver - Version detection with caching - Fallback data support for when CLI is not available - Consistent logging and error handling

Parameters:

Name	Type	Description	Default
settings	Settings	Application settings	required
binary_resolver	BinaryResolver | None	Optional binary resolver instance. If None, creates a new one.	None

Source code in ccproxy/services/cli_detection.py

def __init__(
    self, settings: Settings, binary_resolver: BinaryResolver | None = None
) -> None:
    """Initialize the CLI detection service.

    Args:
        settings: Application settings
        binary_resolver: Optional binary resolver instance. If None, creates a new one.
    """
    self.settings = settings
    self.cache_dir = get_ccproxy_cache_dir()
    self.cache_dir.mkdir(parents=True, exist_ok=True)

    # Use injected resolver or create from settings for backward compatibility
    self.resolver = binary_resolver or BinaryResolver.from_settings(settings)

    # Enhanced TTL cache for detection results (10 minute TTL)
    self._detection_cache = TTLCache(maxsize=64, ttl=600.0)

    # Separate cache for version info (longer TTL since versions change infrequently)
    self._version_cache = TTLCache(maxsize=32, ttl=1800.0)  # 30 minutes

detect_cli async

detect_cli(
    binary_name,
    package_name=None,
    version_flag="--version",
    version_parser=None,
    fallback_data=None,
    cache_key=None,
)

Detect a CLI binary and its version.

Parameters:

Name	Type	Description	Default
binary_name	str	Name of the binary to detect (e.g., "claude", "codex")	required
package_name	str | None	NPM package name if different from binary name	None
version_flag	str	Flag to get version (default: "--version")	'--version'
version_parser	Any | None	Optional callable to parse version output	None
fallback_data	dict[str, Any] | None	Optional fallback data if CLI is not available	None
cache_key	str | None	Optional cache key (defaults to binary_name)	None

Returns:

Type	Description
CLIDetectionResult	CLIDetectionResult with detection information

Source code in ccproxy/services/cli_detection.py

async def detect_cli(
    self,
    binary_name: str,
    package_name: str | None = None,
    version_flag: str = "--version",
    version_parser: Any | None = None,
    fallback_data: dict[str, Any] | None = None,
    cache_key: str | None = None,
) -> CLIDetectionResult:
    """Detect a CLI binary and its version.

    Args:
        binary_name: Name of the binary to detect (e.g., "claude", "codex")
        package_name: NPM package name if different from binary name
        version_flag: Flag to get version (default: "--version")
        version_parser: Optional callable to parse version output
        fallback_data: Optional fallback data if CLI is not available
        cache_key: Optional cache key (defaults to binary_name)

    Returns:
        CLIDetectionResult with detection information
    """
    cache_key = cache_key or binary_name

    # Check TTL cache first
    cached_result = self._detection_cache.get(cache_key)
    if cached_result is not None:
        logger.debug(
            "cli_detection_cached",
            binary=binary_name,
            version=cached_result.version,
            available=cached_result.is_available,
            cache_hit=True,
        )
        return cached_result  # type: ignore[no-any-return]

    # Try to detect the binary
    result = self.resolver.find_binary(binary_name, package_name)

    if result:
        # Binary found - get version
        version = await self._get_cli_version(
            result.command, version_flag, version_parser
        )

        # Determine source
        source = "path" if result.is_direct else "package_manager"

        detection_result = CLIDetectionResult(
            name=binary_name,
            version=version,
            command=result.command,
            is_available=True,
            source=source,
            package_manager=result.package_manager,
            cached=False,
        )

        logger.debug(
            "cli_detection_success",
            binary=binary_name,
            version=version,
            source=source,
            package_manager=result.package_manager,
            command=result.command,
            cached=cached_result is not None,
        )

    elif fallback_data:
        # Use fallback data
        detection_result = CLIDetectionResult(
            name=binary_name,
            version=fallback_data.get("version", "unknown"),
            command=None,
            is_available=False,
            source="fallback",
            package_manager=None,
            cached=False,
            fallback_data=fallback_data,
        )

        logger.warning(
            "cli_detection_using_fallback",
            binary=binary_name,
            reason="CLI not found",
        )

    else:
        # Not found and no fallback
        detection_result = CLIDetectionResult(
            name=binary_name,
            version=None,
            command=None,
            is_available=False,
            source="unknown",
            package_manager=None,
            cached=False,
        )

        logger.error(
            "cli_detection_failed",
            binary=binary_name,
            package=package_name,
        )

    # Cache the result with TTL
    self._detection_cache.set(cache_key, detection_result)

    return detection_result

load_cached_version

load_cached_version(binary_name, cache_file=None)

Load cached version for a binary.

Parameters:

Name	Type	Description	Default
binary_name	str	Name of the binary	required
cache_file	str | None	Optional cache file name	None

Returns:

Type	Description
str | None	Cached version string or None

Source code in ccproxy/services/cli_detection.py

def load_cached_version(
    self, binary_name: str, cache_file: str | None = None
) -> str | None:
    """Load cached version for a binary.

    Args:
        binary_name: Name of the binary
        cache_file: Optional cache file name

    Returns:
        Cached version string or None
    """
    cache_file_name = cache_file or f"{binary_name}_version.json"
    cache_path = self.cache_dir / cache_file_name

    if not cache_path.exists():
        return None

    try:
        with cache_path.open("r") as f:
            data = json.load(f)
            version = data.get("version")
            return str(version) if version is not None else None
    except Exception as e:
        logger.debug("cache_load_error", file=str(cache_path), error=str(e))
        return None

save_cached_version

save_cached_version(
    binary_name,
    version,
    cache_file=None,
    additional_data=None,
)

Save version to cache.

Parameters:

Name	Type	Description	Default
binary_name	str	Name of the binary	required
version	str	Version string to cache	required
cache_file	str | None	Optional cache file name	None
additional_data	dict[str, Any] | None	Additional data to cache	None

Source code in ccproxy/services/cli_detection.py

def save_cached_version(
    self,
    binary_name: str,
    version: str,
    cache_file: str | None = None,
    additional_data: dict[str, Any] | None = None,
) -> None:
    """Save version to cache.

    Args:
        binary_name: Name of the binary
        version: Version string to cache
        cache_file: Optional cache file name
        additional_data: Additional data to cache
    """
    cache_file_name = cache_file or f"{binary_name}_version.json"
    cache_path = self.cache_dir / cache_file_name

    try:
        data = {"binary": binary_name, "version": version}
        if additional_data:
            data.update(additional_data)

        with cache_path.open("w") as f:
            json.dump(data, f, indent=2)

        logger.debug("cache_saved", file=str(cache_path), version=version)
    except Exception as e:
        logger.warning("cache_save_error", file=str(cache_path), error=str(e))

get_cli_info

get_cli_info(binary_name)

Get CLI information in standard format.

Parameters:

Name	Type	Description	Default
binary_name	str	Name of the binary	required

Returns:

Type	Description
CLIInfo	CLIInfo dictionary with structured information

Source code in ccproxy/services/cli_detection.py

def get_cli_info(self, binary_name: str) -> CLIInfo:
    """Get CLI information in standard format.

    Args:
        binary_name: Name of the binary

    Returns:
        CLIInfo dictionary with structured information
    """
    # Check if we have cached detection result
    cached_result = self._detection_cache.get(binary_name)
    if cached_result is not None:
        return CLIInfo(
            name=cached_result.name,
            version=cached_result.version,
            source=cached_result.source,
            path=cached_result.command[0] if cached_result.command else None,
            command=cached_result.command or [],
            package_manager=cached_result.package_manager,
            is_available=cached_result.is_available,
        )

    # Fall back to resolver
    return self.resolver.get_cli_info(binary_name)

clear_cache

clear_cache()

Clear all detection caches.

Source code in ccproxy/services/cli_detection.py

def clear_cache(self) -> None:
    """Clear all detection caches."""
    self._detection_cache.clear()
    self._version_cache.clear()
    self.resolver.clear_cache()
    logger.debug("cli_detection_cache_cleared")

get_all_detected

get_all_detected()

Get all detected CLI binaries.

Returns:

Type	Description
dict[str, CLIDetectionResult]	Dictionary of binary name to detection result

Source code in ccproxy/services/cli_detection.py

def get_all_detected(self) -> dict[str, CLIDetectionResult]:
    """Get all detected CLI binaries.

    Returns:
        Dictionary of binary name to detection result
    """
    # Extract all cached results from TTLCache
    results: dict[str, CLIDetectionResult] = {}
    if hasattr(self._detection_cache, "_cache"):
        for key, (result, _) in self._detection_cache._cache.items():
            if isinstance(key, str) and isinstance(result, CLIDetectionResult):
                results[key] = result
    return results

detect_multiple async

detect_multiple(binaries, parallel=True)

Detect multiple CLI binaries.

Parameters:

Name	Type	Description	Default
binaries	list[tuple[str, str | None]]	List of (binary_name, package_name) tuples	required
parallel	bool	Whether to detect in parallel	True

Returns:

Type	Description
dict[str, CLIDetectionResult]	Dictionary of binary name to detection result

Source code in ccproxy/services/cli_detection.py

async def detect_multiple(
    self,
    binaries: list[tuple[str, str | None]],
    parallel: bool = True,
) -> dict[str, CLIDetectionResult]:
    """Detect multiple CLI binaries.

    Args:
        binaries: List of (binary_name, package_name) tuples
        parallel: Whether to detect in parallel

    Returns:
        Dictionary of binary name to detection result
    """
    if parallel:
        # Detect in parallel
        tasks = [
            self.detect_cli(binary_name, package_name)
            for binary_name, package_name in binaries
        ]
        results = await asyncio.gather(*tasks, return_exceptions=True)

        detected: dict[str, CLIDetectionResult] = {}
        for (binary_name, _), result in zip(binaries, results, strict=False):
            if isinstance(result, Exception):
                logger.error(
                    "cli_detection_error",
                    binary=binary_name,
                    error=str(result),
                )
            elif isinstance(result, CLIDetectionResult):
                detected[binary_name] = result

        return detected
    else:
        # Detect sequentially
        detected = {}
        for binary_name, package_name in binaries:
            try:
                result = await self.detect_cli(binary_name, package_name)
                detected[binary_name] = result
            except Exception as e:
                logger.error(
                    "cli_detection_error",
                    binary=binary_name,
                    error=str(e),
                )

        return detected