client/opensoc_client.py

"""Thin HTTP client for the OpenSOC environment.

Importantly: this module **never imports server-side code** (`env.py`,
`verifier.py`, `rubric.py`).  The OpenEnv hackathon brief calls for
client/server separation so the same client can drive a remote HF Space
or a local container without re-running the verifier locally.

Usage::

    from client import OpenSOCClient
    c = OpenSOCClient(base_url="http://localhost:7860")
    obs = c.reset(task="stage1_basic", mode="defender_only", seed=1)
    result = c.step(
        {"submit_triage": {"action": "monitor",
                           "cited_log_id": "L1-0",
                           "rationale": "..."}},
        task="stage1_basic", mode="defender_only", seed=1,
    )
    grade = c.grade(task="stage1_basic", mode="defender_only", seed=1)
"""

from __future__ import annotations

import os
from typing import Any, Dict, Optional

import requests


class OpenSOCClient:
    """Lightweight requests-based client for the OpenSOC FastAPI server."""

    def __init__(
        self,
        base_url: Optional[str] = None,
        timeout: float = 30.0,
        session: Optional[requests.Session] = None,
    ):
        self.base_url = (base_url or os.getenv("OPENSOC_URL", "http://localhost:7860")).rstrip("/")
        self.timeout = timeout
        self.session = session or requests.Session()

    def health(self) -> Dict[str, Any]:
        return self._get("/health")

    def tasks(self) -> Dict[str, Any]:
        return self._get("/tasks")

    def reset(self, task: str = "stage1_basic", mode: str = "defender_only", seed: int = 0) -> Dict[str, Any]:
        return self._post("/reset", params={"task": task, "mode": mode, "seed": seed})

    def step(
        self,
        action: Dict[str, Any],
        task: str = "stage1_basic",
        mode: str = "defender_only",
        seed: int = 0,
    ) -> Dict[str, Any]:
        return self._post(
            "/step",
            params={"task": task, "mode": mode, "seed": seed},
            json=action,
        )

    def state(self, task: str = "stage1_basic", mode: str = "defender_only", seed: int = 0) -> Dict[str, Any]:
        return self._get("/state", params={"task": task, "mode": mode, "seed": seed})

    def grade(self, task: str = "stage1_basic", mode: str = "defender_only", seed: int = 0) -> Dict[str, Any]:
        return self._post("/grade", params={"task": task, "mode": mode, "seed": seed})

    def _get(self, path: str, params: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
        r = self.session.get(self.base_url + path, params=params, timeout=self.timeout)
        r.raise_for_status()
        return r.json()

    def _post(
        self,
        path: str,
        params: Optional[Dict[str, Any]] = None,
        json: Any = None,
    ) -> Dict[str, Any]:
        r = self.session.post(self.base_url + path, params=params, json=json, timeout=self.timeout)
        r.raise_for_status()
        return r.json()


__all__ = ["OpenSOCClient"]