Function queue_worker

api_utils/queue_worker.py:21–467

Queue worker, processes tasks in the request queue

()

async def queue_worker() -> None:
    """Queue worker, processes tasks in the request queue"""
    # Delayed imports to avoid circularity
    from api_utils.server_state import state
    from config import RESPONSE_COMPLETION_TIMEOUT

    logger = state.logger
    request_queue = state.request_queue
    processing_lock = state.processing_lock
    model_switching_lock = state.model_switching_lock
    params_cache_lock = state.params_cache_lock
    from browser_utils.auth_rotation import perform_auth_rotation
    from browser_utils.page_controller import PageController
    from config.global_state import GlobalState

    from .error_utils import (
        client_cancelled,
        client_disconnected,
        server_error,
    )

    # Internal imports for queue worker logic
    from .request_processor import (
        ClientDisconnectedError,
        _process_request_refactored,
        _test_client_connection,
        save_error_snapshot,
    )
    from .utils_ext.stream import clear_stream_queue

    logger.info("--- Queue Worker Started ---")

    # Validate that required globals are initialized
    if request_queue is None:
        logger.critical("FATAL: request_queue is None! Initialization failed.")
        raise RuntimeError("request_queue not initialized")

    if processing_lock is None:
        logger.critical("FATAL: processing_lock is None! Initialization failed.")
        raise RuntimeError("processing_lock not initialized")

    if model_switching_lock is None:
        logger.critical("FATAL: model_switching_lock is None! Initialization failed.")
        raise RuntimeError("model_switching_lock not initialized")

    if params_cache_lock is None:
        logger.critical("FATAL: params_cache_lock is None! Initialization failed.")
        raise RuntimeError("params_cache_lock not initialized")

    logger.debug(
        f"Queue worker initialized with queue={request_queue}, lock={processing_lock}"
    )

    was_last_request_streaming = False
    last_request_completion_time = 0.0
    shutdown_check_interval = 0.1

    while True: