hermes-agent/plugins/cron/chronos/_nas_client.py

"""Thin HTTP client for the agent → NAS ``agent-cron`` endpoints (Chronos).

The Chronos provider speaks ONLY to NAS — it names no scheduler vendor and
holds no scheduler credentials. NAS owns the external scheduler (an internal
implementation detail) and that scheduler's account; the agent just asks NAS to
"arm a one-shot at time T" / "cancel" / "list", authenticated with the agent's
existing Nous Portal access token (the same token it already uses to call the
portal — no new secret).

Wire contract: ``docs/chronos-managed-cron-contract.md``.
"""

from __future__ import annotations

import logging
from typing import Any, Dict, List, Optional

logger = logging.getLogger("cron.chronos")

# Endpoint paths under the portal base URL.
_PROVISION_PATH = "/api/agent-cron/provision"
_CANCEL_PATH = "/api/agent-cron/cancel"
_LIST_PATH = "/api/agent-cron/list"


class NasCronClientError(RuntimeError):
    """Raised when a NAS agent-cron call fails (non-2xx or transport error)."""


class NasCronClient:
    """Minimal client for the agent→NAS provision/cancel/list endpoints.

    Uses the agent's refresh-aware Nous access token for auth. No scheduler
    vendor, no scheduler creds — NAS hides all of that behind these three calls.
    """

    def __init__(self, portal_url: str, *, timeout_seconds: float = 15.0) -> None:
        self.portal_url = portal_url.rstrip("/")
        self.timeout_seconds = timeout_seconds

    # -- auth -------------------------------------------------------------

    def _access_token(self) -> str:
        """The agent's existing Nous Portal access token (refresh-aware)."""
        from hermes_cli.auth import resolve_nous_access_token
        return resolve_nous_access_token()

    def _headers(self) -> Dict[str, str]:
        return {
            "Authorization": f"Bearer {self._access_token()}",
            "Content-Type": "application/json",
        }

    # -- HTTP -------------------------------------------------------------

    def _post(self, path: str, body: Dict[str, Any]) -> Dict[str, Any]:
        import requests  # lazy: agent already depends on requests

        url = f"{self.portal_url}{path}"
        try:
            resp = requests.post(
                url, json=body, headers=self._headers(), timeout=self.timeout_seconds
            )
        except Exception as e:
            raise NasCronClientError(f"POST {path} failed: {e}") from e
        if resp.status_code // 100 != 2:
            raise NasCronClientError(
                f"POST {path} returned {resp.status_code}: {resp.text[:200]}"
            )
        try:
            return resp.json() if resp.content else {}
        except Exception:
            return {}

    def _get(self, path: str, params: Dict[str, Any]) -> Dict[str, Any]:
        import requests

        url = f"{self.portal_url}{path}"
        try:
            resp = requests.get(
                url, params=params, headers=self._headers(), timeout=self.timeout_seconds
            )
        except Exception as e:
            raise NasCronClientError(f"GET {path} failed: {e}") from e
        if resp.status_code // 100 != 2:
            raise NasCronClientError(
                f"GET {path} returned {resp.status_code}: {resp.text[:200]}"
            )
        try:
            return resp.json() if resp.content else {}
        except Exception:
            return {}

    # -- endpoints --------------------------------------------------------

    def provision(self, *, job_id: str, fire_at: str, agent_callback_url: str,
                  dedup_key: str) -> Dict[str, Any]:
        """Ask NAS to arm a one-shot for ``job_id`` at ``fire_at`` (ISO 8601).

        ``dedup_key`` (``{job_id}:{fire_at}``) makes re-arming the same fire
        idempotent NAS-side. Returns the NAS response (e.g. ``{schedule_id}``).
        """
        return self._post(_PROVISION_PATH, {
            "job_id": job_id,
            "fire_at": fire_at,
            "agent_callback_url": agent_callback_url,
            "dedup_key": dedup_key,
        })

    def cancel(self, *, job_id: str) -> Dict[str, Any]:
        """Ask NAS to cancel any armed one-shot for ``job_id``."""
        return self._post(_CANCEL_PATH, {"job_id": job_id})

    def list_armed(self) -> List[Dict[str, Any]]:
        """List the one-shots NAS currently has armed for this agent.

        Returns a list of ``{job_id, fire_at, schedule_id}``. Best-effort: used
        by reconcile to find orphaned arms on a cold process; on error the
        caller falls back to idempotent re-arm of all desired jobs.
        """
        data = self._get(_LIST_PATH, {})
        items = data.get("armed") if isinstance(data, dict) else None
        return items if isinstance(items, list) else []