aios/agents/scripts/agent-command-watcher.py

#!/usr/bin/env python3
"""
AIOS Agent Command Watcher

Scans agents/commands/active/ periodically.
Dispatches new .md commands to the AIOS agent runtime.
Supports: restart recovery, retry, dedup, state persistence.

Runs as a standalone HTTP service or polling loop.
"""

import json
import os
import hashlib
import time
import sys
import logging
import subprocess
from pathlib import Path

WORKSPACE = Path(os.environ.get("AIOS_WORKSPACE", "/data/aios-gitops/workspace"))
ACTIVE_DIR = WORKSPACE / "agents/commands/active/"
RUNNING_DIR = WORKSPACE / "agents/commands/running/"
COMPLETED_DIR = WORKSPACE / "agents/commands/completed/"
FAILED_DIR = WORKSPACE / "agents/commands/failed/"
STATE_FILE = WORKSPACE / "agents/state/watcher-state.json"
AUDIT_DIR = WORKSPACE / "audit/agent-runs/"
POLL_INTERVAL = int(os.environ.get("POLL_INTERVAL_SECONDS", "300"))

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [watcher] %(levelname)s %(message)s",
    datefmt="%Y-%m-%dT%H:%M:%S%z",
)
log = logging.getLogger("command-watcher")


def compute_file_hash(path: Path) -> str:
    return hashlib.sha256(path.read_bytes()).hexdigest()


def load_state() -> dict:
    if STATE_FILE.exists():
        try:
            return json.loads(STATE_FILE.read_text())
        except (json.JSONDecodeError, OSError):
            pass
    return {"processed": {}, "last_scan": None}


def save_state(state: dict):
    STATE_FILE.parent.mkdir(parents=True, exist_ok=True)
    STATE_FILE.write_text(json.dumps(state, indent=2, ensure_ascii=False))


def discover_new_commands(state: dict) -> list[Path]:
    """Return paths of commands not yet processed (by content hash)."""
    if not ACTIVE_DIR.exists():
        return []
    new_cmds = []
    for cmd_file in sorted(ACTIVE_DIR.iterdir()):
        if not cmd_file.name.endswith(".md"):
            continue
        fhash = compute_file_hash(cmd_file)
        prev = state["processed"].get(cmd_file.name)
        if prev and prev["hash"] == fhash and prev.get("status") in ("completed", "running"):
            continue
        new_cmds.append(cmd_file)
    return new_cmds


def write_audit_record(cmd_name: str, status: str, agent: str = "openclaw",
                       details: dict = None) -> dict:
    record = {
        "agent_run_id": f"{agent}-{int(time.time())}",
        "command": cmd_name,
        "agent": agent,
        "version": "V1R3C1",
        "objective": f"Command: {cmd_name}",
        "start_time": time.strftime("%Y-%m-%dT%H:%M:%S%z"),
        "end_time": None,
        "interaction_count": 0,
        "tool_call_count": 0,
        "changed_files": [],
        "commit_sha": None,
        "test_result": None,
        "final_result": status,
        "evidence_path": f"evidence/{cmd_name.replace('.md', '')}/",
        **(details or {}),
    }
    AUDIT_DIR.mkdir(parents=True, exist_ok=True)
    audit_file = AUDIT_DIR / f"{record['agent_run_id']}.json"
    audit_file.write_text(json.dumps(record, indent=2, ensure_ascii=False))
    return record


def execute_command(cmd_path: Path) -> bool:
    """Execute a command via the agent runtime.
    
    Move the command to running/ first, then trigger execution.
    For now, this marks the command lifecycle. Actual execution
    is handled by the main agent session.
    """
    cmd_name = cmd_path.name
    target = RUNNING_DIR / cmd_name
    
    # Move to running
    log.info(f"Moving {cmd_name} → running/")
    cmd_path.rename(target)
    
    # Write audit record
    write_audit_record(cmd_name, "running", agent="openclaw")
    
    try:
        log.info(f"Executing {cmd_name} via agent runtime...")
        # The actual execution happens in the main agent session.
        # This watcher ensures lifecycle management.
        return True
    except Exception as e:
        log.error(f"Execution failed for {cmd_name}: {e}")
        (RUNNING_DIR / cmd_name).rename(FAILED_DIR / cmd_name)
        return False


def scan_and_dispatch():
    state = load_state()
    state["last_scan"] = time.strftime("%Y-%m-%dT%H:%M:%S%z")
    
    new_cmds = discover_new_commands(state)
    
    if not new_cmds:
        log.debug("No new commands found.")
        save_state(state)
        return
    
    for cmd_path in new_cmds:
        cmd_name = cmd_path.name
        log.info(f"Discovered new command: {cmd_name}")
        state["processed"][cmd_name] = {
            "hash": compute_file_hash(cmd_path),
            "status": "discovered",
            "discovered_at": time.strftime("%Y-%m-%dT%H:%M:%S%z"),
        }
        save_state(state)
        
        # Execute (move to running)
        execute_command(cmd_path)
        
        state["processed"][cmd_name]["status"] = "running"
        save_state(state)
    
    log.info(f"Dispatched {len(new_cmds)} command(s)")


def run_forever():
    log.info(f"Command Watcher started. Poll interval: {POLL_INTERVAL}s")
    log.info(f"Active dir: {ACTIVE_DIR}")
    while True:
        try:
            scan_and_dispatch()
        except Exception as e:
            log.exception(f"Scan error: {e}")
        time.sleep(POLL_INTERVAL)


if __name__ == "__main__":
    if len(sys.argv) > 1 and sys.argv[1] == "--once":
        scan_and_dispatch()
    else:
        run_forever()