feat: Add observability setup guide and integrate Prometheus and Grafana

- Introduced OBSERVABILITY_SETUP.md for comprehensive observability setup instructions.
- Added docker-compose configuration for Prometheus and Grafana.
- Created Grafana dashboards for AntiAtropos live control plane and Kubernetes overview.
- Updated README.md to reflect Kubernetes integration and observability features.
- Implemented Prometheus metrics endpoint in the AntiAtropos server.
- Enhanced Kubernetes integration with explicit workload mapping and telemetry label mapping.
- Added tests for Kubernetes integration and Prometheus adapter functionality.

control/kubernetes_executor.py

@@ -1,5 +1,7 @@
 import os
 import json
+import time
+from uuid import uuid4
 from typing import Optional
 
 class KubernetesExecutor:
@@ -12,47 +14,99 @@ class KubernetesExecutor:
         self.kubeconfig = kubeconfig or os.getenv("KUBECONFIG")
         self.is_mock = not self.kubeconfig or self.kubeconfig.lower() == "mock"
         self.namespace = os.getenv("ANTIATROPOS_K8S_NAMESPACE", "default")
-        self.deployment_prefix = os.getenv("ANTIATROPOS_DEPLOYMENT_PREFIX", "")
         self.min_replicas = int(os.getenv("ANTIATROPOS_MIN_REPLICAS", "1"))
         self.max_replicas = int(os.getenv("ANTIATROPOS_MAX_REPLICAS", "20"))
         self.scale_step = int(os.getenv("ANTIATROPOS_SCALE_STEP", "3"))
         self._apps_v1_api = None
-        self._node_deployment_map = self._load_node_deployment_map()
+        self._node_workload_map = self._load_node_workload_map()
+        self._live_supported_actions = {"NO_OP", "SCALE_UP", "SCALE_DOWN"}
+
+    @staticmethod
+    def _normalize_action_type(action_type) -> str:
+        if hasattr(action_type, "value"):
+            return str(action_type.value)
+        return str(action_type)
         
     def execute(self, action_type: str, target: str, parameter: float) -> str:
         """
         Translates SRE actions to Kube requests (ScaleDeployment, PatchIngress, etc.)
         """
+        return self.execute_with_metadata(action_type, target, parameter)["ack_status"]
+
+    def execute_with_metadata(self, action_type: str, target: str, parameter: float) -> dict:
+        """
+        Execute action and return acknowledgement plus executor metadata.
+        """
+        action_id = str(uuid4())
+        started = time.perf_counter()
+        ack_status = ""
+        error_code = ""
+
         if self.is_mock:
-            return self._mock_execution(action_type, target, parameter)
-            
-        try:
-            return self._real_execution(action_type, target, parameter)
-        except Exception as e:
-            return f"Error: Failed to execute {action_type} on {target}: {str(e)}"
+            ack_status = self._mock_execution(action_type, target, parameter)
+        else:
+            try:
+                ack_status = self._real_execution(action_type, target, parameter)
+            except Exception as e:
+                ack_status = f"Error: Failed to execute {action_type} on {target}: {str(e)}"
+                error_code = "EXECUTION_ERROR"
+
+        if ack_status.startswith("Rejected:") and not error_code:
+            error_code = "REJECTED_ACTION"
+        elif ack_status.startswith("Error:") and not error_code:
+            error_code = "EXECUTION_ERROR"
+
+        latency_ms = (time.perf_counter() - started) * 1000.0
+        return {
+            "action_id": action_id,
+            "ack_status": ack_status,
+            "executor_latency_ms": latency_ms,
+            "executor_error_code": error_code,
+        }
+
+    def live_enabled_actions(self) -> set[str]:
+        """Action types that are actually executable in real live mode."""
+        if self.is_mock:
+            return {"NO_OP"}
+        return set(self._live_supported_actions)
+
+    def live_capability_error(self, action_type: str) -> Optional[str]:
+        """Returns reason when action is not runnable in live mode, else None."""
+        action = self._normalize_action_type(action_type)
+        if action not in self.live_enabled_actions():
+            if self.is_mock:
+                return (
+                    f"Live mode rejected {action}: no real Kubernetes executor is configured "
+                    "(set KUBECONFIG and ANTIATROPOS_WORKLOAD_MAP)."
+                )
+            return f"Live mode rejected {action}: no executor is enabled for this action."
+        return None
 
     def _real_execution(self, action_type: str, target: str, parameter: float) -> str:
         """Execute bounded actions on a Kubernetes cluster."""
-        if action_type == "NO_OP":
+        action = self._normalize_action_type(action_type)
+
+        if action == "NO_OP":
             return "Ack: NO_OP - no cluster mutation"
 
-        if action_type in ("SCALE_UP", "SCALE_DOWN"):
-            return self._scale_deployment(action_type, target, parameter)
+        if action in ("SCALE_UP", "SCALE_DOWN"):
+            return self._scale_deployment(action, target, parameter)
 
-        return f"Rejected: {action_type} is not enabled for live Kubernetes execution"
+        return f"Rejected: {action} is not enabled for live Kubernetes execution"
 
     def _mock_execution(self, action_type: str, target: str, parameter: float) -> str:
         """Returns mock acknowledgement for actions."""
         # TODO: Add realistic latency simulation for K8s control plane
-        return f"Ack: {action_type} for {target} with value {parameter} - Status: Applied"
+        action = self._normalize_action_type(action_type)
+        return f"Ack: {action} for {target} with value {parameter} - Status: Applied"
 
     def _scale_deployment(self, action_type: str, target: str, parameter: float) -> str:
-        deployment_name = self._resolve_deployment_name(target)
+        namespace, deployment_name = self._resolve_workload_target(target)
         apps_v1 = self._get_apps_v1_api()
 
         scale_obj = apps_v1.read_namespaced_deployment_scale(
             name=deployment_name,
-            namespace=self.namespace,
+            namespace=namespace,
         )
 
         current = int(scale_obj.spec.replicas or self.min_replicas)
@@ -70,11 +124,14 @@ class KubernetesExecutor:
 
         apps_v1.patch_namespaced_deployment_scale(
             name=deployment_name,
-            namespace=self.namespace,
+            namespace=namespace,
             body={"spec": {"replicas": desired}},
         )
 
-        return f"Ack: {action_type} for {target} - deployment {deployment_name} scaled {current}->{desired}"
+        return (
+            f"Ack: {action_type} for {target} - deployment {deployment_name} "
+            f"in namespace {namespace} scaled {current}->{desired}"
+        )
 
     def _get_apps_v1_api(self):
         if self._apps_v1_api is not None:
@@ -90,19 +147,84 @@ class KubernetesExecutor:
         self._apps_v1_api = client.AppsV1Api()
         return self._apps_v1_api
 
-    def _load_node_deployment_map(self) -> dict[str, str]:
-        raw = os.getenv("ANTIATROPOS_NODE_DEPLOYMENT_MAP", "")
-        if not raw:
-            return {}
+    def _load_node_workload_map(self) -> dict[str, dict[str, str]]:
+        """
+        Load node->workload mapping.
+
+        Preferred format (ANTIATROPOS_WORKLOAD_MAP):
+        {
+          "node-0": {"deployment": "payments", "namespace": "prod-sre"},
+          "node-1": {"deployment": "checkout"}
+        }
+
+        Legacy fallback (ANTIATROPOS_NODE_DEPLOYMENT_MAP):
+        {
+          "node-0": "payments",
+          "node-1": "checkout"
+        }
+        """
+        raw = os.getenv("ANTIATROPOS_WORKLOAD_MAP", "")
+        if raw:
+            parsed = self._parse_json_mapping(raw)
+            if parsed is not None:
+                return parsed
+
+        legacy_raw = os.getenv("ANTIATROPOS_NODE_DEPLOYMENT_MAP", "")
+        if legacy_raw:
+            legacy = self._parse_legacy_mapping(legacy_raw)
+            if legacy is not None:
+                return legacy
+
+        return {}
+
+    def _parse_json_mapping(self, raw: str) -> Optional[dict[str, dict[str, str]]]:
         try:
             data = json.loads(raw)
-            if isinstance(data, dict):
-                return {str(k): str(v) for k, v in data.items()}
         except json.JSONDecodeError:
-            return {}
-        return {}
+            return None
+
+        if not isinstance(data, dict):
+            return None
+
+        out: dict[str, dict[str, str]] = {}
+        for node_id, workload in data.items():
+            if not isinstance(workload, dict):
+                return None
+            deployment = workload.get("deployment")
+            if not deployment:
+                return None
+            namespace = workload.get("namespace", self.namespace)
+            out[str(node_id)] = {
+                "deployment": str(deployment),
+                "namespace": str(namespace),
+            }
+        return out
+
+    def _parse_legacy_mapping(self, raw: str) -> Optional[dict[str, dict[str, str]]]:
+        try:
+            data = json.loads(raw)
+        except json.JSONDecodeError:
+            return None
+
+        if not isinstance(data, dict):
+            return None
+
+        out: dict[str, dict[str, str]] = {}
+        for node_id, deployment in data.items():
+            if not deployment:
+                return None
+            out[str(node_id)] = {
+                "deployment": str(deployment),
+                "namespace": self.namespace,
+            }
+        return out
+
+    def _resolve_workload_target(self, target: str) -> tuple[str, str]:
+        if target not in self._node_workload_map:
+            raise ValueError(
+                f"Missing workload mapping for target '{target}'. "
+                "Set ANTIATROPOS_WORKLOAD_MAP with node->deployment bindings."
+            )
 
-    def _resolve_deployment_name(self, target: str) -> str:
-        if target in self._node_deployment_map:
-            return self._node_deployment_map[target]
-        return f"{self.deployment_prefix}{target}"
+        workload = self._node_workload_map[target]
+        return workload["namespace"], workload["deployment"]

control/validation.py

@@ -1,5 +1,4 @@
 from typing import List, Optional
-from pydantic import BaseModel
 
 class ActionValidator:
     """
@@ -13,13 +12,27 @@ class ActionValidator:
         """
         Returns (is_valid, error_message).
         """
+        if hasattr(action_type, "value"):
+            action = str(action_type.value)
+        else:
+            action = str(action_type)
+
         if valid_targets is not None and target not in valid_targets:
             return False, f"Unknown target node: {target}"
 
-        if action_type == "SHED_LOAD" and target in self.critical_nodes:
+        if action == "SHED_LOAD" and target in self.critical_nodes:
             return False, f"Forbidden: Load shedding on critical node {target}."
-        
-        if action_type in ["SCALE_UP", "SCALE_DOWN"] and parameter < 0.0:
-            return False, "Negative scaling parameters are not allowed."
+
+        if action in ["SCALE_UP", "SCALE_DOWN"]:
+            if parameter < 0.0:
+                return False, "Negative scaling parameters are not allowed."
+            if parameter > 10.0:
+                return False, "Scaling parameter must be <= 10.0."
+
+        if action in ["REROUTE_TRAFFIC", "SHED_LOAD"] and not (0.0 <= parameter <= 1.0):
+            return False, f"{action} parameter must be in [0.0, 1.0]."
+
+        if action == "NO_OP" and parameter != 0.0:
+            return False, "NO_OP requires parameter=0.0."
             
         return True, "Success"

models.py

@@ -152,6 +152,9 @@ class ClusterObservation(BaseModel):
     metric_timestamp: float = 0.0
     data_freshness_ms: int = 0
     action_ack_status: str = "success"
+    action_id: str = ""
+    executor_latency_ms: float = Field(default=0.0, ge=0.0)
+    executor_error_code: str = ""
     choke_level: float = 0.0
 
     nodes: list[NodeObservation]

pyproject.toml

@@ -28,6 +28,7 @@ dependencies = [
     # "smolagents>=1.22.0,<2",
     "kubernetes>=28.0.0",
     "prometheus-api-client>=0.5.0",
+    "prometheus-client>=0.20.0",
     "requests>=2.31.0",
 ]
 

server/AntiAtropos_environment.py

@@ -1,4 +1,6 @@
 import time
+import json
+import logging
 from uuid import uuid4
 
 from openenv.core.env_server.interfaces import Environment
@@ -8,13 +10,13 @@ try:
     from ..models import SREAction, ClusterObservation, NodeObservation, NodeStatus, EnvironmentMode
     from ..simulator import ClusterSimulator, COST_PER_CAPACITY_UNIT_PER_HOUR
     from ..stability import compute_lyapunov, compute_reward
-    from ..telemetry import PrometheusClient
+    from ..telemetry import PrometheusClient, get_observability_tracker
     from ..control import KubernetesExecutor, ActionValidator
 except ImportError:
     from models import SREAction, ClusterObservation, NodeObservation, NodeStatus, EnvironmentMode  # type: ignore[no-redef]
     from simulator import ClusterSimulator, COST_PER_CAPACITY_UNIT_PER_HOUR  # type: ignore[no-redef]
     from stability import compute_lyapunov, compute_reward  # type: ignore[no-redef]
-    from telemetry import PrometheusClient  # type: ignore[no-redef]
+    from telemetry import PrometheusClient, get_observability_tracker  # type: ignore[no-redef]
     from control import KubernetesExecutor, ActionValidator  # type: ignore[no-redef]
 
 
@@ -57,12 +59,17 @@ class AntiAtroposEnvironment(Environment):
         self._telemetry = PrometheusClient()
         self._executor = KubernetesExecutor()
         self._validator = ActionValidator()
+        self._observability = get_observability_tracker()
+        self._logger = logging.getLogger("antiatropos.env")
         
         self._nodes_true: list[dict] = []
         self._nodes_obs: list[dict] = []
         self._prev_lyapunov: float = 0.0
         self._sla_violations: int = 0
         self._action_ack_status: str = "success"
+        self._last_action_id: str = ""
+        self._last_executor_latency_ms: float = 0.0
+        self._last_executor_error_code: str = ""
         self._last_metric_time: float = 0.0
 
     def reset(self, task_id: str = "task-1", mode: str = "simulated") -> ClusterObservation:
@@ -78,6 +85,9 @@ class AntiAtroposEnvironment(Environment):
             
         self._sla_violations = 0
         self._action_ack_status = "success"
+        self._last_action_id = ""
+        self._last_executor_latency_ms = 0.0
+        self._last_executor_error_code = ""
         
         # Only set baseline metric time for hybrid/live to prevent misleading freshness in SIM
         if self._mode in [EnvironmentMode.HYBRID, EnvironmentMode.LIVE]:
@@ -110,29 +120,54 @@ class AntiAtroposEnvironment(Environment):
         Advance the simulation by one discrete time tick.
         """
         self._state.step_count += 1
+        self._last_action_id = str(uuid4())
+        self._last_executor_latency_ms = 0.0
+        self._last_executor_error_code = ""
         
         # 1. Action Validation & Execution
         valid_targets = [n["node_id"] for n in self._nodes_true]
-        is_valid, error = self._validator.validate(
+        is_enabled, mode_error = self._is_action_enabled_for_mode(action.action_type)
+        if not is_enabled:
+            self._action_ack_status = f"Rejected: {mode_error}"
+            self._last_executor_error_code = "MODE_UNSUPPORTED"
+            is_valid = False
+            error = mode_error
+        else:
+            is_valid, error = self._validator.validate(
             action.action_type, 
             action.target_node_id, 
             action.parameter,
             valid_targets=valid_targets
-        )
+            )
         
+        apply_to_simulator = False
+
         if not is_valid:
             self._action_ack_status = f"Rejected: {error}"
+            if not self._last_executor_error_code:
+                self._last_executor_error_code = "VALIDATION_FAILED"
             # Increment invalid action count on the simulator so it's consistent
             self._sim.invalid_action_count += 1
             # Still advance time but the action didn't happen
         else:
             if self._mode == EnvironmentMode.LIVE:
                 # In LIVE mode, we actually hit the cluster
-                self._action_ack_status = self._executor.execute(action.action_type, action.target_node_id, action.parameter)
+                exec_result = self._executor.execute_with_metadata(
+                    action.action_type,
+                    action.target_node_id,
+                    action.parameter,
+                )
+                self._last_action_id = exec_result.get("action_id", self._last_action_id)
+                self._action_ack_status = exec_result.get("ack_status", "Error: missing ack status")
+                self._last_executor_latency_ms = float(exec_result.get("executor_latency_ms", 0.0))
+                self._last_executor_error_code = str(exec_result.get("executor_error_code", ""))
+                apply_to_simulator = self._action_ack_status.startswith("Ack:")
             else:
                 self._action_ack_status = "success (simulated)"
-            
-            # Always update the physics engine to keep tracking expectations
+                apply_to_simulator = True
+
+        # Keep simulator aligned with control-plane ack semantics.
+        if apply_to_simulator:
             self._sim.apply_action(action)
 
         # 2. Advance Physics
@@ -183,6 +218,46 @@ class AntiAtroposEnvironment(Environment):
         obs = self._build_observation()
         obs.done   = done
         obs.reward = reward
+
+        self._observability.record_step(
+            task_id=self._task_id,
+            mode=str(self._mode.value),
+            action_type=str(action.action_type.value),
+            target_node_id=str(action.target_node_id),
+            ack_status=self._action_ack_status,
+            reward=reward,
+            lyapunov_energy=obs.lyapunov_energy,
+            total_queue_backlog=obs.total_queue_backlog,
+            average_latency_ms=obs.average_latency_ms,
+            executor_latency_ms=self._last_executor_latency_ms,
+            executor_error_code=self._last_executor_error_code,
+        )
+
+        self._logger.info(
+            json.dumps(
+                {
+                    "event": "antiatropos_step",
+                    "episode_id": self._state.episode_id,
+                    "task_id": self._task_id,
+                    "mode": self._mode.value,
+                    "step": self._state.step_count,
+                    "action_type": action.action_type.value,
+                    "target_node_id": action.target_node_id,
+                    "parameter": float(action.parameter),
+                    "action_id": self._last_action_id,
+                    "action_ack_status": self._action_ack_status,
+                    "executor_latency_ms": self._last_executor_latency_ms,
+                    "executor_error_code": self._last_executor_error_code,
+                    "reward": reward,
+                    "lyapunov_energy": obs.lyapunov_energy,
+                    "average_latency_ms_norm": obs.average_latency_ms,
+                    "total_queue_backlog_norm": obs.total_queue_backlog,
+                    "error_rate": obs.error_rate,
+                    "done": done,
+                }
+            )
+        )
+
         return obs
 
     @property
@@ -192,6 +267,21 @@ class AntiAtroposEnvironment(Environment):
     # -----------------------------------------------------------------------
     # Logic Helpers
     # -----------------------------------------------------------------------
+    def _is_action_enabled_for_mode(self, action_type: str) -> tuple[bool, str]:
+        if hasattr(action_type, "value"):
+            action = str(action_type.value)
+        else:
+            action = str(action_type)
+        if self._mode in [EnvironmentMode.SIMULATED, EnvironmentMode.HYBRID]:
+            return True, "Enabled"
+
+        if self._mode == EnvironmentMode.LIVE:
+            capability_error = self._executor.live_capability_error(action)
+            if capability_error:
+                return False, capability_error
+            return True, "Enabled"
+
+        return False, f"Unsupported environment mode: {self._mode}"
 
     def _compute_cost(self, nodes_true: list[dict]) -> float:
         """Calculates current running infra cost using provisioned capacity units."""
@@ -271,6 +361,9 @@ class AntiAtroposEnvironment(Environment):
             metric_timestamp=self._last_metric_time,
             data_freshness_ms=freshness,
             action_ack_status=self._action_ack_status,
+            action_id=self._last_action_id,
+            executor_latency_ms=self._last_executor_latency_ms,
+            executor_error_code=self._last_executor_error_code,
             choke_level=0.0,
             done=False,
             reward=0.0,

server/Dockerfile

@@ -87,6 +87,7 @@ ENV ANTIATROPOS_DEPLOYMENT_PREFIX=""
 ENV ANTIATROPOS_MIN_REPLICAS="1"
 ENV ANTIATROPOS_MAX_REPLICAS="20"
 ENV ANTIATROPOS_SCALE_STEP="3"
+ENV ANTIATROPOS_WORKLOAD_MAP="{}"
 ENV ANTIATROPOS_NODE_DEPLOYMENT_MAP="{}"
 
 # Health check

server/app.py

@@ -39,10 +39,12 @@ try:
     # Change these names to match models.py
     from ..models import SREAction, ClusterObservation
     from .AntiAtropos_environment import AntiAtroposEnvironment
+    from ..telemetry import render_prometheus_metrics
 except (ModuleNotFoundError, ImportError):
     # And here as well
     from models import SREAction, ClusterObservation
     from server.AntiAtropos_environment import AntiAtroposEnvironment
+    from telemetry import render_prometheus_metrics
 
 
 # Create the app with web interface and README integration
@@ -56,6 +58,14 @@ app = create_app(
 )
 
 
+@app.get("/metrics")
+def metrics():
+    from fastapi import Response
+
+    payload = render_prometheus_metrics()
+    return Response(content=payload, media_type="text/plain; version=0.0.4; charset=utf-8")
+
+
 def main(host: str = "0.0.0.0", port: int = 8000):
     """
     Entry point for direct execution via uv run or python -m.

server/requirements.txt

@@ -3,4 +3,5 @@ fastapi>=0.115.0
 uvicorn>=0.24.0
 kubernetes>=28.0.0
 prometheus-api-client>=0.5.0
+prometheus-client>=0.20.0
 requests>=2.31.0

telemetry/__init__.py

@@ -1,2 +1,3 @@
 from .prometheus_client import PrometheusClient, TelemetryRecord
 from .mapping import MetricMapper
+from .observability import get_observability_tracker, render_prometheus_metrics

telemetry/mapping.py

@@ -1,5 +1,6 @@
-from typing import Dict, List, Any
+from typing import Dict, List, Any, Optional
 import os
+import json
 
 class MetricMapper:
     """
@@ -8,20 +9,86 @@ class MetricMapper:
     aggregate for a given node_id.
     """
     def __init__(self, mapping_strategy: str = "sum"):
-        self.strategy = mapping_strategy
+        self.strategy = mapping_strategy.lower()
         self.node_mapping = self._load_node_mapping()
 
     def _load_node_mapping(self) -> Dict[str, str]:
-        """Loads pod-to-node mapping from config/labels."""
-        # Simple default mapping for demonstration
+        """Loads label-value -> node_id mapping from env config."""
+        raw = os.getenv("ANTIATROPOS_LABEL_NODE_MAP", "")
+        if raw:
+            try:
+                data = json.loads(raw)
+                if isinstance(data, dict):
+                    return {str(k): str(v) for k, v in data.items()}
+            except json.JSONDecodeError:
+                pass
+
+        # Safe default mapping for local demos.
         return {
             "web-node-1": "node-0",
             "api-node-2": "node-1",
             "db-node-3": "node-2"
         }
 
+    def _resolve_node_id(self, labels: Dict[str, Any]) -> Optional[str]:
+        """Resolve internal node_id from a Prometheus sample labelset."""
+        explicit = labels.get("node_id")
+        if explicit:
+            return str(explicit)
+
+        for key in ("pod", "service", "app", "workload", "deployment", "instance"):
+            label_value = labels.get(key)
+            if label_value and str(label_value) in self.node_mapping:
+                return self.node_mapping[str(label_value)]
+
+        return None
+
+    def _reduce(self, values: List[float]) -> float:
+        if not values:
+            return 0.0
+        if self.strategy == "max":
+            return max(values)
+        if self.strategy == "mean":
+            return sum(values) / len(values)
+        # default: sum
+        return sum(values)
+
     def aggregate_node_metrics(self, raw_metrics: List[Dict[str, Any]]) -> Dict[str, Dict[str, float]]:
-        """Aggregates pod-level metrics into node-level telemetry."""
-        # TODO: Implement aggregation based on strategy (sum, mean, max)
-        # For now, we return a structural placeholder
-        return {}
+        """
+        Aggregates labeled metric samples into node-level telemetry.
+
+        Expected sample shape:
+        {
+          "metric_name": "request_rate",
+          "labels": {"pod": "web-node-1"},
+          "value": 42.0
+        }
+        """
+        bucket: Dict[str, Dict[str, List[float]]] = {}
+
+        for sample in raw_metrics:
+            metric_name = str(sample.get("metric_name", ""))
+            labels = sample.get("labels") or {}
+            value = sample.get("value")
+            if not metric_name or not isinstance(labels, dict) or value is None:
+                continue
+
+            node_id = self._resolve_node_id(labels)
+            if not node_id:
+                continue
+
+            try:
+                val = float(value)
+            except (TypeError, ValueError):
+                continue
+
+            bucket.setdefault(node_id, {}).setdefault(metric_name, []).append(val)
+
+        aggregated: Dict[str, Dict[str, float]] = {}
+        for node_id, metric_map in bucket.items():
+            aggregated[node_id] = {
+                metric_name: self._reduce(values)
+                for metric_name, values in metric_map.items()
+            }
+
+        return aggregated

telemetry/observability.py

@@ -0,0 +1,127 @@
+import threading
+from typing import Optional
+
+try:
+    from prometheus_client import Counter, Gauge, Histogram, generate_latest
+except ImportError:  # pragma: no cover
+    Counter = Gauge = Histogram = None
+
+    def generate_latest() -> bytes:  # type: ignore[override]
+        return b"# prometheus_client not installed\n"
+
+
+def _enabled() -> bool:
+    return Counter is not None and Gauge is not None and Histogram is not None
+
+
+class ObservabilityTracker:
+    """Prometheus metrics for action/reward/health monitoring."""
+
+    def __init__(self):
+        self._lock = threading.Lock()
+        self._is_enabled = _enabled()
+        if not self._is_enabled:
+            return
+
+        self.steps_total = Counter(
+            "antiatropos_steps_total",
+            "Total environment steps",
+            ["task_id", "mode"],
+        )
+        self.actions_total = Counter(
+            "antiatropos_actions_total",
+            "Actions executed by type/target/status",
+            ["task_id", "mode", "action_type", "target_node_id", "ack_class"],
+        )
+        self.executor_errors_total = Counter(
+            "antiatropos_executor_errors_total",
+            "Executor errors by code",
+            ["mode", "error_code"],
+        )
+        self.executor_latency_ms = Histogram(
+            "antiatropos_executor_latency_ms",
+            "Executor latency in milliseconds",
+            ["mode"],
+            buckets=(1, 5, 10, 25, 50, 100, 250, 500, 1000, 2500, 5000),
+        )
+
+        self.reward_gauge = Gauge(
+            "antiatropos_reward",
+            "Latest reward value",
+            ["task_id", "mode"],
+        )
+        self.lyapunov_gauge = Gauge(
+            "antiatropos_lyapunov_energy",
+            "Latest Lyapunov energy",
+            ["task_id", "mode"],
+        )
+        self.queue_gauge = Gauge(
+            "antiatropos_total_queue_backlog",
+            "Latest normalized total queue backlog",
+            ["task_id", "mode"],
+        )
+        self.latency_gauge = Gauge(
+            "antiatropos_average_latency_norm",
+            "Latest normalized average latency",
+            ["task_id", "mode"],
+        )
+
+    def record_step(
+        self,
+        task_id: str,
+        mode: str,
+        action_type: str,
+        target_node_id: str,
+        ack_status: str,
+        reward: float,
+        lyapunov_energy: float,
+        total_queue_backlog: float,
+        average_latency_ms: float,
+        executor_latency_ms: float,
+        executor_error_code: str,
+    ) -> None:
+        if not self._is_enabled:
+            return
+
+        ack_class = self._classify_ack(ack_status)
+        with self._lock:
+            self.steps_total.labels(task_id=task_id, mode=mode).inc()
+            self.actions_total.labels(
+                task_id=task_id,
+                mode=mode,
+                action_type=action_type,
+                target_node_id=target_node_id,
+                ack_class=ack_class,
+            ).inc()
+            self.reward_gauge.labels(task_id=task_id, mode=mode).set(float(reward))
+            self.lyapunov_gauge.labels(task_id=task_id, mode=mode).set(float(lyapunov_energy))
+            self.queue_gauge.labels(task_id=task_id, mode=mode).set(float(total_queue_backlog))
+            self.latency_gauge.labels(task_id=task_id, mode=mode).set(float(average_latency_ms))
+            self.executor_latency_ms.labels(mode=mode).observe(max(0.0, float(executor_latency_ms)))
+            if executor_error_code:
+                self.executor_errors_total.labels(mode=mode, error_code=executor_error_code).inc()
+
+    @staticmethod
+    def _classify_ack(ack_status: str) -> str:
+        status = str(ack_status)
+        if status.startswith("Ack:") or status.startswith("success"):
+            return "ack"
+        if status.startswith("Rejected:"):
+            return "rejected"
+        if status.startswith("Error:"):
+            return "error"
+        return "unknown"
+
+
+_TRACKER: Optional[ObservabilityTracker] = None
+
+
+def get_observability_tracker() -> ObservabilityTracker:
+    global _TRACKER
+    if _TRACKER is None:
+        _TRACKER = ObservabilityTracker()
+    return _TRACKER
+
+
+def render_prometheus_metrics() -> bytes:
+    return generate_latest()

telemetry/prometheus_client.py

@@ -1,8 +1,9 @@
 import os
 import random
-from typing import Dict, List, Optional
+from typing import Any, Dict, List, Optional
 import requests
 from pydantic import BaseModel
+from .mapping import MetricMapper
 
 class TelemetryRecord(BaseModel):
     node_id: str
@@ -23,26 +24,29 @@ class PrometheusClient:
         self.is_mock = not self.url or self.url.lower() == "mock"
         self.timeout_s = float(os.getenv("ANTIATROPOS_PROM_TIMEOUT_S", "2.5"))
         self.strict_real = os.getenv("ANTIATROPOS_STRICT_REAL", "false").lower() == "true"
+        self.metric_mapper = MetricMapper(
+            mapping_strategy=os.getenv("ANTIATROPOS_METRIC_AGGREGATION", "sum")
+        )
 
         self.request_rate_query = os.getenv(
             "ANTIATROPOS_PROM_QUERY_REQUEST_RATE",
-            'sum(rate(http_requests_total{node_id="{node_id}"}[1m]))'
+            'sum(rate(http_requests_total[1m])) by (pod)'
         )
         self.latency_ms_query = os.getenv(
             "ANTIATROPOS_PROM_QUERY_LATENCY_MS",
-            'histogram_quantile(0.95, sum(rate(http_request_duration_seconds_bucket{node_id="{node_id}"}[5m])) by (le)) * 1000'
+            'histogram_quantile(0.95, sum(rate(http_request_duration_seconds_bucket[5m])) by (pod, le)) * 1000'
         )
         self.error_rate_query = os.getenv(
             "ANTIATROPOS_PROM_QUERY_ERROR_RATE",
-            'sum(rate(http_requests_total{node_id="{node_id}",status=~"5.."}[1m])) / clamp_min(sum(rate(http_requests_total{node_id="{node_id}"}[1m])), 1)'
+            'sum(rate(http_requests_total{status=~"5.."}[1m])) by (pod) / clamp_min(sum(rate(http_requests_total[1m])) by (pod), 1)'
         )
         self.cpu_query = os.getenv(
             "ANTIATROPOS_PROM_QUERY_CPU",
-            'avg(rate(container_cpu_usage_seconds_total{pod=~".*{node_id}.*"}[1m]))'
+            'avg(rate(container_cpu_usage_seconds_total[1m])) by (pod)'
         )
         self.queue_depth_query = os.getenv(
             "ANTIATROPOS_PROM_QUERY_QUEUE_DEPTH",
-            'sum(queue_depth{node_id="{node_id}"})'
+            'sum(queue_depth) by (pod)'
         )
         
     def fetch_latest_metrics(self, node_ids: List[str]) -> Dict[str, TelemetryRecord]:
@@ -66,14 +70,20 @@ class PrometheusClient:
         metrics: Dict[str, TelemetryRecord] = {}
         saw_any_real_signal = False
 
+        req_by_node = self._collect_metric_values("request_rate", self.request_rate_query, node_ids)
+        lat_by_node = self._collect_metric_values("latency_ms", self.latency_ms_query, node_ids)
+        err_by_node = self._collect_metric_values("error_rate", self.error_rate_query, node_ids)
+        cpu_by_node = self._collect_metric_values("cpu_utilization", self.cpu_query, node_ids)
+        q_by_node = self._collect_metric_values("queue_depth", self.queue_depth_query, node_ids)
+
         for node_id in node_ids:
-            req_rate = self._query_scalar(self.request_rate_query.format(node_id=node_id))
-            lat_ms = self._query_scalar(self.latency_ms_query.format(node_id=node_id))
-            err_rate = self._query_scalar(self.error_rate_query.format(node_id=node_id))
-            cpu = self._query_scalar(self.cpu_query.format(node_id=node_id))
-            q_depth = self._query_scalar(self.queue_depth_query.format(node_id=node_id))
+            req_rate = req_by_node.get(node_id)
+            lat_ms = lat_by_node.get(node_id)
+            err_rate = err_by_node.get(node_id)
+            cpu = cpu_by_node.get(node_id)
+            q_depth = q_by_node.get(node_id)
 
-            if any(v is not None for v in [req_rate, lat_ms, err_rate, cpu, q_depth]):
+            if any(v is not None for v in (req_rate, lat_ms, err_rate, cpu, q_depth)):
                 saw_any_real_signal = True
 
             metrics[node_id] = TelemetryRecord(
@@ -90,6 +100,50 @@ class PrometheusClient:
 
         return metrics
 
+    def _collect_metric_values(
+        self,
+        metric_name: str,
+        query: str,
+        node_ids: List[str],
+    ) -> Dict[str, Optional[float]]:
+        """
+        Collect node values for one logical metric.
+
+        If query contains "{node_id}", execute per-node scalar queries.
+        Otherwise run one vector query and aggregate labels via MetricMapper.
+        """
+        out: Dict[str, Optional[float]] = {node_id: None for node_id in node_ids}
+
+        if "{node_id}" in query:
+            for node_id in node_ids:
+                out[node_id] = self._query_scalar(query.format(node_id=node_id))
+            return out
+
+        samples = self._query_vector(query)
+        raw_metrics: List[Dict[str, Any]] = []
+        for sample in samples:
+            labels = sample.get("metric")
+            value = sample.get("value")
+            if not isinstance(labels, dict):
+                continue
+            if not value or len(value) < 2:
+                continue
+            raw_metrics.append(
+                {
+                    "metric_name": metric_name,
+                    "labels": labels,
+                    "value": value[1],
+                }
+            )
+
+        by_node = self.metric_mapper.aggregate_node_metrics(raw_metrics)
+        for node_id in node_ids:
+            metric_map = by_node.get(node_id, {})
+            value = metric_map.get(metric_name)
+            out[node_id] = value if value is not None else None
+
+        return out
+
     def _query_scalar(self, promql: str) -> Optional[float]:
         """Runs a scalar/vector Prometheus instant query and returns the first value."""
         if not self.url:
@@ -119,6 +173,23 @@ class PrometheusClient:
         except (TypeError, ValueError):
             return None
 
+    def _query_vector(self, promql: str) -> List[Dict[str, Any]]:
+        """Runs a Prometheus instant query and returns the full vector result list."""
+        if not self.url:
+            return []
+
+        response = requests.get(
+            f"{self.url.rstrip('/')}/api/v1/query",
+            params={"query": promql},
+            timeout=self.timeout_s,
+        )
+        response.raise_for_status()
+        payload = response.json()
+        if payload.get("status") != "success":
+            return []
+        result = payload.get("data", {}).get("result", [])
+        return result if isinstance(result, list) else []
+
     def _generate_mock_metrics(self, node_ids: List[str]) -> Dict[str, TelemetryRecord]:
         """Generates realistic-looking mock telemetry."""
         metrics = {}