rein-aharness/rein_aharness/ops_run_client.py

"""activity-core ops_run claim client (REIN-A-0002 / ACT-ADR-005).

Primary intake for scheduled automation. Does **not** use issue-core or Forgejo.

Environment:

  ACTIVITY_CORE_URL              default http://127.0.0.1:8010
  ACTIVITY_CORE_WORKER_TOKEN     X-Worker-Token / Bearer (optional if API open)
  AGENT_HARNESS_WORKER_ID        claim owner (default: rein-aharness@hostname)
  AGENT_HARNESS_OPS_LABELS       comma labels for claim filter (default: automated)
  AGENT_HARNESS_OPS_LABELS_MODE  any|all (default: any)
  AGENT_HARNESS_OPS_LEASE_SECONDS  claim lease (default: 900)
  AGENT_HARNESS_REPO_MAP / AGENT_HARNESS_REPO_ROOTS  shared with intake
"""

from __future__ import annotations

import json
import os
import socket
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any

import httpx

from rein_aharness.intake import resolve_target_repo
from rein_aharness.taskspec import TaskSpec, TaskSpecError

DEFAULT_ACTIVITY_CORE_URL = "http://127.0.0.1:8010"
DEFAULT_OPS_LABELS = ("automated",)
DEFAULT_REPO_ROOTS = ("~", "~/work")


class OpsRunError(RuntimeError):
    """Bounded Activity Core failure with optional HTTP classification."""

    def __init__(
        self,
        message: str,
        *,
        action: str | None = None,
        status_code: int | None = None,
    ) -> None:
        super().__init__(message)
        self.action = action
        self.status_code = status_code

    @property
    def lease_rejected(self) -> bool:
        return self.status_code in {401, 403, 404, 409, 410, 412, 423}


@dataclass
class OpsRun:
    """Normalized ops_run row from actcore-api."""

    id: str
    activity_definition_id: str
    idempotency_key: str
    target_repo: str | None
    title: str
    description: str
    labels: list[str] = field(default_factory=list)
    priority: str = "medium"
    state: str = "open"
    claim_owner: str | None = None
    lease_until: str | None = None
    attempt: int = 0
    source_type: str = "rule"
    source_id: str = ""
    triggering_event_id: str = ""
    approach_hint: str | None = None
    harness_profile_ref: str | None = None
    execution_refs: dict[str, Any] = field(default_factory=dict)
    result: dict[str, Any] = field(default_factory=dict)
    raw: dict[str, Any] = field(default_factory=dict)

    @classmethod
    def from_api(cls, data: dict[str, Any]) -> "OpsRun":
        return cls(
            id=str(data.get("id") or ""),
            activity_definition_id=str(data.get("activity_definition_id") or ""),
            idempotency_key=str(data.get("idempotency_key") or ""),
            target_repo=data.get("target_repo"),
            title=str(data.get("title") or ""),
            description=str(data.get("description") or ""),
            labels=[str(x) for x in (data.get("labels") or [])],
            priority=str(data.get("priority") or "medium"),
            state=str(data.get("state") or "open"),
            claim_owner=data.get("claim_owner"),
            lease_until=data.get("lease_until"),
            attempt=int(data.get("attempt") or 0),
            source_type=str(data.get("source_type") or "rule"),
            source_id=str(data.get("source_id") or ""),
            triggering_event_id=str(data.get("triggering_event_id") or ""),
            approach_hint=data.get("approach_hint"),
            harness_profile_ref=data.get("harness_profile_ref"),
            execution_refs=dict(data.get("execution_refs") or {}),
            result=dict(data.get("result") or {}),
            raw=data,
        )


@dataclass
class OpsRunConfig:
    base_url: str = DEFAULT_ACTIVITY_CORE_URL
    worker_token: str = ""
    worker_id: str = "rein-aharness"
    claim_labels: tuple[str, ...] = DEFAULT_OPS_LABELS
    labels_mode: str = "any"
    lease_seconds: int = 900
    repo_map: dict[str, str] = field(default_factory=dict)
    repo_roots: tuple[str, ...] = DEFAULT_REPO_ROOTS
    timeout: float = 30.0

    @classmethod
    def from_env(cls) -> "OpsRunConfig":
        labels_raw = os.environ.get("AGENT_HARNESS_OPS_LABELS", "automated")
        labels = tuple(p.strip() for p in labels_raw.split(",") if p.strip())
        roots_raw = os.environ.get("AGENT_HARNESS_REPO_ROOTS", "~:~/work")
        roots = tuple(p.strip() for p in roots_raw.split(":") if p.strip())
        repo_map: dict[str, str] = {}
        map_raw = os.environ.get("AGENT_HARNESS_REPO_MAP", "").strip()
        if map_raw:
            repo_map = {str(k): str(v) for k, v in json.loads(map_raw).items()}
        host = socket.gethostname().split(".")[0]
        default_worker = f"rein-aharness@{host}"
        try:
            lease = max(30, int(os.environ.get("AGENT_HARNESS_OPS_LEASE_SECONDS", "900")))
        except ValueError:
            lease = 900
        mode = (os.environ.get("AGENT_HARNESS_OPS_LABELS_MODE") or "any").strip().lower()
        if mode not in {"any", "all"}:
            mode = "any"
        return cls(
            base_url=os.environ.get(
                "ACTIVITY_CORE_URL", DEFAULT_ACTIVITY_CORE_URL
            ).rstrip("/"),
            worker_token=(
                os.environ.get("ACTIVITY_CORE_WORKER_TOKEN")
                or os.environ.get("AGENT_HARNESS_WORKER_TOKEN")
                or ""
            ).strip(),
            worker_id=os.environ.get("AGENT_HARNESS_WORKER_ID", default_worker).strip()
            or default_worker,
            claim_labels=labels or DEFAULT_OPS_LABELS,
            labels_mode=mode,
            lease_seconds=lease,
            repo_map=repo_map,
            repo_roots=roots or DEFAULT_REPO_ROOTS,
        )


class ActivityCoreOpsClient:
    """REST client for POST /ops-runs/claim|heartbeat|complete|fail."""

    def __init__(self, config: OpsRunConfig | None = None):
        self.config = config or OpsRunConfig.from_env()

    def _headers(self) -> dict[str, str]:
        headers = {"Accept": "application/json", "Content-Type": "application/json"}
        if self.config.worker_token:
            headers["X-Worker-Token"] = self.config.worker_token
            headers["Authorization"] = f"Bearer {self.config.worker_token}"
        return headers

    def list_open(self, *, limit: int = 50) -> list[OpsRun]:
        try:
            resp = httpx.get(
                f"{self.config.base_url}/ops-runs",
                params={"state": "open", "limit": str(limit)},
                headers=self._headers(),
                timeout=self.config.timeout,
            )
            resp.raise_for_status()
        except httpx.HTTPStatusError as exc:
            raise OpsRunError(
                f"list ops-runs failed: HTTP {exc.response.status_code}",
                action="list",
                status_code=exc.response.status_code,
            ) from exc
        except httpx.HTTPError as exc:
            raise OpsRunError("list ops-runs failed: transport error", action="list") from exc
        data = resp.json()
        items = data.get("items") if isinstance(data, dict) else data
        if not isinstance(items, list):
            raise OpsRunError(f"unexpected list payload: {type(data)}")
        return [OpsRun.from_api(item) for item in items if isinstance(item, dict)]

    def claim(
        self,
        *,
        labels: list[str] | None = None,
        labels_mode: str | None = None,
        limit: int = 1,
        lease_seconds: int | None = None,
    ) -> list[OpsRun]:
        body = {
            "worker_id": self.config.worker_id,
            "labels": list(labels if labels is not None else self.config.claim_labels),
            "labels_mode": labels_mode or self.config.labels_mode,
            "limit": max(1, min(limit, 20)),
            "lease_seconds": lease_seconds or self.config.lease_seconds,
        }
        try:
            resp = httpx.post(
                f"{self.config.base_url}/ops-runs/claim",
                json=body,
                headers=self._headers(),
                timeout=self.config.timeout,
            )
            resp.raise_for_status()
        except httpx.HTTPStatusError as exc:
            raise OpsRunError(
                f"claim failed: HTTP {exc.response.status_code}",
                action="claim",
                status_code=exc.response.status_code,
            ) from exc
        except httpx.HTTPError as exc:
            raise OpsRunError("claim failed: transport error", action="claim") from exc
        data = resp.json()
        items = data.get("items") if isinstance(data, dict) else []
        return [OpsRun.from_api(item) for item in items if isinstance(item, dict)]

    def heartbeat(
        self,
        run_id: str,
        *,
        lease_seconds: int | None = None,
    ) -> OpsRun:
        body: dict[str, Any] = {"worker_id": self.config.worker_id}
        if lease_seconds is not None:
            body["lease_seconds"] = lease_seconds
        return self._post_run(run_id, "heartbeat", body)

    def complete(
        self,
        run_id: str,
        *,
        result: dict[str, Any] | None = None,
    ) -> OpsRun:
        return self._post_run(
            run_id,
            "complete",
            {"worker_id": self.config.worker_id, "result": result or {}},
        )

    def fail(
        self,
        run_id: str,
        *,
        error: str = "",
        reopen: bool = False,
        result: dict[str, Any] | None = None,
    ) -> OpsRun:
        return self._post_run(
            run_id,
            "fail",
            {
                "worker_id": self.config.worker_id,
                "error": error,
                "reopen": reopen,
                "result": result or {},
            },
        )

    def _post_run(self, run_id: str, action: str, body: dict[str, Any]) -> OpsRun:
        try:
            resp = httpx.post(
                f"{self.config.base_url}/ops-runs/{run_id}/{action}",
                json=body,
                headers=self._headers(),
                timeout=self.config.timeout,
            )
            resp.raise_for_status()
        except httpx.HTTPStatusError as exc:
            raise OpsRunError(
                f"{action} ops_run {run_id} failed: HTTP {exc.response.status_code}",
                action=action,
                status_code=exc.response.status_code,
            ) from exc
        except httpx.HTTPError as exc:
            raise OpsRunError(
                f"{action} ops_run {run_id} failed: transport error", action=action
            ) from exc
        return OpsRun.from_api(resp.json())


def ops_run_to_taskspec(
    run: OpsRun,
    config: OpsRunConfig | None = None,
    *,
    agent: str = "coach",
    completion_event_type: str = "executor_run",
    timeout_seconds: int = 900,
) -> TaskSpec:
    """Map claimed ops_run to TaskSpec for agent-session approach."""
    cfg = config or OpsRunConfig.from_env()
    if not run.target_repo:
        raise TaskSpecError(f"ops_run {run.id} missing target_repo")
    target = resolve_target_repo(
        run.target_repo,
        repo_map=cfg.repo_map,
        repo_roots=cfg.repo_roots,
    )
    return TaskSpec(
        title=run.title or "(untitled)",
        description=run.description or "",
        target_repo=target,
        agent=agent,
        labels=list(run.labels),
        hub_task_id=run.id,
        completion_event_type=completion_event_type,
        timeout_seconds=timeout_seconds,
    )


def resolve_ops_target(run: OpsRun, config: OpsRunConfig | None = None) -> Path:
    cfg = config or OpsRunConfig.from_env()
    if not run.target_repo:
        raise TaskSpecError(f"ops_run {run.id} missing target_repo")
    return resolve_target_repo(
        run.target_repo,
        repo_map=cfg.repo_map,
        repo_roots=cfg.repo_roots,
    )