hermes-agent/tools/cronjob_tools.py

"""
Cron job management tools for Hermes Agent.

These tools allow the agent to schedule, list, and remove automated tasks.
Only available when running via CLI (hermes-cli toolset).

IMPORTANT: Cronjobs run in isolated sessions with NO prior context.
The prompt must contain ALL necessary information.
"""

import json
import os
from typing import Optional

# Import from cron module (will be available when properly installed)
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).parent.parent))

from cron.jobs import create_job, get_job, list_jobs, remove_job


# =============================================================================
# Tool: schedule_cronjob
# =============================================================================

def schedule_cronjob(
    prompt: str,
    schedule: str,
    name: Optional[str] = None,
    repeat: Optional[int] = None,
    task_id: str = None
) -> str:
    """
    Schedule an automated task to run the agent on a schedule.
    
    IMPORTANT: When the cronjob runs, it starts a COMPLETELY FRESH session.
    The agent will have NO memory of this conversation or any prior context.
    Therefore, the prompt MUST contain ALL necessary information:
    - Full context of what needs to be done
    - Specific file paths, URLs, or identifiers
    - Clear success criteria
    - Any relevant background information
    
    BAD prompt:  "Check on that server issue"
    GOOD prompt: "SSH into server 192.168.1.100 as user 'deploy', check if nginx 
                  is running with 'systemctl status nginx', and verify the site 
                  https://example.com returns HTTP 200. Report any issues found."
    
    Args:
        prompt: Complete, self-contained instructions for the future agent.
                Must include ALL context needed - the agent won't remember anything.
        schedule: When to run. Either:
                  - Duration for one-shot: "30m", "2h", "1d" (runs once)
                  - Interval: "every 30m", "every 2h" (recurring)
                  - Cron expression: "0 9 * * *" (daily at 9am)
                  - ISO timestamp: "2026-02-03T14:00:00" (one-shot at specific time)
        name: Optional human-friendly name for the job (for listing/management)
        repeat: How many times to run. Omit for default behavior:
                - One-shot schedules default to repeat=1 (run once)
                - Intervals/cron default to forever
                - Set repeat=5 to run 5 times then auto-delete
    
    Returns:
        JSON with job_id, next_run time, and confirmation
    """
    try:
        job = create_job(
            prompt=prompt,
            schedule=schedule,
            name=name,
            repeat=repeat
        )
        
        # Format repeat info for display
        times = job["repeat"].get("times")
        if times is None:
            repeat_display = "forever"
        elif times == 1:
            repeat_display = "once"
        else:
            repeat_display = f"{times} times"
        
        return json.dumps({
            "success": True,
            "job_id": job["id"],
            "name": job["name"],
            "schedule": job["schedule_display"],
            "repeat": repeat_display,
            "next_run_at": job["next_run_at"],
            "message": f"Cronjob '{job['name']}' created. It will run {repeat_display}, next at {job['next_run_at']}."
        }, indent=2)
        
    except Exception as e:
        return json.dumps({
            "success": False,
            "error": str(e)
        }, indent=2)


SCHEDULE_CRONJOB_SCHEMA = {
    "name": "schedule_cronjob",
    "description": """Schedule an automated task to run the agent on a schedule.

⚠️ CRITICAL: The cronjob runs in a FRESH SESSION with NO CONTEXT from this conversation.
The prompt must be COMPLETELY SELF-CONTAINED with ALL necessary information including:
- Full context and background
- Specific file paths, URLs, server addresses
- Clear instructions and success criteria
- Any credentials or configuration details

The future agent will NOT remember anything from the current conversation.

SCHEDULE FORMATS:
- One-shot: "30m", "2h", "1d" (runs once after delay)
- Interval: "every 30m", "every 2h" (recurring)  
- Cron: "0 9 * * *" (cron expression for precise scheduling)
- Timestamp: "2026-02-03T14:00:00" (specific date/time)

REPEAT BEHAVIOR:
- One-shot schedules: run once by default
- Intervals/cron: run forever by default
- Set repeat=N to run exactly N times then auto-delete

Use for: reminders, periodic checks, scheduled reports, automated maintenance.""",
    "parameters": {
        "type": "object",
        "properties": {
            "prompt": {
                "type": "string",
                "description": "Complete, self-contained instructions. Must include ALL context - the future agent will have NO memory of this conversation."
            },
            "schedule": {
                "type": "string",
                "description": "When to run: '30m' (once in 30min), 'every 30m' (recurring), '0 9 * * *' (cron), or ISO timestamp"
            },
            "name": {
                "type": "string",
                "description": "Optional human-friendly name for the job"
            },
            "repeat": {
                "type": "integer",
                "description": "How many times to run. Omit for default (once for one-shot, forever for recurring). Set to N for exactly N runs."
            }
        },
        "required": ["prompt", "schedule"]
    }
}


# =============================================================================
# Tool: list_cronjobs
# =============================================================================

def list_cronjobs(include_disabled: bool = False, task_id: str = None) -> str:
    """
    List all scheduled cronjobs.
    
    Returns information about each job including:
    - Job ID (needed for removal)
    - Name
    - Schedule (human-readable)
    - Repeat status (completed/total or 'forever')
    - Next scheduled run time
    - Last run time and status (if any)
    
    Args:
        include_disabled: Whether to include disabled/completed jobs
    
    Returns:
        JSON array of all scheduled jobs
    """
    try:
        jobs = list_jobs(include_disabled=include_disabled)
        
        formatted_jobs = []
        for job in jobs:
            # Format repeat status
            times = job["repeat"].get("times")
            completed = job["repeat"].get("completed", 0)
            if times is None:
                repeat_status = "forever"
            else:
                repeat_status = f"{completed}/{times}"
            
            formatted_jobs.append({
                "job_id": job["id"],
                "name": job["name"],
                "prompt_preview": job["prompt"][:100] + "..." if len(job["prompt"]) > 100 else job["prompt"],
                "schedule": job["schedule_display"],
                "repeat": repeat_status,
                "next_run_at": job.get("next_run_at"),
                "last_run_at": job.get("last_run_at"),
                "last_status": job.get("last_status"),
                "enabled": job.get("enabled", True)
            })
        
        return json.dumps({
            "success": True,
            "count": len(formatted_jobs),
            "jobs": formatted_jobs
        }, indent=2)
        
    except Exception as e:
        return json.dumps({
            "success": False,
            "error": str(e)
        }, indent=2)


LIST_CRONJOBS_SCHEMA = {
    "name": "list_cronjobs",
    "description": """List all scheduled cronjobs with their IDs, schedules, and status.

Use this to:
- See what jobs are currently scheduled
- Find job IDs for removal with remove_cronjob
- Check job status and next run times

Returns job_id, name, schedule, repeat status, next/last run times.""",
    "parameters": {
        "type": "object",
        "properties": {
            "include_disabled": {
                "type": "boolean",
                "description": "Include disabled/completed jobs in the list (default: false)"
            }
        },
        "required": []
    }
}


# =============================================================================
# Tool: remove_cronjob
# =============================================================================

def remove_cronjob(job_id: str, task_id: str = None) -> str:
    """
    Remove a scheduled cronjob by its ID.
    
    Use list_cronjobs first to find the job_id of the job you want to remove.
    
    Args:
        job_id: The ID of the job to remove (from list_cronjobs output)
    
    Returns:
        JSON confirmation of removal
    """
    try:
        job = get_job(job_id)
        if not job:
            return json.dumps({
                "success": False,
                "error": f"Job with ID '{job_id}' not found. Use list_cronjobs to see available jobs."
            }, indent=2)
        
        removed = remove_job(job_id)
        if removed:
            return json.dumps({
                "success": True,
                "message": f"Cronjob '{job['name']}' (ID: {job_id}) has been removed.",
                "removed_job": {
                    "id": job_id,
                    "name": job["name"],
                    "schedule": job["schedule_display"]
                }
            }, indent=2)
        else:
            return json.dumps({
                "success": False,
                "error": f"Failed to remove job '{job_id}'"
            }, indent=2)
            
    except Exception as e:
        return json.dumps({
            "success": False,
            "error": str(e)
        }, indent=2)


REMOVE_CRONJOB_SCHEMA = {
    "name": "remove_cronjob",
    "description": """Remove a scheduled cronjob by its ID.

Use list_cronjobs first to find the job_id of the job you want to remove.
Jobs that have completed their repeat count are auto-removed, but you can
use this to cancel a job before it completes.""",
    "parameters": {
        "type": "object",
        "properties": {
            "job_id": {
                "type": "string",
                "description": "The ID of the cronjob to remove (from list_cronjobs output)"
            }
        },
        "required": ["job_id"]
    }
}


# =============================================================================
# Requirements check
# =============================================================================

def check_cronjob_requirements() -> bool:
    """
    Check if cronjob tools can be used.
    
    Only available in interactive CLI mode (HERMES_INTERACTIVE=1).
    """
    return os.getenv("HERMES_INTERACTIVE") == "1"


# =============================================================================
# Exports
# =============================================================================

def get_cronjob_tool_definitions():
    """Return tool definitions for cronjob management."""
    return [
        SCHEDULE_CRONJOB_SCHEMA,
        LIST_CRONJOBS_SCHEMA,
        REMOVE_CRONJOB_SCHEMA
    ]


# For direct testing
if __name__ == "__main__":
    # Test the tools
    print("Testing schedule_cronjob:")
    result = schedule_cronjob(
        prompt="Test prompt for cron job",
        schedule="5m",
        name="Test Job"
    )
    print(result)
    
    print("\nTesting list_cronjobs:")
    result = list_cronjobs()
    print(result)