feat(warmup): add sliding usage window warm-up scheduler with multi-account fan-out

Configurable scheduled lightweight LLM requests to shift provider rate-limit
windows. Supports explicit backend:model routes, weekend control, random jitter,
retry-once for transient errors, and openai-codex multi-account fan-out.

Author: Mateusz

src/connectors/_openai_codex_connector.py

@@ -647,6 +647,55 @@ def _configure_managed_oauth_credential_manager(self) -> None:
                     exc_info=True,
                 )
 
+    async def list_managed_oauth_account_ids(self) -> list[str]:
+        """Return eligible managed OAuth account IDs for warm-up fan-out.
+
+        Accounts requiring re-auth are excluded. Allowed account filtering configured
+        in managed OAuth settings is applied by delegating to the credential manager.
+        """
+        selector = getattr(self._credential_manager, "_managed_selector", None)
+        if selector is None:
+            return []
+
+        try:
+            reload_accounts = getattr(selector, "reload_accounts", None)
+            if callable(reload_accounts):
+                result = reload_accounts()
+                if inspect.isawaitable(result):
+                    await result
+
+            now_ms = int(time.time() * 1000)
+            available_accounts_fn = getattr(selector, "_available_accounts", None)
+            if callable(available_accounts_fn):
+                maybe_accounts = available_accounts_fn(now_ms)
+                if (
+                    not isinstance(maybe_accounts, tuple)
+                    or len(maybe_accounts) != 2
+                    or not isinstance(maybe_accounts[0], Sequence)
+                    or not isinstance(maybe_accounts[1], Sequence)
+                ):
+                    return []
+                available, eligible = maybe_accounts
+                if available and logger.isEnabledFor(logging.DEBUG):
+                    logger.debug(
+                        "OpenAI Codex warm-up account fan-out candidates: %d available, %d eligible",
+                        len(available),
+                        len(eligible),
+                    )
+                return [
+                    account.account_id
+                    for account in eligible
+                    if isinstance(account.account_id, str) and account.account_id
+                ]
+        except Exception as exc:
+            if logger.isEnabledFor(logging.WARNING):
+                logger.warning(
+                    "Failed to enumerate managed OAuth accounts for warm-up fan-out: %s",
+                    exc,
+                    exc_info=True,
+                )
+        return []
+
     @property
     def _auth_credentials(self) -> dict[str, Any] | None:
         return getattr(self._credential_manager, "_auth_credentials", None)

src/connectors/openai_codex/credentials.py

@@ -1100,7 +1100,7 @@ def validate_current_credentials(self) -> ValidationResult:
             return ValidationResult.failure("No credentials loaded")
         return self._validate_credentials_structure(self._auth_credentials)
 
-    def get_account_id(self) -> str | None:
+    def get_account_id(self) -> str | None:
         """Return ChatGPT account ID from loaded credentials.
 
         Returns:
@@ -1148,7 +1148,15 @@ def get_account_id(self) -> str | None:
             if isinstance(token_account_id, str) and token_account_id:
                 return token_account_id
 
-        return None
+        return None
+
+    async def list_managed_oauth_account_ids(self) -> list[str]:
+        """Return eligible managed OAuth account IDs for warm-up fan-out."""
+        if not self._managed_enabled():
+            return []
+
+        await self._managed_selector.reload_accounts()
+        return await self._managed_selector.list_eligible_account_ids()
 
 
 def _extract_chatgpt_account_id_from_jwt(token: str) -> str | None:

src/connectors/openai_codex/interfaces.py

@@ -160,6 +160,11 @@ def get_account_id(self) -> str | None:
         """
         ...
 
+    @abstractmethod
+    async def list_managed_oauth_account_ids(self) -> list[str]:
+        """Return eligible managed OAuth account IDs for fan-out warm-up calls."""
+        ...
+
 
 class IPayloadBuilder(ABC):
     """Interface for building Codex payloads.

src/connectors/openai_codex/managed_oauth_selector.py

@@ -127,9 +127,18 @@ def _available_accounts(
             if account.needs_reauth:
                 continue
             available.append(account)
-        eligible = [account for account in available if not account.is_rate_limited(now_ms)]
+        eligible = [
+            account for account in available if not account.is_rate_limited(now_ms)
+        ]
         return available, eligible
 
+    async def list_eligible_account_ids(self) -> list[str]:
+        """Return currently eligible account IDs after reload/refresh gating."""
+        await self._ensure_accounts_loaded()
+        now_ms = int(time.time() * 1000)
+        _, eligible = self._available_accounts(now_ms)
+        return [account.account_id for account in eligible]
+
     def _select_by_strategy(
         self,
         eligible: list[ManagedOAuthAccount],
@@ -333,4 +342,3 @@ async def rotate_on_auth_failure(
             session_id=session_id,
             ignore_session_affinity=True,
         )
-

src/core/app/application_builder.py

@@ -630,6 +630,9 @@ async def lifespan(app: FastAPI):  # type: ignore[no-untyped-def,no-any-return,m
                 from src.core.services.backend_startup_disablement import (
                     apply_backend_disablement_at_startup,
                 )
+                from src.core.services.usage_window_warmup_service import (
+                    UsageWindowWarmupService,
+                )
 
                 app_config = service_provider.get_service(AppConfig)
                 backend_lifecycle_manager = service_provider.get_service(
@@ -644,6 +647,9 @@ async def lifespan(app: FastAPI):  # type: ignore[no-untyped-def,no-any-return,m
                     )
 
                 routing_service = service_provider.get_service(BackendRoutingService)
+                usage_window_warmup_service = service_provider.get_service(
+                    UsageWindowWarmupService
+                )
                 if routing_service is not None:
                     active_routing_service = routing_service
 
@@ -665,7 +671,7 @@ async def _startup_capability_refresh() -> None:
                     )
 
                     refresh_interval_seconds = 0.0
-                    if app_config is not None and app_config.routing is not None:
+                    if app_config is not None:
                         refresh_interval_seconds = (
                             app_config.routing.capability_refresh_interval_seconds
                             or 0.0
@@ -674,6 +680,8 @@ async def _startup_capability_refresh() -> None:
                         _ = asyncio.create_task(  # noqa: RUF006 - fire-and-forget
                             active_routing_service.start_model_capability_refresh()
                         )
+                if usage_window_warmup_service is not None:
+                    await usage_window_warmup_service.start()
             except Exception as exc:
                 if logger.isEnabledFor(logging.WARNING):
                     logger.warning(
@@ -690,6 +698,31 @@ async def _startup_capability_refresh() -> None:
                 logger.info("Shutting down application")
 
             # Stop capability refresh loop before backend shutdown.
+            try:
+                from src.core.services.usage_window_warmup_service import (
+                    UsageWindowWarmupService,
+                )
+
+                usage_window_warmup_service = service_provider.get_service(
+                    UsageWindowWarmupService
+                )
+                if usage_window_warmup_service is not None:
+                    await usage_window_warmup_service.stop()
+            except (RuntimeError, AttributeError, asyncio.CancelledError) as exc:
+                if logger.isEnabledFor(logging.WARNING):
+                    logger.warning(
+                        "Failed to stop usage window warm-up scheduler: %s",
+                        type(exc).__name__,
+                        exc_info=True,
+                    )
+            except Exception as exc:
+                if logger.isEnabledFor(logging.WARNING):
+                    logger.warning(
+                        "Failed to stop usage window warm-up scheduler: %s",
+                        type(exc).__name__,
+                        exc_info=True,
+                    )
+
             try:
                 from src.core.services.backend_routing_service import (
                     BackendRoutingService,

src/core/app/stages/core_services.py

@@ -19,8 +19,15 @@
 
 from src.core.config.app_config import AppConfig
 from src.core.di.container import ServiceCollection
+from src.core.di.registrations._shared import register_singleton_if_absent
 from src.core.interfaces.application_state_interface import IApplicationState
-from src.core.interfaces.di_interface import IServiceProvider
+from src.core.interfaces.backend_completion_flow_interface import (
+    IBackendCompletionFlow,
+)
+from src.core.interfaces.backend_lifecycle_manager_interface import (
+    IBackendLifecycleManager,
+)
+from src.core.interfaces.di_interface import IServiceProvider
 from src.core.interfaces.response_parser_interface import IResponseParser
 
 # from src.core.interfaces.secure_state_interface import ISecureStateService # Removed unresolved import
@@ -323,6 +330,9 @@ def secure_state_factory(provider: IServiceProvider) -> SecureStateService:
         # Register connection activity tracker (if enabled)
         self._register_activity_tracker(services, config)
 
+        # Register usage window warm-up scheduler (if enabled)
+        self._register_usage_window_warmup_service(services, config)
+
         # Register wire capture service
         self._register_wire_capture_service(services)
 
@@ -475,6 +485,44 @@ def cleanup_scheduler_factory(
         if logger.isEnabledFor(logging.INFO):
             logger.info("Connection tracker cleanup scheduler registered")
 
+    def _register_usage_window_warmup_service(
+        self, services: ServiceCollection, config: AppConfig
+    ) -> None:
+        if not config.usage_window_warmup.enabled:
+            if logger.isEnabledFor(logging.DEBUG):
+                logger.debug("Usage window warm-up scheduler disabled")
+            return
+
+        from src.core.services.usage_window_warmup_service import (
+            UsageWindowWarmupService,
+        )
+
+        def warmup_service_factory(
+            provider: IServiceProvider,
+        ) -> UsageWindowWarmupService:
+            completion_flow = provider.get_required_service(
+                IBackendCompletionFlow  # type: ignore[type-abstract]
+            )
+            backend_lifecycle_manager = provider.get_service(
+                IBackendLifecycleManager  # type: ignore[type-abstract]
+            )
+            from src.core.services.warmup_target_resolver import WarmupTargetResolver
+
+            target_resolver = WarmupTargetResolver(backend_lifecycle_manager)
+            return UsageWindowWarmupService(
+                completion_flow=completion_flow,
+                config=config.usage_window_warmup,
+                target_resolver=target_resolver,
+            )
+
+        register_singleton_if_absent(
+            services,
+            UsageWindowWarmupService,
+            implementation_factory=warmup_service_factory,
+        )
+        if logger.isEnabledFor(logging.INFO):
+            logger.info("Usage window warm-up scheduler registered")
+
     def _register_wire_capture_service(self, services: ServiceCollection) -> None:
         """Register wire capture service.
 

src/core/config/app_config.py

@@ -107,6 +107,7 @@ def save(self, path: str | Path) -> None:
             "usage_tracking",
             "replacement",
             "health_check",
+            "usage_window_warmup",
             "failure_handling",
             "routing",
             "dynamic_compression",

src/core/config/models/app_config_model.py

@@ -44,6 +44,9 @@
 )
 from src.core.domain.configuration.replacement_config import ReplacementConfig
 from src.core.domain.configuration.sandboxing_config import SandboxingConfiguration
+from src.core.domain.configuration.usage_window_warmup_config import (
+    UsageWindowWarmupConfig,
+)
 from src.core.domain.model_utils import ModelDefaults
 from src.core.interfaces.configuration_interface import IConfig
 from src.core.interfaces.model_bases import DomainModel
@@ -104,6 +107,9 @@ class AppConfigModel(DomainModel, IConfig):
     end_of_session: EndOfSessionConfig = Field(default_factory=EndOfSessionConfig)
     replacement: ReplacementConfig = Field(default_factory=ReplacementConfig)
     health_check: HealthCheckConfig = Field(default_factory=HealthCheckConfig)
+    usage_window_warmup: UsageWindowWarmupConfig = Field(
+        default_factory=UsageWindowWarmupConfig
+    )
     failure_handling: FailureHandlingConfig = Field(
         default_factory=FailureHandlingConfig
     )
@@ -141,18 +147,21 @@ class AppConfigModel(DomainModel, IConfig):
 
     @field_validator("auto_append_first_prompt_filename", mode="before")
     @classmethod
-    def validate_auto_append_first_prompt_filename(cls, v: str | None) -> str | None:
-        if v is None or (isinstance(v, str) and not v.strip()):
+    def validate_auto_append_first_prompt_filename(cls, v: Any) -> str | None:
+        if v is None:
             return None
         if not isinstance(v, str):
             raise ValueError("auto_append_first_prompt_filename must be a string")
-        suf = Path(v.strip()).suffix.lower()
+        stripped = v.strip()
+        if not stripped:
+            return None
+        suf = Path(stripped).suffix.lower()
         if suf not in (".txt", ".md"):
             raise ValueError(
                 f"Invalid auto_append_first_prompt_filename {v!r}: "
                 "must end with .txt or .md"
             )
-        return v.strip()
+        return stripped
 
     def model_is_functional(self, model_id: str) -> bool:
         return self.backends.model_is_functional(model_id)

src/core/domain/configuration/usage_window_warmup_config.py

@@ -0,0 +1,74 @@
+"""Configuration for scheduled sliding usage window warm-up requests."""
+
+from __future__ import annotations
+
+import re
+
+from pydantic import ConfigDict, Field, field_validator
+
+from src.core.domain.model_utils import (
+    has_explicit_backend_selector,
+    parse_model_backend,
+)
+from src.core.interfaces.model_bases import DomainModel
+
+_TIME_PATTERN = re.compile(r"^(?:[01]\d|2[0-3]):[0-5]\d$")
+
+
+class UsageWindowWarmupEntryConfig(DomainModel):
+    """Single scheduled warm-up entry for a concrete backend:model route."""
+
+    model_config = ConfigDict(frozen=True)
+
+    model: str = Field(...)
+    time: str = Field(...)
+    execute_on_weekend: bool = False
+
+    @field_validator("model")
+    @classmethod
+    def validate_model(cls, value: str) -> str:
+        trimmed = value.strip()
+        if not trimmed:
+            raise ValueError("usage window warm-up model must be provided")
+        if "^" in trimmed or "|" in trimmed:
+            raise ValueError(
+                "usage window warm-up model cannot use composite routing operators"
+            )
+        if not has_explicit_backend_selector(trimmed):
+            raise ValueError(
+                "usage window warm-up model must use an explicit backend:model route"
+            )
+
+        parsed = parse_model_backend(trimmed)
+        if not parsed.backend_type.strip() or not parsed.model_name.strip():
+            raise ValueError(
+                "usage window warm-up model must include a non-empty backend and model"
+            )
+        return trimmed
+
+    @field_validator("time")
+    @classmethod
+    def validate_time(cls, value: str) -> str:
+        trimmed = value.strip()
+        if not _TIME_PATTERN.fullmatch(trimmed):
+            raise ValueError("usage window warm-up time must use HH:MM 24-hour format")
+        return trimmed
+
+
+class UsageWindowWarmupConfig(DomainModel):
+    """Top-level warm-up scheduler configuration."""
+
+    model_config = ConfigDict(frozen=True)
+
+    enabled: bool = False
+    entries: list[UsageWindowWarmupEntryConfig] = Field(default_factory=list)
+
+
+DEFAULT_USAGE_WINDOW_WARMUP_CONFIG = UsageWindowWarmupConfig()
+
+
+__all__ = [
+    "DEFAULT_USAGE_WINDOW_WARMUP_CONFIG",
+    "UsageWindowWarmupConfig",
+    "UsageWindowWarmupEntryConfig",
+]