feat: time-aware model routing for cron jobs #889

scripts/time-aware-model-router.py

@@ -0,0 +1,145 @@
+#!/usr/bin/env python3
+"""
+time-aware-model-router.py — Route cron jobs to better models during high-error hours.
+
+Empirical finding (audit 2026-04-12): Error rate peaks at 18:00 (9.4%) during
+evening cron batches vs 4.0% at 09:00 during interactive work.
+
+This script provides a model resolver that selects a more capable model during
+high-error hours (17:00-22:00) and the default model otherwise.
+
+Usage:
+    # As a standalone resolver
+    python3 scripts/time-aware-model-router.py
+    # Returns: {"provider": "nous", "model": "xiaomi/mimo-v2-pro"}
+
+    # With hour override for testing
+    python3 scripts/time-aware-model-router.py --hour 18
+    # Returns: {"provider": "openrouter", "model": "anthropic/claude-sonnet-4"}
+
+    # As a cron job wrapper
+    python3 scripts/time-aware-model-router.py --wrap -- prompt goes here
+
+Environment variables:
+    HERMES_DEFAULT_PROVIDER: Default provider for normal hours (default: nous)
+    HERMES_DEFAULT_MODEL: Default model for normal hours (default: xiaomi/mimo-v2-pro)
+    HERMES_PEAK_PROVIDER: Provider for high-error hours (default: openrouter)
+    HERMES_PEAK_MODEL: Model for high-error hours (default: anthropic/claude-sonnet-4)
+    HERMES_PEAK_HOURS: Comma-separated hours for peak routing (default: 17,18,19,20,21,22)
+
+Refs: hermes-agent#889
+"""
+
+import json
+import os
+import sys
+import time
+from datetime import datetime
+
+# ── Config ──────────────────────────────────────────────────────────────────
+
+DEFAULT_PROVIDER = os.environ.get("HERMES_DEFAULT_PROVIDER", "nous")
+DEFAULT_MODEL = os.environ.get("HERMES_DEFAULT_MODEL", "xiaomi/mimo-v2-pro")
+PEAK_PROVIDER = os.environ.get("HERMES_PEAK_PROVIDER", "openrouter")
+PEAK_MODEL = os.environ.get("HERMES_PEAK_MODEL", "anthropic/claude-sonnet-4")
+PEAK_HOURS = set(int(h) for h in os.environ.get("HERMES_PEAK_HOURS", "17,18,19,20,21,22").split(","))
+
+# ── Time-aware routing ─────────────────────────────────────────────────────
+
+def get_current_hour():
+    """Get the current local hour (0-23)."""
+    return datetime.now().hour
+
+
+def is_peak_hour(hour=None):
+    """Check if the given hour (or current hour) is a high-error period."""
+    if hour is None:
+        hour = get_current_hour()
+    return hour in PEAK_HOURS
+
+
+def resolve_model(hour=None):
+    """
+    Resolve which model to use based on time of day.
+
+    Returns dict with 'provider' and 'model' keys.
+    During peak hours (high error rate), uses a more capable model.
+    During normal hours, uses the default model.
+    """
+    if is_peak_hour(hour):
+        return {
+            "provider": PEAK_PROVIDER,
+            "model": PEAK_MODEL,
+            "reason": f"peak_hour ({hour if hour is not None else get_current_hour()}:00)",
+            "confidence_note": "Using stronger model during high-error period"
+        }
+    else:
+        return {
+            "provider": DEFAULT_PROVIDER,
+            "model": DEFAULT_MODEL,
+            "reason": "normal_hour",
+            "confidence_note": "Default model sufficient during low-error period"
+        }
+
+
+def get_routing_info():
+    """Get full routing info including current state and config."""
+    hour = get_current_hour()
+    resolved = resolve_model(hour)
+    return {
+        "current_hour": hour,
+        "is_peak": is_peak_hour(hour),
+        "peak_hours": sorted(PEAK_HOURS),
+        "routing": resolved,
+        "config": {
+            "default": {"provider": DEFAULT_PROVIDER, "model": DEFAULT_MODEL},
+            "peak": {"provider": PEAK_PROVIDER, "model": PEAK_MODEL},
+        },
+        "source": "hermes-agent#889 — empirical audit 2026-04-12",
+    }
+
+
+# ── CLI ─────────────────────────────────────────────────────────────────────
+
+def main():
+    args = sys.argv[1:]
+
+    # Parse --hour
+    hour = None
+    if "--hour" in args:
+        idx = args.index("--hour")
+        if idx + 1 < len(args):
+            hour = int(args[idx + 1])
+
+    # Parse --wrap mode
+    if "--wrap" in args:
+        # Run the remaining args as a command with model override
+        resolved = resolve_model(hour)
+        wrap_idx = args.index("--wrap")
+        cmd_parts = args[wrap_idx + 1:]
+
+        # Inject model/provider into environment
+        env = os.environ.copy()
+        env["HERMES_MODEL"] = resolved["model"]
+        env["HERMES_PROVIDER"] = resolved["provider"]
+
+        if cmd_parts:
+            import subprocess
+            result = subprocess.run(cmd_parts, env=env)
+            sys.exit(result.returncode)
+        else:
+            print(json.dumps(resolved, indent=2))
+            sys.exit(0)
+
+    # Parse --info mode
+    if "--info" in args:
+        print(json.dumps(get_routing_info(), indent=2))
+        sys.exit(0)
+
+    # Default: output resolved model as JSON
+    resolved = resolve_model(hour)
+    print(json.dumps(resolved, indent=2))
+
+
+if __name__ == "__main__":
+    main()