ccproxy.plugins.codex.tasks

ccproxy.plugins.codex.tasks

Scheduled tasks for Codex plugin.

CodexDetectionRefreshTask

CodexDetectionRefreshTask(
    name,
    interval_seconds,
    detection_service,
    enabled=True,
    skip_initial_run=True,
    **kwargs,
)

Bases: BaseScheduledTask

Task to periodically refresh Codex CLI detection headers.

Parameters:

Name	Type	Description	Default
name	str	Task name	required
interval_seconds	float	Interval between refreshes	required
detection_service	CodexDetectionService	The Codex detection service to refresh	required
enabled	bool	Whether the task is enabled	True
skip_initial_run	bool	Whether to skip the initial run at startup	True
**kwargs	Any	Additional arguments for BaseScheduledTask	{}

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

def __init__(
    self,
    name: str,
    interval_seconds: float,
    detection_service: "CodexDetectionService",
    enabled: bool = True,
    skip_initial_run: bool = True,
    **kwargs: Any,
) -> None:
    """Initialize the Codex detection refresh task.

    Args:
        name: Task name
        interval_seconds: Interval between refreshes
        detection_service: The Codex detection service to refresh
        enabled: Whether the task is enabled
        skip_initial_run: Whether to skip the initial run at startup
        **kwargs: Additional arguments for BaseScheduledTask
    """
    super().__init__(
        name=name,
        interval_seconds=interval_seconds,
        enabled=enabled,
        **kwargs,
    )
    self.detection_service = detection_service
    self.skip_initial_run = skip_initial_run
    self._first_run = True

run async

run()

Execute the detection refresh.

Returns:

Type	Description
bool	True if refresh was successful, False otherwise

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

async def run(self) -> bool:
    """Execute the detection refresh.

    Returns:
        True if refresh was successful, False otherwise
    """
    # Skip the first run if configured to do so
    if self._first_run and self.skip_initial_run:
        self._first_run = False
        logger.debug(
            "codex_detection_refresh_skipped_initial",
            task_name=self.name,
            reason="Initial run skipped to avoid duplicate detection at startup",
        )
        return True  # Return success to avoid triggering backoff

    self._first_run = False

    try:
        logger.info(
            "codex_detection_refresh_starting",
            task_name=self.name,
        )

        # Refresh the detection data
        detection_data = await self.detection_service.initialize_detection()

        logger.info(
            "codex_detection_refresh_completed",
            task_name=self.name,
            version=detection_data.codex_version if detection_data else "unknown",
            has_cached_data=detection_data is not None,
        )

        return True

    except Exception as e:
        logger.error(
            "codex_detection_refresh_failed",
            task_name=self.name,
            error=str(e),
            error_type=type(e).__name__,
        )
        return False

setup async

setup()

Perform any setup required before task execution starts.

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

async def setup(self) -> None:
    """Perform any setup required before task execution starts."""
    logger.debug(
        "codex_detection_refresh_setup",
        task_name=self.name,
        interval_seconds=self.interval_seconds,
    )

cleanup async

cleanup()

Perform any cleanup required after task execution stops.

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

async def cleanup(self) -> None:
    """Perform any cleanup required after task execution stops."""
    logger.info(
        "codex_detection_refresh_cleanup",
        task_name=self.name,
    )