signal human review done

agentx/monitor/client.py

@@ -3,7 +3,7 @@
 import logging
 import os
 import time
-from typing import Any, List, Optional
+from typing import Any, Dict, List, Optional
 
 import requests
 
@@ -116,6 +116,10 @@ def __init__(
         # The human-review queue (list / queue / label / dismiss) - what makes the
         # label-and-calibrate loop scriptable instead of dashboard-only.
         self.review_queue = ReviewQueueClient(self)
+        from agentx.monitor.rules import MonitorRulesClient
+
+        # Automation rules: route matching traffic into review / a dataset / a webhook.
+        self.rules = MonitorRulesClient(self)
         from agentx.monitor.scorers import ScorersClient
         # Scorers-catalog administration as code: template enable/disable, code/external scorer
         # CRUD and dry runs - full parity with the dashboard's Scorers page (P1.3).
@@ -264,6 +268,28 @@ def kpis(self, window: str = "7d") -> dict:
         plus deltas vs the prior window and the run-outcome breakdown."""
         return self._request("GET", "/kpis", params={"window": window})
 
+    def topics(self, window: str = "7d") -> dict:
+        """The Topics view's data over a window ("24h", "7d", "30d"): LLM-classified themes of
+        sampled production traffic with per-topic counts and sentiment. Empty until Topics is
+        enabled project-wide via ``set_topics(True)`` - classification spends one judge call
+        per sampled trace, so it is off by default."""
+        return self._request(
+            "GET", "/agent-monitoring/topics",
+            base=self._api_root(), params={"window": window},
+        )
+
+    def set_topics(self, enabled: bool, sample_rate: Optional[float] = None) -> dict:
+        """Turn Topics classification on/off for the whole project (Platform Settings >
+        Monitoring Defaults). ``sample_rate`` (0-1) optionally bounds what fraction of traffic
+        is classified - each classified trace costs one judge call."""
+        payload: Dict[str, Any] = {"topicsEnabled": enabled}
+        if sample_rate is not None:
+            payload["topicsSampleRate"] = sample_rate
+        return self._request(
+            "PUT", "/agent-monitoring/settings/monitoring-defaults",
+            base=self._api_root(), json=payload,
+        )
+
     def calibration(self, window: str = "7d") -> "CalibrationSummary":
         """Project-level judge calibration over a window ("24h", "7d", or "30d"): how often
         AgentX's own verdicts agreed with real-world ground truth reported later (ops outcomes

agentx/monitor/rules.py

@@ -0,0 +1,78 @@
+from __future__ import annotations
+
+import logging
+from typing import Any, Dict, List, Optional, TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from agentx.monitor.client import MonitorClient
+
+logger = logging.getLogger(__name__)
+
+
+class MonitorRule(dict):
+    """Wire object for one automation rule (dict subclass so unknown fields round-trip)."""
+
+    @property
+    def id(self) -> str:
+        return self["_id"]
+
+    @property
+    def enabled(self) -> bool:
+        return bool(self.get("enabled"))
+
+
+class MonitorRulesClient:
+    """Surfaced as ``client.monitor.rules``: automation rules, evaluated on every ingested root
+    trace. A RULE routes traffic somewhere (it never scores): ``action`` is one of
+
+    - ``"review"`` - sample matching traces into the human-review queue (the stream that feeds
+      judge calibration and tuning),
+    - ``"dataset"`` - append matching traces as cases on a dataset (``action_config
+      {"datasetId": ...}``),
+    - ``"webhook"`` - POST the matching trace to your URL (``action_config {"url": ...}``).
+
+    ``filter`` narrows what matches: ``{"model": ..., "status": "error"|"any", "contains": ...,
+    "scopeMode": "all"|"selected", "agentIds": [...]}``; ``sample_rate`` (0-1, default 1)
+    down-samples the matches.
+    """
+
+    def __init__(self, client: "MonitorClient"):
+        self._client = client
+
+    def _request(self, method: str, path: str, **kwargs: Any) -> Any:
+        return self._client._request(method, path, base=self._client._api_root(), **kwargs)
+
+    def list(self) -> List[MonitorRule]:
+        data = self._request("GET", "/agent-monitoring/rules")
+        return [MonitorRule(r) for r in data.get("rules", [])]
+
+    def create(
+        self,
+        name: str,
+        action: str,
+        *,
+        filter: Optional[Dict[str, Any]] = None,
+        sample_rate: Optional[float] = None,
+        action_config: Optional[Dict[str, Any]] = None,
+        enabled: bool = True,
+    ) -> MonitorRule:
+        payload: Dict[str, Any] = {"name": name, "action": action, "enabled": enabled}
+        if filter is not None:
+            payload["filter"] = filter
+        if sample_rate is not None:
+            payload["sampleRate"] = sample_rate
+        if action_config is not None:
+            payload["actionConfig"] = action_config
+        data = self._request("POST", "/agent-monitoring/rules", json=payload)
+        return MonitorRule(data.get("rule", data))
+
+    def update(self, rule_id: str, **fields: Any) -> MonitorRule:
+        """Sparse update. snake_case keys are mapped to the wire (``sample_rate`` ->
+        ``sampleRate``, ``action_config`` -> ``actionConfig``)."""
+        aliases = {"sample_rate": "sampleRate", "action_config": "actionConfig"}
+        payload = {aliases.get(k, k): v for k, v in fields.items()}
+        data = self._request("PUT", f"/agent-monitoring/rules/{rule_id}", json=payload)
+        return MonitorRule(data.get("rule", data))
+
+    def delete(self, rule_id: str) -> None:
+        self._request("DELETE", f"/agent-monitoring/rules/{rule_id}")