AgentEvalTool/backend/agenteval/intelligent_eval/openclaw_client.py

"""OpenClaw CLI client for managing cron jobs.

Wraps `openclaw automations` commands to create, delete, and list cron jobs.
"""

import asyncio
import json
import logging
from dataclasses import dataclass
from typing import Optional

_logger = logging.getLogger("agenteval")


@dataclass
class OpenClawCron:
    """OpenClaw cron job info."""

    id: str
    name: str
    schedule: str
    enabled: bool
    state: Optional[dict] = None


class OpenClawClient:
    """Client for OpenClaw CLI commands."""

    def __init__(self, openclaw_bin: str = "docker", container_name: str | None = None, gateway_token: str | None = None):
        from agenteval.config.settings import get_settings
        settings = get_settings()
        self.openclaw_bin = openclaw_bin
        self.container_name = container_name if container_name is not None else settings.openclaw_container_name
        self.gateway_token = gateway_token if gateway_token is not None else settings.openclaw_gateway_token

    async def _run_command(self, *args: str) -> tuple[int, str, str]:
        """Run an OpenClaw CLI command via docker exec.

        Returns:
            (returncode, stdout, stderr)
        """
        # token 需要放在子命令後面
        cmd = [self.openclaw_bin, "exec", self.container_name, "openclaw"] + list(args) + [f"--token={self.gateway_token}"]
        _logger.debug(f"Running OpenClaw command: {' '.join(cmd)}")

        process = await asyncio.create_subprocess_exec(
            *cmd,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
        )
        stdout, stderr = await process.communicate()

        return (
            process.returncode or 0,
            stdout.decode("utf-8"),
            stderr.decode("utf-8"),
        )

    async def create_cron(
        self,
        *,
        name: str,
        schedule: str,
        skill: str,
        state: Optional[dict] = None,
    ) -> str:
        """Create a cron job.

        Args:
            name: Cron job name
            schedule: Cron schedule expression (e.g., "* * * * *")
            skill: Skill to execute (sent as message to trigger skill)
            state: Initial state (JSON, stored in description)

        Returns:
            Cron job ID

        Raises:
            RuntimeError: If creation fails
        """
        args = [
            "cron",
            "add",
            "--cron", schedule,
            f"--name={name}",
            f"--message={skill}",
            "--json",
        ]

        if state:
            # Store state in description since there's no --state option
            args.append(f"--description=state:{json.dumps(state)}")

        returncode, stdout, stderr = await self._run_command(*args)

        if returncode != 0:
            raise RuntimeError(f"Failed to create cron: {stderr}")

        # Parse cron ID from JSON output
        try:
            data = json.loads(stdout)
            cron_id = data.get("id")
            if cron_id:
                _logger.info(f"Created OpenClaw cron {cron_id}: {name}")
                return cron_id
        except json.JSONDecodeError:
            pass

        raise RuntimeError(f"Failed to parse cron ID from output: {stdout}")

    async def delete_cron(self, cron_id: str) -> None:
        """Delete a cron job.

        Args:
            cron_id: Cron job ID

        Raises:
            RuntimeError: If deletion fails
        """
        returncode, stdout, stderr = await self._run_command(
            "cron",
            "rm",
            cron_id,
        )

        if returncode != 0:
            raise RuntimeError(f"Failed to delete cron {cron_id}: {stderr}")

        _logger.info(f"Deleted OpenClaw cron {cron_id}")

    async def list_crons(self) -> list[OpenClawCron]:
        """List all cron jobs.

        Returns:
            List of cron jobs

        Raises:
            RuntimeError: If listing fails
        """
        returncode, stdout, stderr = await self._run_command(
            "cron",
            "list",
            "--json",
        )

        if returncode != 0:
            raise RuntimeError(f"Failed to list crons: {stderr}")

        try:
            data = json.loads(stdout)
            crons = []
            for item in data:
                crons.append(
                    OpenClawCron(
                        id=item["id"],
                        name=item.get("name", ""),
                        schedule=item.get("schedule", ""),
                        enabled=item.get("enabled", True),
                        state=item.get("state"),
                    )
                )
            return crons
        except (json.JSONDecodeError, KeyError) as e:
            raise RuntimeError(f"Failed to parse cron list: {e}") from e

    async def get_cron_state(self, cron_id: str) -> Optional[dict]:
        """Get cron job state.

        Args:
            cron_id: Cron job ID

        Returns:
            State dict, or None if not found
        """
        crons = await self.list_crons()
        for cron in crons:
            if cron.id == cron_id:
                return cron.state
        return None