hermes-agent/deploy-crons.py

#!/usr/bin/env python3
"""
deploy-crons — normalize cron job schemas for consistent model field types.

This script ensures that the model field in jobs.json is always a dict when
either model or provider is specified, preventing schema inconsistency.

Usage:
    python deploy-crons.py [--dry-run] [--jobs-file PATH]
"""

import argparse
import json
import sys
from pathlib import Path
from typing import Any, Dict, Optional


def normalize_job(job: Dict[str, Any]) -> Dict[str, Any]:
    """
    Normalize a job dict to ensure consistent model field types.

    Before normalization:
    - If model AND provider: model = raw string, provider = raw string (inconsistent)
    - If only model: model = raw string
    - If only provider: provider = raw string at top level

    After normalization:
    - If model exists: model = {"model": "xxx"}
    - If provider exists: model = {"provider": "yyy"}
    - If both exist: model = {"model": "xxx", "provider": "yyy"}
    - If neither: model = None
    """
    job = dict(job)  # Create a copy to avoid modifying the original
    
    model = job.get("model")
    provider = job.get("provider")
    
    # Skip if already normalized (model is a dict)
    if isinstance(model, dict):
        return job
    
    # Build normalized model dict
    model_dict = {}
    
    if model is not None and isinstance(model, str):
        model_dict["model"] = model.strip()
    
    if provider is not None and isinstance(provider, str):
        model_dict["provider"] = provider.strip()
    
    # Set model field
    if model_dict:
        job["model"] = model_dict
    else:
        job["model"] = None
    
    # Remove top-level provider field if it was moved into model dict
    if provider is not None and "provider" in model_dict:
        # Keep provider field for backward compatibility but mark it as deprecated
        # This allows existing code that reads job["provider"] to continue working
        pass
    
    return job


def normalize_jobs_file(jobs_file: Path, dry_run: bool = False) -> int:
    """
    Normalize all jobs in a jobs.json file.
    
    Returns the number of jobs that were modified.
    """
    if not jobs_file.exists():
        print(f"Error: Jobs file not found: {jobs_file}", file=sys.stderr)
        return 1
    
    try:
        with open(jobs_file, 'r', encoding='utf-8') as f:
            data = json.load(f)
    except json.JSONDecodeError as e:
        print(f"Error: Invalid JSON in {jobs_file}: {e}", file=sys.stderr)
        return 1
    
    jobs = data.get("jobs", [])
    if not jobs:
        print("No jobs found in file.")
        return 0
    
    modified_count = 0
    for i, job in enumerate(jobs):
        original_model = job.get("model")
        original_provider = job.get("provider")
        
        normalized_job = normalize_job(job)
        
        # Check if anything changed
        if (normalized_job.get("model") != original_model or
            normalized_job.get("provider") != original_provider):
            jobs[i] = normalized_job
            modified_count += 1
            
            job_id = job.get("id", "?")
            job_name = job.get("name", "(unnamed)")
            print(f"Normalized job {job_id} ({job_name}):")
            print(f"  model: {original_model!r} -> {normalized_job.get('model')!r}")
            print(f"  provider: {original_provider!r} -> {normalized_job.get('provider')!r}")
    
    if modified_count == 0:
        print("All jobs already have consistent model field types.")
        return 0
    
    if dry_run:
        print(f"DRY RUN: Would normalize {modified_count} jobs.")
        return 0
    
    # Write back to file
    data["jobs"] = jobs
    try:
        with open(jobs_file, 'w', encoding='utf-8') as f:
            json.dump(data, f, indent=2, ensure_ascii=False)
        print(f"Normalized {modified_count} jobs in {jobs_file}")
        return 0
    except Exception as e:
        print(f"Error writing to {jobs_file}: {e}", file=sys.stderr)
        return 1


def main():
    parser = argparse.ArgumentParser(
        description="Normalize cron job schemas for consistent model field types."
    )
    parser.add_argument(
        "--dry-run",
        action="store_true",
        help="Show what would be changed without modifying the file."
    )
    parser.add_argument(
        "--jobs-file",
        type=Path,
        default=Path.home() / ".hermes" / "cron" / "jobs.json",
        help="Path to jobs.json file (default: ~/.hermes/cron/jobs.json)"
    )
    
    args = parser.parse_args()
    
    if args.dry_run:
        print("DRY RUN MODE — no changes will be made.")
        print()
    
    return normalize_jobs_file(args.jobs_file, args.dry_run)


if __name__ == "__main__":
    sys.exit(main())