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
8 changed files with 340 additions and 988 deletions

View File

@@ -4,6 +4,7 @@ from __future__ import annotations
import os
import re
from datetime import datetime
from typing import Any, Dict, Optional
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"),
"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": (
route.get("model"),
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 ()),
),
}
# ---------------------------------------------------------------------------
# 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:
# provider: openrouter
# 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

View File

@@ -762,8 +762,8 @@ def run_job(job: dict) -> tuple[bool, str, str, Optional[str]]:
message = format_runtime_provider_error(exc)
raise RuntimeError(message) from exc
from agent.smart_model_routing import resolve_turn_route
turn_route = resolve_turn_route(
from agent.smart_model_routing import resolve_cron_turn_route
turn_route = resolve_cron_turn_route(
prompt,
smart_routing,
{
@@ -776,6 +776,8 @@ def run_job(job: dict) -> tuple[bool, str, str, Optional[str]]:
"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({
"model": turn_route["model"],

View File

@@ -1,5 +0,0 @@
"""Hermes daemon services — long-running background processes."""
from daemon.confirmation_server import ConfirmationServer
__all__ = ["ConfirmationServer"]

View File

@@ -1,664 +0,0 @@
"""Human Confirmation Daemon — route high-risk actions through human review.
HTTP server on port 6000 that intercepts dangerous operations and holds them
until a human approves or denies. Integrates with the existing approval
system (tools/approval.py) and notifies humans via Telegram/Discord/CLI.
Endpoints:
POST /confirm — submit a high-risk action for review
POST /confirm/{id}/approve — approve a pending confirmation
POST /confirm/{id}/deny — deny a pending confirmation
GET /confirm/{id} — check status of a confirmation
GET /audit — recent audit log entries
GET /health — liveness probe
Every decision is logged to SQLite for audit.
"""
from __future__ import annotations
import asyncio
import json
import logging
import os
import sqlite3
import threading
import time
import uuid
from dataclasses import dataclass, field, asdict
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Dict, List, Optional
logger = logging.getLogger(__name__)
try:
from aiohttp import web
AIOHTTP_AVAILABLE = True
except ImportError:
AIOHTTP_AVAILABLE = False
web = None # type: ignore[assignment]
# ---------------------------------------------------------------------------
# Configuration
# ---------------------------------------------------------------------------
DEFAULT_HOST = "127.0.0.1"
DEFAULT_PORT = 6000
# Actions that always require human confirmation (not bypassable)
HIGH_RISK_ACTIONS = {
"deploy_production",
"delete_data",
"transfer_funds",
"modify_permissions",
"shutdown_service",
"wipe_database",
"exec_remote",
"publish_package",
"rotate_keys",
"migrate_database",
}
# Default rate limits: max N confirmations per action type per window
DEFAULT_RATE_LIMIT = 10 # max confirmations per action type
RATE_LIMIT_WINDOW = 3600 # 1 hour in seconds
# ---------------------------------------------------------------------------
# Data model
# ---------------------------------------------------------------------------
@dataclass
class ConfirmationRequest:
"""A single pending or resolved confirmation."""
id: str
action: str
description: str
details: Dict[str, Any] = field(default_factory=dict)
requester: str = "" # agent or user who requested
session_key: str = ""
status: str = "pending" # pending | approved | denied | expired
resolved_by: str = ""
resolved_at: Optional[float] = None
created_at: float = field(default_factory=time.time)
timeout_seconds: int = 300 # 5 min default
def to_dict(self) -> dict:
d = asdict(self)
d["created_at_iso"] = _ts_to_iso(d["created_at"])
d["resolved_at_iso"] = _ts_to_iso(d["resolved_at"]) if d["resolved_at"] else None
return d
def _ts_to_iso(ts: Optional[float]) -> Optional[str]:
if ts is None:
return None
return datetime.fromtimestamp(ts, tz=timezone.utc).isoformat()
# ---------------------------------------------------------------------------
# Audit log (SQLite)
# ---------------------------------------------------------------------------
class AuditLog:
"""SQLite-backed audit log for all confirmation decisions."""
def __init__(self, db_path: Optional[str] = None):
if db_path is None:
home = Path(os.getenv("HERMES_HOME", Path.home() / ".hermes"))
home.mkdir(parents=True, exist_ok=True)
db_path = str(home / "confirmation_audit.db")
self._conn = sqlite3.connect(db_path, check_same_thread=False)
self._conn.execute("PRAGMA journal_mode=WAL")
self._conn.execute("""
CREATE TABLE IF NOT EXISTS audit_log (
id TEXT PRIMARY KEY,
action TEXT NOT NULL,
description TEXT NOT NULL,
details TEXT NOT NULL DEFAULT '{}',
requester TEXT NOT NULL DEFAULT '',
session_key TEXT NOT NULL DEFAULT '',
status TEXT NOT NULL,
resolved_by TEXT NOT NULL DEFAULT '',
created_at REAL NOT NULL,
resolved_at REAL,
resolved_at_iso TEXT,
created_at_iso TEXT
)
""")
self._conn.commit()
def log(self, req: ConfirmationRequest) -> None:
d = req.to_dict()
self._conn.execute(
"""INSERT OR REPLACE INTO audit_log
(id, action, description, details, requester, session_key,
status, resolved_by, created_at, resolved_at,
resolved_at_iso, created_at_iso)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
(
d["id"], d["action"], d["description"],
json.dumps(d["details"]), d["requester"], d["session_key"],
d["status"], d["resolved_by"],
d["created_at"], d["resolved_at"],
d["resolved_at_iso"], d["created_at_iso"],
),
)
self._conn.commit()
def recent(self, limit: int = 50) -> List[dict]:
rows = self._conn.execute(
"SELECT * FROM audit_log ORDER BY created_at DESC LIMIT ?", (limit,)
).fetchall()
cols = [d[0] for d in self._conn.description]
return [dict(zip(cols, row)) for row in rows]
def close(self) -> None:
self._conn.close()
# ---------------------------------------------------------------------------
# Rate limiter
# ---------------------------------------------------------------------------
class RateLimiter:
"""Simple sliding-window rate limiter per action type."""
def __init__(self, max_per_window: int = DEFAULT_RATE_LIMIT,
window: int = RATE_LIMIT_WINDOW):
self._max = max_per_window
self._window = window
self._timestamps: Dict[str, List[float]] = {} # action -> [ts, ...]
self._lock = threading.Lock()
def check(self, action: str) -> bool:
"""Return True if the action is within rate limits."""
now = time.time()
with self._lock:
timestamps = self._timestamps.get(action, [])
# Prune expired
timestamps = [t for t in timestamps if now - t < self._window]
self._timestamps[action] = timestamps
if len(timestamps) >= self._max:
return False
timestamps.append(now)
return True
def remaining(self, action: str) -> int:
now = time.time()
with self._lock:
timestamps = self._timestamps.get(action, [])
timestamps = [t for t in timestamps if now - t < self._window]
return max(0, self._max - len(timestamps))
# ---------------------------------------------------------------------------
# Notification dispatcher
# ---------------------------------------------------------------------------
async def _notify_human(request: ConfirmationRequest) -> None:
"""Send a notification about a pending confirmation to humans.
Tries Telegram first, then Discord, then falls back to log warning.
Uses the existing send_message infrastructure.
"""
msg = (
f"\U0001f514 Confirmation Required\n"
f"Action: {request.action}\n"
f"Description: {request.description}\n"
f"Requester: {request.requester}\n"
f"ID: {request.id}\n\n"
f"Approve: POST /confirm/{request.id}/approve\n"
f"Deny: POST /confirm/{request.id}/deny"
)
sent = False
# Try Telegram
try:
from tools.send_message_tool import _handle_send
result = _handle_send({
"target": "telegram",
"message": msg,
})
result_dict = json.loads(result) if isinstance(result, str) else result
if "error" not in result_dict:
sent = True
logger.info("Confirmation %s: notified via Telegram", request.id)
except Exception as e:
logger.debug("Telegram notify failed: %s", e)
# Try Discord if Telegram failed
if not sent:
try:
from tools.send_message_tool import _handle_send
result = _handle_send({
"target": "discord",
"message": msg,
})
result_dict = json.loads(result) if isinstance(result, str) else result
if "error" not in result_dict:
sent = True
logger.info("Confirmation %s: notified via Discord", request.id)
except Exception as e:
logger.debug("Discord notify failed: %s", e)
if not sent:
logger.warning(
"Confirmation %s: no messaging channel available. "
"Action '%s' requires human review -- check /confirm/%s",
request.id, request.action, request.id,
)
# ---------------------------------------------------------------------------
# Whitelist manager
# ---------------------------------------------------------------------------
class Whitelist:
"""Configurable whitelist for actions that skip confirmation."""
def __init__(self):
self._allowed: Dict[str, set] = {} # session_key -> {action, ...}
self._global_allowed: set = set()
self._lock = threading.Lock()
def is_whitelisted(self, action: str, session_key: str = "") -> bool:
with self._lock:
if action in self._global_allowed:
return True
if session_key and action in self._allowed.get(session_key, set()):
return True
return False
def add(self, action: str, session_key: str = "") -> None:
with self._lock:
if session_key:
self._allowed.setdefault(session_key, set()).add(action)
else:
self._global_allowed.add(action)
def remove(self, action: str, session_key: str = "") -> None:
with self._lock:
if session_key:
self._allowed.get(session_key, set()).discard(action)
else:
self._global_allowed.discard(action)
# ---------------------------------------------------------------------------
# Confirmation Server
# ---------------------------------------------------------------------------
class ConfirmationServer:
"""HTTP server for human confirmation of high-risk actions.
Usage:
server = ConfirmationServer()
await server.start() # blocks
# or:
server.start_background() # non-blocking
...
await server.stop()
"""
def __init__(self, host: str = DEFAULT_HOST, port: int = DEFAULT_PORT,
db_path: Optional[str] = None,
rate_limit: int = DEFAULT_RATE_LIMIT):
if not AIOHTTP_AVAILABLE:
raise RuntimeError(
"aiohttp is required for the confirmation daemon. "
"Install with: pip install aiohttp"
)
self._host = host
self._port = port
self._audit = AuditLog(db_path)
self._rate_limiter = RateLimiter(max_per_window=rate_limit)
self._whitelist = Whitelist()
self._pending: Dict[str, ConfirmationRequest] = {}
self._lock = threading.Lock()
self._app: Optional[web.Application] = None
self._runner: Optional[web.AppRunner] = None
self._bg_thread: Optional[threading.Thread] = None
# --- Lifecycle ---
def _build_app(self) -> web.Application:
app = web.Application(client_max_size=1_048_576) # 1 MB
app["server"] = self
app.router.add_post("/confirm", self._handle_submit)
app.router.add_post("/confirm/{req_id}/approve", self._handle_approve)
app.router.add_post("/confirm/{req_id}/deny", self._handle_deny)
app.router.add_get("/confirm/{req_id}", self._handle_status)
app.router.add_get("/audit", self._handle_audit)
app.router.add_get("/health", self._handle_health)
return app
async def start(self) -> None:
"""Start the server and block until stopped."""
self._app = self._build_app()
self._runner = web.AppRunner(self._app)
await self._runner.setup()
site = web.TCPSite(self._runner, self._host, self._port)
await site.start()
logger.info(
"Confirmation daemon listening on http://%s:%s",
self._host, self._port,
)
# Run until cancelled
try:
await asyncio.Event().wait()
except asyncio.CancelledError:
pass
finally:
await self.stop()
def start_background(self) -> None:
"""Start the server in a background thread (non-blocking)."""
def _run():
asyncio.run(self.start())
self._bg_thread = threading.Thread(target=_run, daemon=True)
self._bg_thread.start()
async def stop(self) -> None:
"""Gracefully stop the server."""
if self._runner:
await self._runner.cleanup()
self._runner = None
self._audit.close()
logger.info("Confirmation daemon stopped")
# --- Internal helpers ---
def _get_request(self, req_id: str) -> Optional[ConfirmationRequest]:
with self._lock:
return self._pending.get(req_id)
def _expire_old_requests(self) -> int:
"""Mark expired requests. Returns count expired."""
now = time.time()
expired = 0
with self._lock:
for req in list(self._pending.values()):
if req.status == "pending" and (now - req.created_at) > req.timeout_seconds:
req.status = "expired"
req.resolved_at = now
req.resolved_by = "system:timeout"
self._audit.log(req)
expired += 1
return expired
# --- HTTP handlers ---
async def _handle_submit(self, request: web.Request) -> web.Response:
"""POST /confirm -- submit a new confirmation request."""
try:
body = await request.json()
except Exception:
return web.json_response(
{"error": "Invalid JSON body"}, status=400
)
action = (body.get("action") or "").strip()
description = (body.get("description") or "").strip()
details = body.get("details") or {}
requester = (body.get("requester") or "agent").strip()
session_key = (body.get("session_key") or "").strip()
timeout = body.get("timeout_seconds", 300)
if not action:
return web.json_response(
{"error": "Field 'action' is required"}, status=400
)
if not description:
return web.json_response(
{"error": "Field 'description' is required"}, status=400
)
# Whitelist check
if self._whitelist.is_whitelisted(action, session_key):
auto_id = str(uuid.uuid4())[:8]
return web.json_response({
"id": auto_id,
"action": action,
"status": "auto_approved",
"message": f"Action '{action}' is whitelisted for this session",
})
# Rate limit check
if not self._rate_limiter.check(action):
remaining = self._rate_limiter.remaining(action)
return web.json_response({
"error": f"Rate limit exceeded for action '{action}'",
"remaining": remaining,
"window_seconds": RATE_LIMIT_WINDOW,
}, status=429)
# Enforce timeout bounds
try:
timeout = max(30, min(int(timeout), 3600))
except (ValueError, TypeError):
timeout = 300
# Create request
req = ConfirmationRequest(
id=str(uuid.uuid4())[:12],
action=action,
description=description,
details=details,
requester=requester,
session_key=session_key,
timeout_seconds=timeout,
)
with self._lock:
self._pending[req.id] = req
# Audit log
self._audit.log(req)
# Notify humans (fire-and-forget)
asyncio.create_task(_notify_human(req))
logger.info(
"Confirmation %s submitted: action=%s requester=%s",
req.id, action, requester,
)
return web.json_response({
"id": req.id,
"action": req.action,
"status": req.status,
"timeout_seconds": req.timeout_seconds,
"message": "Confirmation pending. Awaiting human review.",
"approve_url": f"/confirm/{req.id}/approve",
"deny_url": f"/confirm/{req.id}/deny",
}, status=202)
async def _handle_approve(self, request: web.Request) -> web.Response:
"""POST /confirm/{id}/approve -- approve a pending confirmation."""
req_id = request.match_info["req_id"]
req = self._get_request(req_id)
if req is None:
return web.json_response(
{"error": f"Confirmation '{req_id}' not found"}, status=404
)
if req.status != "pending":
return web.json_response({
"error": f"Confirmation '{req_id}' already resolved",
"status": req.status,
}, status=409)
# Parse optional approver identity
try:
body = await request.json()
approver = (body.get("approver") or "api").strip()
except Exception:
approver = "api"
req.status = "approved"
req.resolved_by = approver
req.resolved_at = time.time()
# Audit log
self._audit.log(req)
logger.info(
"Confirmation %s APPROVED by %s (action=%s)",
req_id, approver, req.action,
)
return web.json_response({
"id": req.id,
"action": req.action,
"status": "approved",
"resolved_by": approver,
"resolved_at": req.resolved_at,
})
async def _handle_deny(self, request: web.Request) -> web.Response:
"""POST /confirm/{id}/deny -- deny a pending confirmation."""
req_id = request.match_info["req_id"]
req = self._get_request(req_id)
if req is None:
return web.json_response(
{"error": f"Confirmation '{req_id}' not found"}, status=404
)
if req.status != "pending":
return web.json_response({
"error": f"Confirmation '{req_id}' already resolved",
"status": req.status,
}, status=409)
try:
body = await request.json()
denier = (body.get("denier") or "api").strip()
reason = (body.get("reason") or "").strip()
except Exception:
denier = "api"
reason = ""
req.status = "denied"
req.resolved_by = denier
req.resolved_at = time.time()
# Audit log
self._audit.log(req)
logger.info(
"Confirmation %s DENIED by %s (action=%s, reason=%s)",
req_id, denier, req.action, reason,
)
resp = {
"id": req.id,
"action": req.action,
"status": "denied",
"resolved_by": denier,
"resolved_at": req.resolved_at,
}
if reason:
resp["reason"] = reason
return web.json_response(resp)
async def _handle_status(self, request: web.Request) -> web.Response:
"""GET /confirm/{id} -- check status of a confirmation."""
req_id = request.match_info["req_id"]
req = self._get_request(req_id)
if req is None:
return web.json_response(
{"error": f"Confirmation '{req_id}' not found"}, status=404
)
# Check for expiration
if req.status == "pending":
now = time.time()
if (now - req.created_at) > req.timeout_seconds:
req.status = "expired"
req.resolved_at = now
req.resolved_by = "system:timeout"
self._audit.log(req)
return web.json_response(req.to_dict())
async def _handle_audit(self, request: web.Request) -> web.Response:
"""GET /audit -- recent audit log entries."""
try:
limit = int(request.query.get("limit", "50"))
limit = max(1, min(limit, 500))
except (ValueError, TypeError):
limit = 50
entries = self._audit.recent(limit)
return web.json_response({
"count": len(entries),
"entries": entries,
})
async def _handle_health(self, request: web.Request) -> web.Response:
"""GET /health -- liveness probe."""
return web.json_response({
"status": "ok",
"pending_count": len(self._pending),
"timestamp": time.time(),
})
# ---------------------------------------------------------------------------
# CLI entry point
# ---------------------------------------------------------------------------
def main():
"""Run the confirmation daemon as a standalone process."""
import argparse
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(name)s] %(levelname)s: %(message)s",
)
parser = argparse.ArgumentParser(
description="Hermes Human Confirmation Daemon"
)
parser.add_argument(
"--host", default=os.getenv("CONFIRMATION_HOST", DEFAULT_HOST),
help="Bind address (default: 127.0.0.1)",
)
parser.add_argument(
"--port", type=int,
default=int(os.getenv("CONFIRMATION_PORT", DEFAULT_PORT)),
help="Bind port (default: 6000)",
)
parser.add_argument(
"--db-path", default=None,
help="SQLite database path (default: ~/.hermes/confirmation_audit.db)",
)
parser.add_argument(
"--rate-limit", type=int,
default=int(os.getenv("CONFIRMATION_RATE_LIMIT", DEFAULT_RATE_LIMIT)),
help="Max confirmations per action per hour (default: 10)",
)
args = parser.parse_args()
if not AIOHTTP_AVAILABLE:
print("ERROR: aiohttp is required. Install with: pip install aiohttp")
raise SystemExit(1)
server = ConfirmationServer(
host=args.host,
port=args.port,
db_path=args.db_path,
rate_limit=args.rate_limit,
)
print(f"Starting Confirmation Daemon on http://{args.host}:{args.port}")
asyncio.run(server.start())
if __name__ == "__main__":
main()

View File

@@ -299,6 +299,13 @@ DEFAULT_CONFIG = {
"max_simple_chars": 160,
"max_simple_words": 28,
"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.

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"

View File

@@ -1,316 +0,0 @@
"""Tests for the Human Confirmation Daemon."""
import asyncio
import json
import time
from unittest.mock import patch, MagicMock
import pytest
# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------
# We import after the fixtures to avoid aiohttp import issues in test envs
try:
from aiohttp import web
from aiohttp.test_utils import AioHTTPTestCase, unittest_run_loop
AIOHTTP_AVAILABLE = True
except ImportError:
AIOHTTP_AVAILABLE = False
if AIOHTTP_AVAILABLE:
from daemon.confirmation_server import (
ConfirmationServer,
ConfirmationRequest,
AuditLog,
RateLimiter,
Whitelist,
HIGH_RISK_ACTIONS,
DEFAULT_RATE_LIMIT,
RATE_LIMIT_WINDOW,
)
@pytest.mark.skipif(not AIOHTTP_AVAILABLE, reason="aiohttp not installed")
class TestRateLimiter:
"""Unit tests for the RateLimiter."""
def test_allows_within_limit(self):
rl = RateLimiter(max_per_window=3, window=60)
assert rl.check("deploy") is True
assert rl.check("deploy") is True
assert rl.check("deploy") is True
def test_blocks_over_limit(self):
rl = RateLimiter(max_per_window=2, window=60)
assert rl.check("deploy") is True
assert rl.check("deploy") is True
assert rl.check("deploy") is False
def test_remaining_count(self):
rl = RateLimiter(max_per_window=5, window=60)
assert rl.remaining("deploy") == 5
rl.check("deploy")
assert rl.remaining("deploy") == 4
def test_separate_actions_independent(self):
rl = RateLimiter(max_per_window=2, window=60)
assert rl.check("deploy") is True
assert rl.check("deploy") is True
assert rl.check("deploy") is False
assert rl.check("shutdown") is True # different action
@pytest.mark.skipif(not AIOHTTP_AVAILABLE, reason="aiohttp not installed")
class TestWhitelist:
"""Unit tests for the Whitelist."""
def test_global_whitelist(self):
wl = Whitelist()
assert wl.is_whitelisted("deploy") is False
wl.add("deploy")
assert wl.is_whitelisted("deploy") is True
def test_session_scoped_whitelist(self):
wl = Whitelist()
assert wl.is_whitelisted("deploy", "session1") is False
wl.add("deploy", "session1")
assert wl.is_whitelisted("deploy", "session1") is True
assert wl.is_whitelisted("deploy", "session2") is False
def test_remove(self):
wl = Whitelist()
wl.add("deploy")
assert wl.is_whitelisted("deploy") is True
wl.remove("deploy")
assert wl.is_whitelisted("deploy") is False
@pytest.mark.skipif(not AIOHTTP_AVAILABLE, reason="aiohttp not installed")
class TestAuditLog:
"""Unit tests for the AuditLog."""
def test_log_and_retrieve(self, tmp_path):
db = str(tmp_path / "test_audit.db")
log = AuditLog(db_path=db)
req = ConfirmationRequest(
id="test-123",
action="deploy_production",
description="Deploy v2.0 to prod",
requester="timmy",
)
log.log(req)
entries = log.recent(limit=10)
assert len(entries) == 1
assert entries[0]["id"] == "test-123"
assert entries[0]["action"] == "deploy_production"
assert entries[0]["status"] == "pending"
log.close()
def test_update_on_resolve(self, tmp_path):
db = str(tmp_path / "test_audit.db")
log = AuditLog(db_path=db)
req = ConfirmationRequest(
id="test-456",
action="delete_data",
description="Purge old records",
)
log.log(req)
# Resolve
req.status = "approved"
req.resolved_by = "alexander"
req.resolved_at = time.time()
log.log(req)
entries = log.recent(limit=10)
assert len(entries) == 1
assert entries[0]["status"] == "approved"
assert entries[0]["resolved_by"] == "alexander"
log.close()
@pytest.mark.skipif(not AIOHTTP_AVAILABLE, reason="aiohttp not installed")
class TestConfirmationRequest:
"""Unit tests for the data model."""
def test_to_dict(self):
req = ConfirmationRequest(
id="abc123",
action="deploy_production",
description="Ship it",
details={"version": "2.0"},
)
d = req.to_dict()
assert d["id"] == "abc123"
assert d["status"] == "pending"
assert d["created_at_iso"] is not None
assert d["resolved_at_iso"] is None
assert d["details"]["version"] == "2.0"
# ---------------------------------------------------------------------------
# Integration tests (HTTP)
# ---------------------------------------------------------------------------
@pytest.mark.skipif(not AIOHTTP_AVAILABLE, reason="aiohttp not installed")
class TestConfirmationHTTP(AioHTTPTestCase):
"""Full HTTP integration tests for the ConfirmationServer."""
async def get_application(self):
# Suppress notification during tests
with patch("daemon.confirmation_server._notify_human", return_value=None):
server = ConfirmationServer(
host="127.0.0.1",
port=6000,
db_path=":memory:",
)
self._server = server
return server._build_app()
@unittest_run_loop
async def test_health(self):
resp = await self.client.request("GET", "/health")
assert resp.status == 200
data = await resp.json()
assert data["status"] == "ok"
@unittest_run_loop
async def test_submit_confirmation(self):
with patch("daemon.confirmation_server._notify_human", return_value=None):
resp = await self.client.request("POST", "/confirm", json={
"action": "deploy_production",
"description": "Deploy v2.0 to production",
"requester": "timmy",
"session_key": "test-session",
})
assert resp.status == 202
data = await resp.json()
assert data["status"] == "pending"
assert data["action"] == "deploy_production"
assert "id" in data
@unittest_run_loop
async def test_submit_missing_action(self):
resp = await self.client.request("POST", "/confirm", json={
"description": "Something",
})
assert resp.status == 400
@unittest_run_loop
async def test_submit_missing_description(self):
resp = await self.client.request("POST", "/confirm", json={
"action": "deploy_production",
})
assert resp.status == 400
@unittest_run_loop
async def test_approve_flow(self):
# Submit
with patch("daemon.confirmation_server._notify_human", return_value=None):
submit_resp = await self.client.request("POST", "/confirm", json={
"action": "deploy_production",
"description": "Ship it",
})
assert submit_resp.status == 202
submit_data = await submit_resp.json()
req_id = submit_data["id"]
# Approve
approve_resp = await self.client.request(
"POST", f"/confirm/{req_id}/approve",
json={"approver": "alexander"},
)
assert approve_resp.status == 200
approve_data = await approve_resp.json()
assert approve_data["status"] == "approved"
assert approve_data["resolved_by"] == "alexander"
# Check status
status_resp = await self.client.request("GET", f"/confirm/{req_id}")
assert status_resp.status == 200
status_data = await status_resp.json()
assert status_data["status"] == "approved"
@unittest_run_loop
async def test_deny_flow(self):
with patch("daemon.confirmation_server._notify_human", return_value=None):
submit_resp = await self.client.request("POST", "/confirm", json={
"action": "delete_data",
"description": "Wipe everything",
})
req_id = (await submit_resp.json())["id"]
deny_resp = await self.client.request(
"POST", f"/confirm/{req_id}/deny",
json={"denier": "alexander", "reason": "Too risky"},
)
assert deny_resp.status == 200
deny_data = await deny_resp.json()
assert deny_data["status"] == "denied"
assert deny_data["reason"] == "Too risky"
@unittest_run_loop
async def test_double_approve_returns_409(self):
with patch("daemon.confirmation_server._notify_human", return_value=None):
submit_resp = await self.client.request("POST", "/confirm", json={
"action": "deploy_production",
"description": "Ship it",
})
req_id = (await submit_resp.json())["id"]
await self.client.request(f"POST", f"/confirm/{req_id}/approve")
resp2 = await self.client.request(f"POST", f"/confirm/{req_id}/approve")
assert resp2.status == 409
@unittest_run_loop
async def test_not_found(self):
resp = await self.client.request("GET", "/confirm/nonexistent")
assert resp.status == 404
@unittest_run_loop
async def test_audit_log(self):
with patch("daemon.confirmation_server._notify_human", return_value=None):
await self.client.request("POST", "/confirm", json={
"action": "deploy_production",
"description": "Ship it",
})
resp = await self.client.request("GET", "/audit")
assert resp.status == 200
data = await resp.json()
assert data["count"] >= 1
@unittest_run_loop
async def test_rate_limit(self):
# Exhaust rate limit (default is 10)
with patch("daemon.confirmation_server._notify_human", return_value=None):
for i in range(10):
await self.client.request("POST", "/confirm", json={
"action": "test_rate_action",
"description": f"Request {i}",
})
# 11th should be rate-limited
resp = await self.client.request("POST", "/confirm", json={
"action": "test_rate_action",
"description": "Over the limit",
})
assert resp.status == 429
@unittest_run_loop
async def test_whitelist_auto_approves(self):
self._server._whitelist.add("safe_action")
resp = await self.client.request("POST", "/confirm", json={
"action": "safe_action",
"description": "This is whitelisted",
})
assert resp.status == 200
data = await resp.json()
assert data["status"] == "auto_approved"