ccproxy.plugins.codex.detection_service

ccproxy.plugins.codex.detection_service

Service for detecting Codex CLI using centralized detection.

CodexDetectionService

CodexDetectionService(
    settings,
    cli_service=None,
    codex_settings=None,
    redact_sensitive_cache=True,
)

Service for automatically detecting Codex CLI headers at startup.

Parameters:

Name	Type	Description	Default
settings	Settings	Application settings	required
cli_service	CLIDetectionService | None	Optional CLI detection service for dependency injection. If None, creates its own instance.	None
codex_settings	CodexSettings | None	Optional Codex plugin settings for plugin-specific configuration. If None, uses default configuration.	None

Source code in ccproxy/plugins/codex/detection_service.py

def __init__(
    self,
    settings: Settings,
    cli_service: CLIDetectionService | None = None,
    codex_settings: CodexSettings | None = None,
    redact_sensitive_cache: bool = True,
) -> None:
    """Initialize Codex detection service.

    Args:
        settings: Application settings
        cli_service: Optional CLI detection service for dependency injection.
                    If None, creates its own instance.
        codex_settings: Optional Codex plugin settings for plugin-specific configuration.
                       If None, uses default configuration.
    """
    self.settings = settings
    self.codex_settings = codex_settings if codex_settings else CodexSettings()
    self.cache_dir = get_ccproxy_cache_dir()
    self.cache_dir.mkdir(parents=True, exist_ok=True)
    self._cached_data: CodexCacheData | None = None
    self._cli_service = cli_service or CLIDetectionService(settings)
    self._cli_info: CodexCliInfo | None = None
    self._redact_sensitive_cache = redact_sensitive_cache

initialize_detection async

initialize_detection()

Initialize Codex detection at startup.

Source code in ccproxy/plugins/codex/detection_service.py

async def initialize_detection(self) -> CodexCacheData:
    """Initialize Codex detection at startup."""
    try:
        # Get current Codex version
        current_version = await self._get_codex_version()

        detected_data = None
        # Try to load from cache first
        cached = False
        try:
            detected_data = self._load_from_cache(current_version)
            cached = detected_data is not None
        except Exception as e:
            logger.warning(
                "invalid_cache_file",
                error=str(e),
                category="plugin",
                exc_info=e,
            )

        if not cached:
            # No cache or version changed - detect fresh
            detected_data = await self._detect_codex_headers(current_version)
            # Cache the results
            self._save_to_cache(detected_data)

        self._cached_data = detected_data

        logger.trace(
            "detection_headers_completed",
            version=current_version,
            cached=cached,
        )

        if detected_data is None:
            raise ValueError("Codex detection failed")
        return detected_data

    except Exception as e:
        logger.warning(
            "detection_codex_headers_failed",
            fallback=True,
            exc_info=e,
            category="plugin",
        )
        # Return fallback data
        fallback_data = self._get_fallback_data()
        self._cached_data = fallback_data
        return fallback_data

get_cached_data

get_cached_data()

Get currently cached detection data.

Source code in ccproxy/plugins/codex/detection_service.py

def get_cached_data(self) -> CodexCacheData | None:
    """Get currently cached detection data."""
    return self._cached_data

get_detected_headers

get_detected_headers()

Return cached headers as structured data.

Source code in ccproxy/plugins/codex/detection_service.py

def get_detected_headers(self) -> DetectedHeaders:
    """Return cached headers as structured data."""

    data = self.get_cached_data()
    if not data:
        return DetectedHeaders()
    return data.headers

get_detected_prompts

get_detected_prompts()

Return cached prompt metadata as structured data.

Source code in ccproxy/plugins/codex/detection_service.py

def get_detected_prompts(self) -> DetectedPrompts:
    """Return cached prompt metadata as structured data."""

    data = self.get_cached_data()
    if not data:
        return DetectedPrompts()
    return data.prompts

get_ignored_headers

get_ignored_headers()

Headers that should be ignored when forwarding CLI values.

Source code in ccproxy/plugins/codex/detection_service.py

def get_ignored_headers(self) -> list[str]:
    """Headers that should be ignored when forwarding CLI values."""

    return list(self.ignores_header)

get_redacted_headers

get_redacted_headers()

Headers that must always be removed before forwarding.

Source code in ccproxy/plugins/codex/detection_service.py

def get_redacted_headers(self) -> list[str]:
    """Headers that must always be removed before forwarding."""

    return list(getattr(self, "REDACTED_HEADERS", []))

get_version

get_version()

Get the Codex CLI version.

Returns:

Type	Description
str	Version string or "unknown" if not available

Source code in ccproxy/plugins/codex/detection_service.py

def get_version(self) -> str:
    """Get the Codex CLI version.

    Returns:
        Version string or "unknown" if not available
    """
    data = self.get_cached_data()
    return data.codex_version if data else "unknown"

get_cli_path

get_cli_path()

Get the Codex CLI command with caching.

Returns:

Type	Description
list[str] | None	Command list to execute Codex CLI if found, None otherwise

Source code in ccproxy/plugins/codex/detection_service.py

def get_cli_path(self) -> list[str] | None:
    """Get the Codex CLI command with caching.

    Returns:
        Command list to execute Codex CLI if found, None otherwise
    """
    info = self._cli_service.get_cli_info("codex")
    return info["command"] if info["is_available"] else None

get_binary_path

get_binary_path()

Alias for get_cli_path for backward compatibility.

Source code in ccproxy/plugins/codex/detection_service.py

def get_binary_path(self) -> list[str] | None:
    """Alias for get_cli_path for backward compatibility."""
    return self.get_cli_path()

get_cli_health_info

get_cli_health_info()

Get lightweight CLI health info using centralized detection, cached locally.

Returns:

Type	Description
CodexCliInfo	CodexCliInfo with availability, version, and binary path

Source code in ccproxy/plugins/codex/detection_service.py

def get_cli_health_info(self) -> CodexCliInfo:
    """Get lightweight CLI health info using centralized detection, cached locally.

    Returns:
        CodexCliInfo with availability, version, and binary path
    """
    from .models import CodexCliInfo, CodexCliStatus

    if self._cli_info is not None:
        return self._cli_info

    info = self._cli_service.get_cli_info("codex")
    status = (
        CodexCliStatus.AVAILABLE
        if info["is_available"]
        else CodexCliStatus.NOT_INSTALLED
    )
    cli_info = CodexCliInfo(
        status=status,
        version=info.get("version"),
        binary_path=info.get("path"),
    )
    self._cli_info = cli_info
    return cli_info

invalidate_cache

invalidate_cache()

Clear all cached detection data.

Source code in ccproxy/plugins/codex/detection_service.py

def invalidate_cache(self) -> None:
    """Clear all cached detection data."""
    # Clear the async cache for _get_codex_version
    if hasattr(self._get_codex_version, "cache_clear"):
        self._get_codex_version.cache_clear()
    self._cli_info = None
    logger.debug("detection_cache_cleared", category="plugin")

get_system_prompt

get_system_prompt(mode=None)

Return an instructions dict for injection based on cached prompts.

Source code in ccproxy/plugins/codex/detection_service.py

def get_system_prompt(self, mode: str | None = None) -> dict[str, Any]:
    """Return an instructions dict for injection based on cached prompts."""
    prompts = self.get_detected_prompts()
    return prompts.instructions_payload()