Anton Palgunov
907f660dfd
- agent/quota_watchdog.py: core logic — SessionDB token tracking, Codex API integration, threshold comparison, auto-fallback - scripts/quota_watchdog.py: CLI entry point with --status, --dry-run, --quiet-unchanged (default in non-tty cron mode) - Config: quota section with thresholds in config.yaml - Cron: 30-min no_agent watchdog job (fb918d5e5dd1)
48 lines
1.8 KiB
Python
48 lines
1.8 KiB
Python
#!/usr/bin/env python3
|
|
"""Run Hermes quota watchdog.
|
|
|
|
This script is safe for cron: with --quiet-unchanged it prints only new
|
|
warning/critical alerts, so no_agent cron jobs stay silent when nothing changed.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
# Support running from ~/.hermes/scripts/quota_watchdog.py when Hermes is not
|
|
# installed as an editable package but the source checkout exists in the usual
|
|
# location.
|
|
REPO = Path.home() / ".hermes" / "hermes-agent"
|
|
if REPO.exists() and str(REPO) not in sys.path:
|
|
sys.path.insert(0, str(REPO))
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description="Check paid provider quotas and force local fallback at critical threshold.")
|
|
parser.add_argument("--dry-run", action="store_true", help="do not write config/state")
|
|
parser.add_argument("--quiet-unchanged", action="store_true", help="only print newly emitted alerts")
|
|
parser.add_argument("--status", action="store_true", help="always print the current quota status")
|
|
args = parser.parse_args()
|
|
# In cron/non-interactive mode, default to quiet-unchanged to avoid spamming
|
|
if not sys.stdin.isatty() and not args.status:
|
|
args.quiet_unchanged = True
|
|
|
|
from agent.quota_watchdog import evaluate_and_apply, format_status
|
|
|
|
result = evaluate_and_apply(apply=not args.dry_run, quiet_unchanged=args.quiet_unchanged)
|
|
lines: list[str]
|
|
if args.status or not args.quiet_unchanged:
|
|
lines = format_status(result)
|
|
else:
|
|
lines = list(result.alerts)
|
|
if result.switched and not lines:
|
|
lines = [f"Quota critical; forced fallback to {result.fallback_model} via {result.fallback_provider}"]
|
|
if lines:
|
|
print("\n".join(lines))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|