Compare commits

...

1 Commits

Author SHA1 Message Date
Alexander Whitestone
e6c346c6bb feat: time-aware model routing for cron jobs (#317)
Some checks failed
Forge CI / smoke-and-build (pull_request) Failing after 1m5s
Route cron jobs to different models based on time of day.

- Peak hours (user active): cheaper model, user catches errors
- Off-peak hours (user absent): stronger model, errors go uncorrected

Config under smart_model_routing.cron_time_routing:
  enabled, timezone, peak_hours (start/end), peak_model, offpeak_model

Integrates into cron/scheduler.py via resolve_cron_turn_route(), which
checks time-aware routing first, then falls back to normal smart routing.

12 tests added (test_cron_time_routing.py). All 18 routing tests pass.

Closes #317
2026-04-13 17:30:07 -04:00
5 changed files with 340 additions and 3 deletions

View File

@@ -4,6 +4,7 @@ from __future__ import annotations
import os import os
import re import re
from datetime import datetime
from typing import Any, Dict, Optional from typing import Any, Dict, Optional
from utils import is_truthy_value from utils import is_truthy_value
@@ -182,7 +183,7 @@ def resolve_turn_route(user_message: str, routing_config: Optional[Dict[str, Any
"command": runtime.get("command"), "command": runtime.get("command"),
"args": list(runtime.get("args") or []), "args": list(runtime.get("args") or []),
}, },
"label": f"smart route {route.get('model')} ({runtime.get('provider')})", "label": f"smart route \u2192 {route.get('model')} ({runtime.get('provider')})",
"signature": ( "signature": (
route.get("model"), route.get("model"),
runtime.get("provider"), runtime.get("provider"),
@@ -192,3 +193,151 @@ def resolve_turn_route(user_message: str, routing_config: Optional[Dict[str, Any
tuple(runtime.get("args") or ()), tuple(runtime.get("args") or ()),
), ),
} }
# ---------------------------------------------------------------------------
# Time-aware cron model routing
# ---------------------------------------------------------------------------
# During peak hours (user active), cron jobs use a cheaper model because the
# user is present to catch and correct errors. During off-peak hours (user
# absent), cron jobs use a stronger model because errors go uncorrected.
#
# Config (under smart_model_routing.cron_time_routing):
# enabled: true
# timezone: "America/New_York" # IANA timezone name (default: UTC)
# peak_hours:
# start: 9 # inclusive, 0-23
# end: 18 # exclusive, 0-23
# peak_model: # model to use during peak hours
# provider: openrouter
# model: xiaomi/mimo-v2-pro
# offpeak_model: # model to use during off-peak hours
# provider: openrouter
# model: anthropic/claude-sonnet-4
def _get_current_hour_in_tz(tz_name: str) -> int:
"""Return the current hour (0-23) in the given IANA timezone."""
try:
from zoneinfo import ZoneInfo
tz = ZoneInfo(tz_name)
except Exception:
try:
import pytz
tz = pytz.timezone(tz_name)
except Exception:
return datetime.utcnow().hour
return datetime.now(tz).hour
def _is_peak_hour(hour: int, peak_start: int, peak_end: int) -> bool:
"""Return True if *hour* falls within [peak_start, peak_end).
Handles wrap-around (e.g. start=22, end=6 means 22:00-05:59 is peak).
"""
if peak_start <= peak_end:
return peak_start <= hour < peak_end
else:
# Wraps midnight: e.g. 22-6 means 22,23,0,1,2,3,4,5
return hour >= peak_start or hour < peak_end
def resolve_cron_time_route(
routing_config: Optional[Dict[str, Any]],
) -> Optional[Dict[str, Any]]:
"""Return a time-aware model override for cron jobs.
Considers the current hour in the configured timezone and picks
between a peak-hours model (cheaper, user present) and an off-peak
model (stronger, user absent, errors go uncorrected).
Returns None when time-aware routing is disabled or misconfigured,
so the caller falls through to normal routing.
"""
cfg = routing_config or {}
cron_cfg = cfg.get("cron_time_routing") or {}
if not _coerce_bool(cron_cfg.get("enabled"), False):
return None
tz_name = str(cron_cfg.get("timezone", "UTC")).strip()
peak = cron_cfg.get("peak_hours") or {}
peak_start = _coerce_int(peak.get("start"), 9)
peak_end = _coerce_int(peak.get("end"), 18)
current_hour = _get_current_hour_in_tz(tz_name)
is_peak = _is_peak_hour(current_hour, peak_start, peak_end)
if is_peak:
model_cfg = cron_cfg.get("peak_model") or {}
reason = "cron_peak_hours"
else:
model_cfg = cron_cfg.get("offpeak_model") or {}
reason = "cron_offpeak_hours"
provider = str(model_cfg.get("provider") or "").strip().lower()
model = str(model_cfg.get("model") or "").strip()
if not provider or not model:
return None
return {
"provider": provider,
"model": model,
"base_url": model_cfg.get("base_url", ""),
"api_key_env": model_cfg.get("api_key_env", ""),
"routing_reason": reason,
"is_peak_hour": is_peak,
"hour": current_hour,
}
def resolve_cron_turn_route(
user_message: str,
routing_config: Optional[Dict[str, Any]],
primary: Dict[str, Any],
) -> Dict[str, Any]:
"""Resolve model route for a cron job turn with time-awareness.
Checks time-aware routing first (cron_time_routing), then falls
back to normal smart routing, then falls back to primary.
"""
# 1. Time-aware cron routing (peak vs off-peak)
time_route = resolve_cron_time_route(routing_config)
if time_route:
from hermes_cli.runtime_provider import resolve_runtime_provider
explicit_api_key = None
api_key_env = str(time_route.get("api_key_env") or "").strip()
if api_key_env:
explicit_api_key = os.getenv(api_key_env) or None
try:
runtime = resolve_runtime_provider(
requested=time_route.get("provider"),
explicit_api_key=explicit_api_key,
explicit_base_url=time_route.get("base_url"),
)
peak_label = "peak" if time_route.get("is_peak_hour") else "off-peak"
return {
"model": time_route.get("model"),
"runtime": {
"api_key": runtime.get("api_key"),
"base_url": runtime.get("base_url"),
"provider": runtime.get("provider"),
"api_mode": runtime.get("api_mode"),
"command": runtime.get("command"),
"args": list(runtime.get("args") or []),
},
"label": f"cron {peak_label} -> {time_route.get('model')} ({runtime.get('provider')})",
"signature": (
time_route.get("model"),
runtime.get("provider"),
runtime.get("base_url"),
runtime.get("api_mode"),
runtime.get("command"),
tuple(runtime.get("args") or ()),
),
}
except Exception:
pass # Fall through to normal routing
# 2. Normal smart routing (simple-turn cheap model)
return resolve_turn_route(user_message, routing_config, primary)

View File

@@ -87,6 +87,21 @@ model:
# cheap_model: # cheap_model:
# provider: openrouter # provider: openrouter
# model: google/gemini-2.5-flash # model: google/gemini-2.5-flash
# # Time-aware cron routing: pick model based on hour of day.
# # Peak hours = user present, cheaper model OK (they catch errors).
# # Off-peak = user absent, stronger model (errors go uncorrected).
# cron_time_routing:
# enabled: true
# timezone: "America/New_York" # IANA timezone (default: UTC)
# peak_hours:
# start: 9 # inclusive, 0-23
# end: 18 # exclusive, 0-23
# peak_model: # model during peak hours (user active)
# provider: openrouter
# model: xiaomi/mimo-v2-pro
# offpeak_model: # model during off-peak (user absent)
# provider: openrouter
# model: anthropic/claude-sonnet-4
# ============================================================================= # =============================================================================
# Git Worktree Isolation # Git Worktree Isolation

View File

@@ -762,8 +762,8 @@ def run_job(job: dict) -> tuple[bool, str, str, Optional[str]]:
message = format_runtime_provider_error(exc) message = format_runtime_provider_error(exc)
raise RuntimeError(message) from exc raise RuntimeError(message) from exc
from agent.smart_model_routing import resolve_turn_route from agent.smart_model_routing import resolve_cron_turn_route
turn_route = resolve_turn_route( turn_route = resolve_cron_turn_route(
prompt, prompt,
smart_routing, smart_routing,
{ {
@@ -776,6 +776,8 @@ def run_job(job: dict) -> tuple[bool, str, str, Optional[str]]:
"args": list(runtime.get("args") or []), "args": list(runtime.get("args") or []),
}, },
) )
if turn_route.get("label"):
logger.info("Job '%s': %s", job_name, turn_route["label"])
_agent_kwargs = _safe_agent_kwargs({ _agent_kwargs = _safe_agent_kwargs({
"model": turn_route["model"], "model": turn_route["model"],

View File

@@ -299,6 +299,13 @@ DEFAULT_CONFIG = {
"max_simple_chars": 160, "max_simple_chars": 160,
"max_simple_words": 28, "max_simple_words": 28,
"cheap_model": {}, "cheap_model": {},
"cron_time_routing": {
"enabled": False,
"timezone": "UTC",
"peak_hours": {"start": 9, "end": 18},
"peak_model": {},
"offpeak_model": {},
},
}, },
# Auxiliary model config — provider:model for each side task. # Auxiliary model config — provider:model for each side task.

View File

@@ -0,0 +1,164 @@
"""Tests for time-aware cron model routing."""
from agent.smart_model_routing import (
_is_peak_hour,
resolve_cron_time_route,
resolve_cron_turn_route,
)
# ---------------------------------------------------------------------------
# _is_peak_hour
# ---------------------------------------------------------------------------
def test_peak_hour_within_normal_range():
assert _is_peak_hour(10, 9, 18) is True
assert _is_peak_hour(12, 9, 18) is True
assert _is_peak_hour(17, 9, 18) is True
def test_peak_hour_outside_normal_range():
assert _is_peak_hour(8, 9, 18) is False
assert _is_peak_hour(18, 9, 18) is False
assert _is_peak_hour(22, 9, 18) is False
assert _is_peak_hour(0, 9, 18) is False
def test_peak_hour_at_boundaries():
assert _is_peak_hour(9, 9, 18) is True # start inclusive
assert _is_peak_hour(18, 9, 18) is False # end exclusive
def test_peak_hour_wraps_midnight():
# 22-6 means peak from 22:00 to 05:59
assert _is_peak_hour(22, 22, 6) is True
assert _is_peak_hour(23, 22, 6) is True
assert _is_peak_hour(0, 22, 6) is True
assert _is_peak_hour(5, 22, 6) is True
assert _is_peak_hour(6, 22, 6) is False
assert _is_peak_hour(12, 22, 6) is False
assert _is_peak_hour(21, 22, 6) is False
# ---------------------------------------------------------------------------
# resolve_cron_time_route
# ---------------------------------------------------------------------------
_CRON_ROUTING_CFG = {
"cron_time_routing": {
"enabled": True,
"timezone": "UTC",
"peak_hours": {"start": 9, "end": 18},
"peak_model": {
"provider": "openrouter",
"model": "xiaomi/mimo-v2-pro",
},
"offpeak_model": {
"provider": "openrouter",
"model": "anthropic/claude-sonnet-4",
},
},
}
def test_returns_none_when_disabled():
cfg = {"cron_time_routing": {"enabled": False}}
assert resolve_cron_time_route(cfg) is None
def test_returns_none_when_no_config():
assert resolve_cron_time_route(None) is None
assert resolve_cron_time_route({}) is None
def test_returns_none_when_models_missing():
cfg = {
"cron_time_routing": {
"enabled": True,
"peak_model": {"provider": "", "model": ""},
"offpeak_model": {"provider": "", "model": ""},
}
}
assert resolve_cron_time_route(cfg) is None
def test_returns_route_with_hour_injection(monkeypatch):
"""Force hour=14 (peak) via _get_current_hour_in_tz patch."""
monkeypatch.setattr(
"agent.smart_model_routing._get_current_hour_in_tz",
lambda tz: 14,
)
result = resolve_cron_time_route(_CRON_ROUTING_CFG)
assert result is not None
assert result["model"] == "xiaomi/mimo-v2-pro"
assert result["is_peak_hour"] is True
assert result["hour"] == 14
assert result["routing_reason"] == "cron_peak_hours"
def test_returns_offpeak_route(monkeypatch):
monkeypatch.setattr(
"agent.smart_model_routing._get_current_hour_in_tz",
lambda tz: 3,
)
result = resolve_cron_time_route(_CRON_ROUTING_CFG)
assert result is not None
assert result["model"] == "anthropic/claude-sonnet-4"
assert result["is_peak_hour"] is False
assert result["hour"] == 3
assert result["routing_reason"] == "cron_offpeak_hours"
# ---------------------------------------------------------------------------
# resolve_cron_turn_route
# ---------------------------------------------------------------------------
_PRIMARY = {
"model": "anthropic/claude-opus-4",
"provider": "openrouter",
"base_url": "https://openrouter.ai/api/v1",
"api_mode": "chat_completions",
"api_key": "***",
}
def test_cron_turn_route_uses_time_awareness(monkeypatch):
monkeypatch.setattr(
"agent.smart_model_routing._get_current_hour_in_tz",
lambda tz: 2, # off-peak
)
monkeypatch.setattr(
"hermes_cli.runtime_provider.resolve_runtime_provider",
lambda **kw: {
"api_key": "test-key",
"base_url": "https://openrouter.ai/api/v1",
"provider": "openrouter",
"api_mode": "chat_completions",
"command": None,
"args": [],
},
)
result = resolve_cron_turn_route("check status", _CRON_ROUTING_CFG, _PRIMARY)
assert result["model"] == "anthropic/claude-sonnet-4"
assert "cron off-peak" in (result.get("label") or "")
def test_cron_turn_route_falls_back_to_primary_when_no_config():
result = resolve_cron_turn_route("check status", None, _PRIMARY)
assert result["model"] == "anthropic/claude-opus-4"
assert result["label"] is None # no smart routing match
def test_cron_turn_route_falls_back_on_runtime_error(monkeypatch):
"""If time-route runtime resolution fails, fall back to normal routing."""
monkeypatch.setattr(
"agent.smart_model_routing._get_current_hour_in_tz",
lambda tz: 2,
)
monkeypatch.setattr(
"hermes_cli.runtime_provider.resolve_runtime_provider",
lambda **kw: (_ for _ in ()).throw(RuntimeError("bad")),
)
result = resolve_cron_turn_route("check status", _CRON_ROUTING_CFG, _PRIMARY)
# Falls back to primary since the time-route runtime failed
assert result["model"] == "anthropic/claude-opus-4"