basic of grafan, kuber, and prometheus added

client.py

@@ -94,6 +94,7 @@ class AntiAtroposEnv(
         observation = ClusterObservation(
             cluster_id=obs_data.get("cluster_id", ""),
             task_id=obs_data.get("task_id", "task-1"),
+            mode=obs_data.get("mode", "simulated"),
             active_nodes=obs_data.get("active_nodes", 0),
             average_latency_ms=obs_data.get("average_latency_ms", 0.0),
             error_rate=obs_data.get("error_rate", 0.0),
@@ -106,6 +107,10 @@ class AntiAtroposEnv(
             sla_violations=obs_data.get("sla_violations", 0),
             invalid_action_count=obs_data.get("invalid_action_count", 0),
             vip_failure_count=obs_data.get("vip_failure_count", 0),
+            metric_timestamp=obs_data.get("metric_timestamp", 0.0),
+            data_freshness_ms=obs_data.get("data_freshness_ms", 0),
+            action_ack_status=obs_data.get("action_ack_status", "success"),
+            choke_level=obs_data.get("choke_level", 0.0),
             done=payload.get("done", False),
             reward=payload.get("reward", 0.0),
         )

control/__init__.py

@@ -0,0 +1,2 @@
+from .kubernetes_executor import KubernetesExecutor
+from .validation import ActionValidator

control/kubernetes_executor.py

@@ -0,0 +1,108 @@
+import os
+import json
+from typing import Optional
+
+class KubernetesExecutor:
+    """
+    Executes high-level SRE actions on a Kubernetes cluster.
+    Provides a safe layer between SREAgent and actual infrastructure.
+    """
+    def __init__(self, kubeconfig: Optional[str] = None):
+        # Use provided path or env var, defaulting to mock if neither is found
+        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()
+        
+    def execute(self, action_type: str, target: str, parameter: float) -> str:
+        """
+        Translates SRE actions to Kube requests (ScaleDeployment, PatchIngress, etc.)
+        """
+        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)}"
+
+    def _real_execution(self, action_type: str, target: str, parameter: float) -> str:
+        """Execute bounded actions on a Kubernetes cluster."""
+        if action_type == "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)
+
+        return f"Rejected: {action_type} 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"
+
+    def _scale_deployment(self, action_type: str, target: str, parameter: float) -> str:
+        deployment_name = self._resolve_deployment_name(target)
+        apps_v1 = self._get_apps_v1_api()
+
+        scale_obj = apps_v1.read_namespaced_deployment_scale(
+            name=deployment_name,
+            namespace=self.namespace,
+        )
+
+        current = int(scale_obj.spec.replicas or self.min_replicas)
+        delta = max(1, int(float(parameter) * self.scale_step))
+        if action_type == "SCALE_UP":
+            desired = min(self.max_replicas, current + delta)
+        else:
+            desired = max(self.min_replicas, current - delta)
+
+        if desired == current:
+            return (
+                f"Ack: {action_type} for {target} - replicas unchanged at {current} "
+                f"(bounds {self.min_replicas}-{self.max_replicas})"
+            )
+
+        apps_v1.patch_namespaced_deployment_scale(
+            name=deployment_name,
+            namespace=self.namespace,
+            body={"spec": {"replicas": desired}},
+        )
+
+        return f"Ack: {action_type} for {target} - deployment {deployment_name} scaled {current}->{desired}"
+
+    def _get_apps_v1_api(self):
+        if self._apps_v1_api is not None:
+            return self._apps_v1_api
+
+        from kubernetes import client, config
+
+        if self.kubeconfig and self.kubeconfig.lower() not in ("mock", ""):
+            config.load_kube_config(config_file=self.kubeconfig)
+        else:
+            config.load_incluster_config()
+
+        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 {}
+        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 {}
+
+    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}"

control/validation.py

@@ -0,0 +1,25 @@
+from typing import List, Optional
+from pydantic import BaseModel
+
+class ActionValidator:
+    """
+    Validates SRE actions to ensure they stay within safety boundaries.
+    Prevents destructive operations like 100% shedding on critical nodes.
+    """
+    def __init__(self, critical_nodes: Optional[List[str]] = None):
+        self.critical_nodes = critical_nodes or ["node-0", "node-1", "node-2"]
+
+    def validate(self, action_type: str, target: str, parameter: float, valid_targets: Optional[List[str]] = None) -> (bool, str):
+        """
+        Returns (is_valid, error_message).
+        """
+        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:
+            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."
+            
+        return True, "Success"

models.py

@@ -2,6 +2,11 @@ from enum import Enum
 from typing import Annotated, Literal, Optional
 from pydantic import BaseModel, Field
 
+class EnvironmentMode(str, Enum):
+    SIMULATED = "simulated"
+    HYBRID = "hybrid"
+    LIVE = "live"
+
 # ---------------------------------------------------------------------------
 # SRE Action Schema (Control Plane)
 # ---------------------------------------------------------------------------
@@ -90,6 +95,8 @@ class ClusterObservation(BaseModel):
     step: int
     max_steps: int
     
+    mode: EnvironmentMode = EnvironmentMode.SIMULATED
+    
     active_nodes: int = Field(ge=0, le=5)
     
     average_latency_ms: float = Field(
@@ -141,8 +148,15 @@ class ClusterObservation(BaseModel):
         description="Number of failed VIP nodes in the current observation.",
     )
 
+    # New fields for Prometheus/Kubernetes integration
+    metric_timestamp: float = 0.0
+    data_freshness_ms: int = 0
+    action_ack_status: str = "success"
+    choke_level: float = 0.0
+
     nodes: list[NodeObservation]
 
     # Episode interaction fields (handled by framework)
     done: bool = False
     reward: float = 0.0
+

pyproject.toml

@@ -26,6 +26,9 @@ dependencies = [
     # "gymnasium>=0.29.0",
     # "openspiel>=1.0.0",
     # "smolagents>=1.22.0,<2",
+    "kubernetes>=28.0.0",
+    "prometheus-api-client>=0.5.0",
+    "requests>=2.31.0",
 ]
 
 [project.optional-dependencies]
@@ -41,5 +44,5 @@ server = "AntiAtropos.server.app:main"
 
 [tool.setuptools]
 include-package-data = true
-packages = ["AntiAtropos", "AntiAtropos.server"]
-package-dir = { "AntiAtropos" = ".", "AntiAtropos.server" = "server" }
+packages = ["AntiAtropos", "AntiAtropos.server", "AntiAtropos.control", "AntiAtropos.telemetry"]
+package-dir = { "AntiAtropos" = ".", "AntiAtropos.server" = "server", "AntiAtropos.control" = "control", "AntiAtropos.telemetry" = "telemetry" }

server/AntiAtropos_environment.py

@@ -1,35 +1,21 @@
-# Copyright (c) Meta Platforms, Inc. and affiliates.
-# All rights reserved.
-#
-# This source code is licensed under the BSD-style license found in the
-# LICENSE file in the root directory of this source tree.
-
-"""
-AntiAtropos Environment — Server-Side Implementation.
-
-This module is the central orchestration layer. It:
-  1. Receives a typed SREAction from the OpenEnv framework.
-  2. Feeds it into the simulator (simulator.py) to advance time by one tick.
-  3. Queries stability.py to compute Lyapunov energy and the scalar reward.
-  4. Packages the resulting cluster state into a ClusterObservation and returns it.
-
-The environment implements a discrete-time fluid-queue model where stability 
-is governed by Lyapunov Drift-Plus-Penalty theory.
-"""
-
+import time
 from uuid import uuid4
 
 from openenv.core.env_server.interfaces import Environment
 from openenv.core.env_server.types import State
 
 try:
-    from ..models import SREAction, ClusterObservation, NodeObservation, NodeStatus
+    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 ..control import KubernetesExecutor, ActionValidator
 except ImportError:
-    from models import SREAction, ClusterObservation, NodeObservation, NodeStatus  # type: ignore[no-redef]
+    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 control import KubernetesExecutor, ActionValidator  # type: ignore[no-redef]
 
 
 # ---------------------------------------------------------------------------
@@ -64,22 +50,55 @@ class AntiAtroposEnvironment(Environment):
         """Initialise environment metadata and the simulation core."""
         self._state = State(episode_id=str(uuid4()), step_count=0)
         self._task_id: str = "task-1"
+        self._mode: EnvironmentMode = EnvironmentMode.SIMULATED
+        
+        # Core components
         self._sim: ClusterSimulator = ClusterSimulator(n_nodes=N_NODES, task_id="task-1")
+        self._telemetry = PrometheusClient()
+        self._executor = KubernetesExecutor()
+        self._validator = ActionValidator()
+        
         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_metric_time: float = 0.0
 
-    def reset(self, task_id: str = "task-1") -> ClusterObservation:
+    def reset(self, task_id: str = "task-1", mode: str = "simulated") -> ClusterObservation:
         """
-        Start a fresh episode with a specific task profile.
+        Start a fresh episode with a specific task profile and mode.
         """
         self._state = State(episode_id=str(uuid4()), step_count=0)
         self._task_id = task_id
+        try:
+            self._mode = EnvironmentMode(mode)
+        except ValueError:
+            self._mode = EnvironmentMode.SIMULATED
+            
         self._sla_violations = 0
+        self._action_ack_status = "success"
+        
+        # Only set baseline metric time for hybrid/live to prevent misleading freshness in SIM
+        if self._mode in [EnvironmentMode.HYBRID, EnvironmentMode.LIVE]:
+            self._last_metric_time = time.time()
+        else:
+            self._last_metric_time = 0.0
+
+        # Initialize core components based on mode
+        if self._mode != EnvironmentMode.SIMULATED:
+            # In Hybrid/Live mode, we might want to connect to real endpoints
+            # self._telemetry = PrometheusClient(url=os.getenv("PROMETHEUS_URL"))
+            pass
 
-        # Initialize the production simulator
         self._sim.reset(task_id=task_id)
+        
+        # If in hybrid mode, immediately pull a baseline
+        if self._mode in [EnvironmentMode.HYBRID, EnvironmentMode.LIVE]:
+            node_ids = [n["node_id"] for n in self._sim.state(for_agent=False)]
+            metrics = self._telemetry.fetch_latest_metrics(node_ids)
+            self._sim.reconcile_state(metrics)
+
         self._nodes_true = self._sim.state(for_agent=False)
         self._nodes_obs  = self._sim.state(for_agent=True)
         self._prev_lyapunov = compute_lyapunov(self._nodes_true)
@@ -91,26 +110,56 @@ class AntiAtroposEnvironment(Environment):
         Advance the simulation by one discrete time tick.
         """
         self._state.step_count += 1
-
-        # 1. Apply management action and advance physics
-        self._sim.apply_action(action)
+        
+        # 1. Action Validation & Execution
+        valid_targets = [n["node_id"] for n in self._nodes_true]
+        is_valid, error = self._validator.validate(
+            action.action_type, 
+            action.target_node_id, 
+            action.parameter,
+            valid_targets=valid_targets
+        )
+        
+        if not is_valid:
+            self._action_ack_status = f"Rejected: {error}"
+            # 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)
+            else:
+                self._action_ack_status = "success (simulated)"
+            
+            # Always update the physics engine to keep tracking expectations
+            self._sim.apply_action(action)
+
+        # 2. Advance Physics
         self._sim.tick()
         
-        # 2. Extract states (Ground Truth for reward; Observation for agent)
+        # 3. Telemetry Ingestion (Hybrid / Live)
+        if self._mode in [EnvironmentMode.HYBRID, EnvironmentMode.LIVE]:
+            node_ids = [n["node_id"] for n in self._nodes_true]
+            metrics = self._telemetry.fetch_latest_metrics(node_ids)
+            self._sim.reconcile_state(metrics)
+            self._last_metric_time = time.time()
+        
+        # 4. Extract states (Ground Truth for reward; Observation for agent)
         self._nodes_true = self._sim.state(for_agent=False)
         self._nodes_obs  = self._sim.state(for_agent=True)
 
-        # 3. SLA Check (must happen BEFORE reward so it is synchronized)
+        # 5. SLA Check
         avg_latency = self._avg_latency(self._nodes_true)
         error_rate  = self._error_rate(self._nodes_true)
         sla_violation_step = 1 if (avg_latency > 200.0 or error_rate > 0.05) else 0
         if sla_violation_step:
             self._sla_violations += 1
 
-        # 4. Compute Lyapunov stability metrics from Ground Truth
+        # 6. Compute Lyapunov stability metrics from Ground Truth
         current_lyapunov = compute_lyapunov(self._nodes_true)
         
-        # 5. Compute scalar reward using per-step SLA penalty for clean credit assignment
+        # 7. Compute scalar reward
         cost = self._compute_cost(self._nodes_true)
         reward = compute_reward(
             v_prev=self._prev_lyapunov,
@@ -124,13 +173,13 @@ class AntiAtroposEnvironment(Environment):
         
         self._prev_lyapunov = current_lyapunov
 
-        # 6. Termination check
+        # 8. Termination check
         done = (
             self._state.step_count >= MAX_STEPS
             or all(n["status"] == NodeStatus.FAILED for n in self._nodes_true)
         )
 
-        # 7. Package Observation (from OBSERVED state)
+        # 9. Package Observation
         obs = self._build_observation()
         obs.done   = done
         obs.reward = reward
@@ -150,7 +199,6 @@ class AntiAtroposEnvironment(Environment):
         for node in nodes_true:
             if node["status"] == NodeStatus.FAILED:
                 continue
-            # Both live and pending units are billed as infrastructure is provisioned.
             total_capacity_units += int(node.get("capacity_units", 0))
             total_capacity_units += int(node.get("pending_capacity_units", 0))
         return total_capacity_units * COST_PER_CAPACITY_UNIT_PER_HOUR
@@ -177,10 +225,6 @@ class AntiAtroposEnvironment(Environment):
         total_incoming = sum(float(n.get("incoming_request_rate", 0.0)) * float(n.get("importance_weight", 1.0)) for n in nodes)
         if total_incoming <= 0:
             return 0.0
-
-        # dropped_requests already includes both:
-        # - explicit shedding (SHED_LOAD)
-        # - traffic sent to FAILED nodes in simulator._update_queues
         total_drops = sum(float(n.get("dropped_requests", 0.0)) * float(n.get("importance_weight", 1.0)) for n in nodes)
         return min(1.0, total_drops / total_incoming)
 
@@ -206,12 +250,12 @@ class AntiAtroposEnvironment(Environment):
             for n in self._nodes_obs
         ]
 
-        # Aggregate metrics for the cluster-level dashboard
-        # CRITICAL: We use TRUE state for the objective metrics (grader-facing) 
-        # so that sensor dropout (-1.0 latency) doesn't fake a 'good' score.
+        freshness = int((time.time() - self._last_metric_time) * 1000) if self._last_metric_time > 0 else 0
+
         return ClusterObservation(
             cluster_id=self._state.episode_id,
             task_id=self._task_id,
+            mode=self._mode,
             active_nodes=sum(1 for n in self._nodes_true if n["status"] != NodeStatus.FAILED),
             average_latency_ms=min(1.0, max(0.0, self._avg_latency(self._nodes_true) / MAX_LATENCY_NORM)),
             error_rate=self._error_rate(self._nodes_true),
@@ -224,6 +268,12 @@ class AntiAtroposEnvironment(Environment):
             sla_violations=self._sla_violations,
             invalid_action_count=self._sim.invalid_action_count,
             vip_failure_count=self._vip_failure_count(self._nodes_true),
+            metric_timestamp=self._last_metric_time,
+            data_freshness_ms=freshness,
+            action_ack_status=self._action_ack_status,
+            choke_level=0.0,
             done=False,
             reward=0.0,
         )
+
+

server/Dockerfile

@@ -59,6 +59,11 @@ FROM ${BASE_IMAGE}
 
 WORKDIR /app
 
+# Ensure runtime has curl for health checks and diagnostics
+RUN apt-get update && \
+    apt-get install -y --no-install-recommends curl && \
+    rm -rf /var/lib/apt/lists/*
+
 # Copy the virtual environment from builder
 COPY --from=builder /app/env/.venv /app/.venv
 
@@ -71,6 +76,19 @@ ENV PATH="/app/.venv/bin:$PATH"
 # Set PYTHONPATH so imports work correctly
 ENV PYTHONPATH="/app/env:$PYTHONPATH"
 
+# Default integration config (safe local fallback)
+# Override these at runtime for real Hybrid/Live deployments.
+ENV ANTIATROPOS_ENV_MODE="simulated"
+ENV ANTIATROPOS_STRICT_REAL="false"
+ENV PROMETHEUS_URL="mock"
+ENV KUBECONFIG="mock"
+ENV ANTIATROPOS_K8S_NAMESPACE="default"
+ENV ANTIATROPOS_DEPLOYMENT_PREFIX=""
+ENV ANTIATROPOS_MIN_REPLICAS="1"
+ENV ANTIATROPOS_MAX_REPLICAS="20"
+ENV ANTIATROPOS_SCALE_STEP="3"
+ENV ANTIATROPOS_NODE_DEPLOYMENT_MAP="{}"
+
 # Health check
 HEALTHCHECK --interval=30s --timeout=3s --start-period=5s --retries=3 \
     CMD curl -f http://localhost:8000/health || exit 1

server/requirements.txt

@@ -1,6 +1,6 @@
 openenv[core]>=0.2.0
 fastapi>=0.115.0
 uvicorn>=0.24.0
-
-
-
+kubernetes>=28.0.0
+prometheus-api-client>=0.5.0
+requests>=2.31.0

simulator.py

@@ -433,3 +433,53 @@ class ClusterSimulator:
                 n.status = NodeStatus.DEGRADED
             elif n.status == NodeStatus.DEGRADED and n.queue_depth < (OVERLOAD_THRESHOLD / 2):
                 n.status = NodeStatus.HEALTHY
+
+    def reconcile_state(self, telemetry_map: dict) -> None:
+        """
+        Reconcile internal simulator state with external telemetry signals.
+        Used in 'hybrid' or 'live' modes to align the physics engine with reality.
+        
+        telemetry_map: node_id -> TelemetryRecord (or dict)
+        """
+        for node in self._nodes:
+            if node.node_id in telemetry_map:
+                record = telemetry_map[node.node_id]
+                # If record is an object (TelemetryRecord), access fields; otherwise treat as dict
+                if hasattr(record, "queue_depth"):
+                    q_ext = float(record.queue_depth)
+                    r_ext = float(record.request_rate)
+                    c_ext = float(record.cpu_utilization)
+                    e_ext = float(record.error_rate)
+                    l_ext = float(record.latency_ms)
+                else:
+                    q_ext = float(record.get("queue_depth", node.queue_depth))
+                    r_ext = float(record.get("request_rate", node.incoming_request_rate))
+                    c_ext = float(record.get("cpu_utilization", node.cpu_utilization))
+                    e_ext = float(record.get("error_rate", 0.0))
+                    l_ext = float(record.get("latency_ms", node.latency_ms))
+
+                # Smoothly blend the external state into physics to prevent step jumps
+                # trust: 0.7 (reality) / 0.3 (simulation prediction)
+                node.queue_depth = (node.queue_depth * 0.3) + (q_ext * 0.7)
+                node.incoming_request_rate = (node.incoming_request_rate * 0.3) + (r_ext * 0.7)
+                node.cpu_utilization = (node.cpu_utilization * 0.3) + (c_ext * 0.7)
+                node.latency_ms = (node.latency_ms * 0.3) + (l_ext * 0.7)
+
+                # Status reconciliation: Fail the node if error rate is high
+                if e_ext > 0.5:
+                    node.status = NodeStatus.FAILED
+                elif e_ext > 0.1 and node.status == NodeStatus.HEALTHY:
+                    node.status = NodeStatus.DEGRADED
+                elif e_ext <= 0.05 and node.status == NodeStatus.DEGRADED:
+                    # Allow recovery if telemetry says it's clean (physics will still check queue)
+                    node.status = NodeStatus.HEALTHY
+
+        # Crucial: re-derive metrics so latency/cpu are consistent (except for the blended values)
+        # Note: We blend latency/cpu above, but _update_derived_metrics might overwrite them.
+        # So we update them after blending if we want physics to win, or before if telemetry wins.
+        # Usually, for SRE dashboard, we want the blended 'reality'.
+        # However, _update_derived_metrics is used to compute 'current' state in pure sim.
+        # We'll skip it if we just reconciled to keep the blended values, OR refine it.
+        # For now, let's just make sure statuses are updated based on new queue depths.
+        self._update_statuses()
+

telemetry/__init__.py

@@ -0,0 +1,2 @@
+from .prometheus_client import PrometheusClient, TelemetryRecord
+from .mapping import MetricMapper

telemetry/mapping.py

@@ -0,0 +1,27 @@
+from typing import Dict, List, Any
+import os
+
+class MetricMapper:
+    """
+    Utility for mapping Prometheus label sets into internal node IDs.
+    In environments with many pods, we need to decide which pods to 
+    aggregate for a given node_id.
+    """
+    def __init__(self, mapping_strategy: str = "sum"):
+        self.strategy = mapping_strategy
+        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
+        return {
+            "web-node-1": "node-0",
+            "api-node-2": "node-1",
+            "db-node-3": "node-2"
+        }
+
+    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 {}

telemetry/prometheus_client.py

@@ -0,0 +1,134 @@
+import os
+import random
+from typing import Dict, List, Optional
+import requests
+from pydantic import BaseModel
+
+class TelemetryRecord(BaseModel):
+    node_id: str
+    latency_ms: float
+    request_rate: float
+    error_rate: float
+    cpu_utilization: float
+    queue_depth: float
+
+class PrometheusClient:
+    """
+    Adapter to fetch and normalize metrics from Prometheus.
+    Supports a mock mode for local development without a live cluster.
+    """
+    def __init__(self, prometheus_url: Optional[str] = None):
+        # Use provided URL or env var, defaulting to mock if neither is found
+        self.url = prometheus_url or os.getenv("PROMETHEUS_URL")
+        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.request_rate_query = os.getenv(
+            "ANTIATROPOS_PROM_QUERY_REQUEST_RATE",
+            'sum(rate(http_requests_total{node_id="{node_id}"}[1m]))'
+        )
+        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'
+        )
+        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)'
+        )
+        self.cpu_query = os.getenv(
+            "ANTIATROPOS_PROM_QUERY_CPU",
+            'avg(rate(container_cpu_usage_seconds_total{pod=~".*{node_id}.*"}[1m]))'
+        )
+        self.queue_depth_query = os.getenv(
+            "ANTIATROPOS_PROM_QUERY_QUEUE_DEPTH",
+            'sum(queue_depth{node_id="{node_id}"})'
+        )
+        
+    def fetch_latest_metrics(self, node_ids: List[str]) -> Dict[str, TelemetryRecord]:
+        """
+        Query Prometheus for the latest metrics for the given nodes.
+        Returns a mapping from node_id to TelemetryRecord.
+        """
+        if self.is_mock:
+            return self._generate_mock_metrics(node_ids)
+        
+        # Real implementation using queries
+        try:
+            return self._fetch_real_metrics(node_ids)
+        except Exception:
+            if self.strict_real:
+                raise
+            return self._generate_mock_metrics(node_ids)
+
+    def _fetch_real_metrics(self, node_ids: List[str]) -> Dict[str, TelemetryRecord]:
+        """Fetches node telemetry from Prometheus instant queries."""
+        metrics: Dict[str, TelemetryRecord] = {}
+        saw_any_real_signal = False
+
+        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))
+
+            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(
+                node_id=node_id,
+                latency_ms=float(lat_ms if lat_ms is not None else 20.0),
+                request_rate=float(req_rate if req_rate is not None else 0.0),
+                error_rate=max(0.0, min(1.0, float(err_rate if err_rate is not None else 0.0))),
+                cpu_utilization=max(0.0, min(1.0, float(cpu if cpu is not None else 0.0))),
+                queue_depth=max(0.0, float(q_depth if q_depth is not None else 0.0)),
+            )
+
+        if self.strict_real and not saw_any_real_signal:
+            raise RuntimeError("Prometheus returned no usable real telemetry for requested node IDs.")
+
+        return metrics
+
+    def _query_scalar(self, promql: str) -> Optional[float]:
+        """Runs a scalar/vector Prometheus instant query and returns the first value."""
+        if not self.url:
+            return None
+
+        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 None
+
+        result = payload.get("data", {}).get("result", [])
+        if not result:
+            return None
+
+        value = result[0].get("value")
+        if not value or len(value) < 2:
+            return None
+
+        try:
+            return float(value[1])
+        except (TypeError, ValueError):
+            return None
+
+    def _generate_mock_metrics(self, node_ids: List[str]) -> Dict[str, TelemetryRecord]:
+        """Generates realistic-looking mock telemetry."""
+        metrics = {}
+        for nid in node_ids:
+            metrics[nid] = TelemetryRecord(
+                node_id=nid,
+                latency_ms=random.uniform(20.0, 150.0),
+                request_rate=random.uniform(10.0, 50.0),
+                error_rate=random.uniform(0.0, 0.05),
+                cpu_utilization=random.uniform(0.1, 0.8),
+                queue_depth=random.uniform(0.0, 50.0)
+            )
+        return metrics