simulator.py

# 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 Core Simulation Physics.

A discrete-time fluid-queue model simulating a 5-node microservice cluster.
Each node has stateful queues, capacities, and failure probabilities. 
Dynamic traffic is injected per tick, and management actions shift capacity
and routing parameters.
"""

import random
from dataclasses import dataclass, field
from enum import Enum
from typing import Optional

# ---------------------------------------------------------------------------
# Simulator Constants (World Physics)
# ---------------------------------------------------------------------------

DEFAULT_CAPACITY:     float = 3.0     # Baseline service rate (μ) per node
MAX_CAPACITY:         float = 999.0   # High sentinel — cost function (not an artificial cap) limits scaling.  Real K8s has no per-node pod ceiling; RL must learn cost discipline.
MAX_SCALING_STEP:     int   = 3       # Largest allowed SCALE_UP param (was 5)
BOOT_DELAY_TICKS:     int   = 5       # Time it takes to bring up infrastructure
BASE_LATENCY_MS:      float = 20.0    # Minimum processing time
OVERLOAD_THRESHOLD:   int   = 80      # Request count where node begins to "fail" (DEGRADED)
LATENCY_STEEPNESS:    float = 2.0     # Increased to ensure SLA violations before death
FATAL_FAIL_THRESHOLD: int   = 200     # Hard cap on queue depth (catastrophic failure boundary)
CASCADE_WINDOW_TICKS: int = 3     # Ticks after a failure to check for cascade effects
CASCADE_QUEUE_MULTIPLIER: float = 1.2  # Queue must exceed FATAL_FAIL_THRESHOLD * this to cascade
NODE_RECOVERY_TICKS: int   = 20      # Ticks before a FAILED node auto-recovers
BACKPRESSURE_THRESHOLD: float = 60.0   # Queue depth that triggers backpressure
BACKPRESSURE_MAX_FACTOR: float = 0.4   # Maximum service rate reduction (40%)

SENSOR_DROPOUT_PROB:  float = 0.05    # P(node.queue, latency reports 0 or -1.0)
NODE_FAILURE_PROB:    float = 0.00    # P(node fails naturally) — largely driven by task profile

# Cost model -- two-tier: cheap for capacity needed to handle current
# traffic, expensive for idling excess (punishes over-provisioning
# without discouraging necessary scaling).
COST_PER_CAPACITY_UNIT_PER_HOUR: float = 0.05     # Base rate for needed capacity
OVERPROVISION_COST_PER_UNIT:     float = 1.0      # Penalty rate for excess (20x base)

# Task Profiles (Domain Randomization)
# Task 1: Start at 92-99% of ingress capacity (randomised in _randomize_domain).
# DAG ingress capacity = 2 ingress nodes * DEFAULT_CAPACITY * 15 = 90 req/tick.
# lambda_init ≈ 83-89 so each ingress node sees ~41-44 req/tick (just under 45 capacity).
T1_INITIAL_LAMBDA: float = 86.0   # midpoint of [82.8, 89.1]; overridden by _randomize_domain
T1_RAMP_SLOPE:     float = 0.5    # +0.5 req/tick globally per tick
# Task 2: lambda at 100-110% of ingress capacity — guarantees immediate ingress overload.
T2_INITIAL_LAMBDA: float = 95.0   # midpoint of [90, 99]; overridden by _randomize_domain
T2_FAIL_TICK:      int   = 20
T3_INITIAL_LAMBDA: float = 60.0

# Task 3 surge parameters — base window, jitter applied per episode
T3_SURGE_CYCLE:        int   = 60   # Cycle length (ticks)
T3_SURGE_BASE_START:   int   = 30   # Nominal start of surge within cycle
T3_SURGE_BASE_END:     int   = 40   # Nominal end of surge within cycle
T3_SURGE_JITTER:       int   = 10   # ±jitter applied to start/end each episode
T3_SURGE_MAGNITUDE:    float = 60.0  # Extra req/tick added to node-1 and node-2 (4 SCALE_UPs at delta=1 covers it: 5x15=75 = 15 DAG + 60 surge)

# Only the VIP payment gateway (node-0) is hard-protected from shedding --
# this is a safety guardrail, not a learned behavior.  The agent must learn
# from the reward signal (error_rate -> sla_penalty at gamma=6.0) that
# shedding surge nodes (node-1/2) is rarely optimal.
CRITICAL_NODES: list[str] = ["node-0"]

# VIP / business-critical node weights.
# node-0 is the payment portal, so its queue growth or failure matters more.
# Reduced from 4.0 → 2.0 to prevent reward gradient from creating
# a local optimum where the agent only scales node-0.
# At 2×, node-0 is still prioritized but other nodes remain viable targets.
VIP_NODE_WEIGHTS: dict[str, float] = {
    "node-0": 2.0,
}

# ---------------------------------------------------------------------------
# Graph Topology (DAG — fixed 5-node cluster architecture)
# ---------------------------------------------------------------------------

# Directed edges: parent -> list of direct children.
# node-0 (payments/VIP) is the primary ingress; node-4 (auth) is independent.
CLUSTER_TOPOLOGY: dict[str, list[str]] = {
    "node-0": ["node-1", "node-2"],
    "node-1": [],
    "node-2": ["node-3"],
    "node-3": [],
    "node-4": [],
}

# Nodes that receive raw external traffic directly.
EXTERNAL_TRAFFIC_NODES: set[str] = {"node-0", "node-4"}

# 50/50 external λ split between the two ingress nodes.
# node-0 (payments/VIP) and node-4 (auth) each receive half of total_lambda.
# total_lambda is the cluster-wide external arrival rate (req/tick).
# Each ingress node therefore sees total_lambda * 0.5 req/tick at its input.
EXTERNAL_LAMBDA_FRACTION: float = 0.5

# Default upstream-to-downstream routing weights (parent → child splits).
# These represent the baseline traffic split before agent rerouting.
DEFAULT_ROUTING_SPLIT: dict[str, dict[str, float]] = {
    "node-0": {"node-1": 0.5, "node-2": 0.5},
    "node-2": {"node-3": 1.0},
}

# Pre-computed topological order (Kahn's BFS on CLUSTER_TOPOLOGY).
# Ensures parents are always processed before their children in _inject_traffic().
# Order: node-0, node-4 (roots) → node-1, node-2 (node-0 children) → node-3 (node-2 child).
_TOPOLOGICAL_ORDER: tuple[str, ...] = ("node-0", "node-4", "node-1", "node-2", "node-3")


class NodeStatus(str, Enum):
    HEALTHY  = "HEALTHY"
    DEGRADED = "DEGRADED"
    FAILED   = "FAILED"


@dataclass
class NodeState:
    node_id: str
    status: NodeStatus = NodeStatus.HEALTHY
    is_vip: bool = False
    
    # Physics parameters
    capacity: float = DEFAULT_CAPACITY
    importance_weight: float = 1.0
    queue_depth: float = 0.0
    latency_ms: float = BASE_LATENCY_MS
    incoming_request_rate: float = 0.0
    cpu_utilization: float = 0.0
    
    # Per-tick accounting (reset each tick)
    dropped_requests: float = 0.0
    shed_fraction: float = 0.0       # Fraction of incoming traffic to drop this tick
    pending_capacity_queue: list[int] = field(default_factory=list)
    recovery_timer: int = 0          # Countdown to auto-recovery from FAILED status
    is_scripted_failure: bool = False  # True if failed due to task scripting (no auto-recovery)
    outflow_rate: float = 0.0         # Requests/tick actually dispatched downstream (DAG edge signal)

    # Derived (recomputed whenever capacity or status changes)
    @property
    def service_rate(self) -> float:
        """The total service capacity (μ) in requests per tick."""
        if self.status == NodeStatus.FAILED:
            return 0.0
        return self.capacity * 15.0  # Base unit = 15 req/tick processing power

    def to_dict(self) -> dict:
        return {
            "node_id": self.node_id,
            "status": self.status,
            "is_vip": self.is_vip,
            "capacity": self.capacity,
            "importance_weight": self.importance_weight,
            "queue_depth": int(self.queue_depth),
            "latency_ms": round(self.latency_ms, 2),
            "incoming_request_rate": round(self.incoming_request_rate, 2),
            "cpu_utilization": round(min(1.0, self.cpu_utilization), 4),
            "dropped_requests": int(self.dropped_requests),
            "shed_fraction": round(self.shed_fraction, 4),
            "capacity_units": int(self.capacity),
            "pending_capacity_units": int(len(self.pending_capacity_queue)),
            "recovery_timer": self.recovery_timer,
            "is_scripted_failure": self.is_scripted_failure,
            "outflow_rate": round(self.outflow_rate, 2),
        }


class ClusterSimulator:
    """
    Multi-node fluid queue simulator.
    
    Operates in discrete ticks. 
    1. Action: Control plane actions (Scaling/Routing/Shedding) are applied.
    2. Tick: Physics engine updates queues based on λ (incoming) and μ (service rate).
    3. Failure Logic: Queue overflows trigger status degradation/node death.
    """

    def __init__(self, n_nodes: int = 5, task_id: str = "task-1", seed: Optional[int] = None):
        self._n_nodes = n_nodes
        self._task_id = task_id
        # Default to non-deterministic RNG seeding so fresh simulator instances
        # do not replay identical domain-randomization sequences.
        # Pass an explicit seed for reproducible experiments.
        self._seed: Optional[int] = seed
        self._rng = random.Random(seed)
        self._tick_count: int = 0
        self._failed_node_id: Optional[str] = None
        self._t1_ramp_slope: float = T1_RAMP_SLOPE
        self._t1_init_lambda: float = T1_INITIAL_LAMBDA
        self._t2_fail_tick: int = T2_FAIL_TICK
        self._t2_init_lambda: float = T2_INITIAL_LAMBDA
        # Task-3 surge window — randomised per episode in _randomize_domain()
        self._t3_surge_start: int = T3_SURGE_BASE_START
        self._t3_surge_end:   int = T3_SURGE_BASE_END
        # Per-node reroute weights for REROUTE_TRAFFIC (node_id → fraction)
        self._reroute_weights: dict[str, float] = {}
        self._cascade_tick: int = 0  # Tick counter for cascade detection window
        self._cascade_triggered: bool = False  # Set True when a NEW overload failure occurs
        self._nodes: list[NodeState] = []
        self.invalid_action_count: int = 0
        self._randomize_domain()
        self._reset_nodes()

    def _randomize_domain(self) -> None:
        """Apply domain randomization for RL robustness across tasks."""
        self._t1_ramp_slope = self._rng.uniform(0.8, 2.0)
        # DAG calibration: total_lambda is split across 2 ingress nodes (node-0, node-4).
        # Each ingress node's capacity is DEFAULT_CAPACITY * 15 req/tick.
        # Ingress cluster capacity = len(EXTERNAL_TRAFFIC_NODES) * DEFAULT_CAPACITY * 15 = 90.
        # Task 1: start between 92-99% of ingress capacity so the ingress nodes are
        # near saturation immediately, producing rich early reward signal.
        n_ingress = len(EXTERNAL_TRAFFIC_NODES)  # 2
        ingress_mu_total = n_ingress * DEFAULT_CAPACITY * 15.0  # 90 req/tick
        self._t1_init_lambda = self._rng.uniform(
            ingress_mu_total * 0.92, ingress_mu_total * 0.99
        )
        self._t2_fail_tick = self._rng.randint(10, 40)
        # Task 2: guarantee immediate ingress overload (slightly above ingress saturation).
        # Each ingress node sees total_lambda/2; target ~102% of individual ingress capacity.
        self._t2_init_lambda = self._rng.uniform(
            ingress_mu_total * 1.00, ingress_mu_total * 1.10
        )
        # Task 3: jitter the surge window so the LLM can't memorise it.
        jitter = self._rng.randint(-T3_SURGE_JITTER, T3_SURGE_JITTER)
        self._t3_surge_start = T3_SURGE_BASE_START + jitter
        self._t3_surge_end   = T3_SURGE_BASE_END   + jitter
        # Reset reroute weights to uniform on each new episode
        self._reroute_weights = {}

    def _reset_nodes(self) -> None:
        self._nodes = [
            NodeState(
                node_id=f"node-{i}",
                is_vip=f"node-{i}" in VIP_NODE_WEIGHTS,
                importance_weight=VIP_NODE_WEIGHTS.get(f"node-{i}", 1.0),
                is_scripted_failure=False,
            )
            for i in range(self._n_nodes)
        ]

    def reset(self, task_id: str = "task-1", seed: Optional[int] = None) -> None:
        """Restart the simulator for a fresh episode."""
        if seed is not None:
            self._seed = seed
            # Reinitialize RNG so episode generation is reproducible for a given seed.
            self._rng = random.Random(seed)
        self._task_id = task_id
        self._tick_count = 0
        self._failed_node_id = None
        self._reroute_weights = {}
        self._cascade_tick = 0
        self._cascade_triggered = False
        self.invalid_action_count = 0
        self._randomize_domain()
        self._reset_nodes()

    def state(self, for_agent: bool = True) -> list[dict]:
        """Return current per-node state as a list of plain dicts."""
        state_list = []
        for n in self._nodes:
            d = n.to_dict()
            if for_agent and self._rng.random() < SENSOR_DROPOUT_PROB:
                d["queue_depth"] = 0
                d["latency_ms"] = -1.0
            state_list.append(d)
        return state_list

    def apply_action(self, action_model) -> bool:
        """Update simulator world-state parameters based on agent command.

        Returns:
            True if the action had a measurable effect on state, False if it
            was silently discarded (e.g. SCALE_UP at capacity cap).
        """
        at = action_model.action_type if hasattr(action_model, "action_type") else action_model["action_type"]
        node_id = action_model.target_node_id if hasattr(action_model, "target_node_id") else action_model["target_node_id"]
        param = action_model.parameter if hasattr(action_model, "parameter") else action_model["parameter"]

        # 1. NO_OP always succeeds regardless of target node
        if at == "NO_OP":
            return True

        # 2. Target node lookup (required for all other actions)
        target = next((n for n in self._nodes if n.node_id == node_id), None)
        if not target:
            return False

        # 3. Command implementation
        if at == "SCALE_UP":
            delta = max(1, int(param * MAX_SCALING_STEP))
            for _ in range(delta):
                target.pending_capacity_queue.append(BOOT_DELAY_TICKS)
            
            # If scaling up, we forcefully 'repair' DEGRADED status (SRE manual intervention)
            was_degraded = target.status == NodeStatus.DEGRADED
            if was_degraded:
                target.status = NodeStatus.HEALTHY

            return True  # Always has effect — adds pending capacity, costs money

        elif at == "SCALE_DOWN":
            delta = max(1, int(param * MAX_SCALING_STEP))
            # First cancel any pending capacity (cancel boot queue entries)
            # This is like canceling a VM launch — it hasn't served traffic yet.
            cancelled = 0
            while cancelled < delta and target.pending_capacity_queue:
                target.pending_capacity_queue.pop()  # Remove newest pending first
                cancelled += 1
            # If still need to remove more, reduce active capacity
            remaining = delta - cancelled
            old_capacity = target.capacity
            if remaining > 0:
                target.capacity = max(1, target.capacity - remaining)
            # Had effect if we cancelled pending or reduced active capacity
            return cancelled > 0 or target.capacity != old_capacity

        elif at == "REROUTE_TRAFFIC":
            # Physically offload traffic FROM the target node by proportion `param`.
            # `param` ∈ [0, 1]: fraction of target node's share to redistribute.
            # The shed amount is split evenly across healthy peer nodes.
            frac = min(1.0, max(0.0, float(param)))
            # Record a reroute weight for the target so _inject_traffic() can
            # reduce its lambda and bump peers accordingly.
            self._reroute_weights[node_id] = frac
            return True  # Always has effect (sets a routing weight)

        elif at == "SHED_LOAD":
            # Rule: Cannot shed critical nodes (database/control plane).
            # This forces the agent to handle Task-3 surge via Scaling/Rerouting.
            if node_id in CRITICAL_NODES:
                self.invalid_action_count += 1
                return False  # Action rejected on critical node

            frac = min(1.0, param)
            target.shed_fraction = frac
            return True  # Had effect (set shed fraction)
            # Note: physically applied in _update_queues() to incoming traffic

        # NO_OP or unrecognised action types — always intentional, count as having effect
        return True

    def tick(self) -> None:
        """
        Advance simulated time by one step.
        Evaluates physics: Capacity growth, Traffic injection, Queue processing.
        """
        self._tick_count += 1
        self._update_capacity()
        self._inject_traffic()
        # Save original capacities; backpressure temporarily reduces service_rate
        # for this tick only. Restore after _update_queues so the reduction does
        # not compound across ticks and permanently cripple parent nodes.
        saved_capacities = {n.node_id: n.capacity for n in self._nodes}
        self._apply_backpressure()
        # Reset per-tick shed counters before physics update
        for node in self._nodes:
            node.dropped_requests = 0.0
        self._update_queues()
        # Restore capacities so the next tick starts from the true provisioned level
        for n in self._nodes:
            n.capacity = saved_capacities.get(n.node_id, n.capacity)
        self._update_derived_metrics()
        self._update_statuses()
        self._cascade_failures()
        self._process_recovery()
        # Decay shed fractions gradually
        # *= 0.5 retains 50% per tick (fast decay).
        for node in self._nodes:
            node.shed_fraction *= 0.5
            if node.shed_fraction < 0.05:
                node.shed_fraction = 0.0

    def _update_capacity(self) -> None:
        """Process pending capacity from SCALE_UP actions"""
        for node in self._nodes:
            for i in range(len(node.pending_capacity_queue)):
                node.pending_capacity_queue[i] -= 1
            
            # Move ready capacity to live
            ready = sum(1 for delay in node.pending_capacity_queue if delay <= 0)
            node.capacity += ready
            node.pending_capacity_queue = [delay for delay in node.pending_capacity_queue if delay > 0]

    def _inject_traffic(self) -> None:
        """
        Distribute traffic through the cluster DAG in three phases.

        Phase 1 — Task lambda + scripted events:
            Compute total external λ for this tick and apply any task-specific
            mutations (node failure scripting, surge flags).  No early returns;
            all branches fall through to the shared DAG in Phase 2.

        Phase 2 — Topological DAG distribution:
            Traverse _TOPOLOGICAL_ORDER (roots first).  Each parent's
            processed outflow (min(incoming, service_rate)) is split across
            its children via DEFAULT_ROUTING_SPLIT.  A FAILED node has
            service_rate=0, so outflow=0 and its children are naturally
            starved — this is the causal failure chain the RL agent must
            learn to route around.

        Phase 3 — Reroute weight correction:
            Apply REROUTE_TRAFFIC weight adjustments post-DAG, then decay
            weights.  Keeps reroute semantics identical to pre-DAG behaviour.
        """
        # -------------------------------------------------------------------
        # Phase 1: task-specific lambda + scripted events (no early returns)
        # -------------------------------------------------------------------
        total_lambda: float = 0.0
        # direct_injections: extra traffic added directly to a node ON TOP OF
        # the DAG distribution.  Used for Task-3 surge bursts that model a
        # side-channel load source (e.g. bulk import hitting checkout/catalog
        # directly), while the base λ still travels through node-0 as ingress.
        direct_injections: dict[str, float] = {}

        if self._task_id == "task-1":
            # Linear ramp — starts near cluster capacity
            total_lambda = self._t1_init_lambda + (self._t1_ramp_slope * self._tick_count)

        elif self._task_id == "task-2":
            total_lambda = self._t2_init_lambda
            # Scripted node failure fires at the configured tick
            if self._tick_count >= self._t2_fail_tick and not self._failed_node_id:
                self._failed_node_id = self._rng.choice(
                    [n.node_id for n in self._nodes if n.node_id != "node-0"]
                )
                target = next((n for n in self._nodes if n.node_id == self._failed_node_id), None)
                if target:
                    target.is_scripted_failure = True
            # No early return: DAG distributes traffic to the failed node normally.
            # The dead node's service_rate=0 means outflow=0, so its children are
            # starved. _update_queues() converts all its incoming traffic to
            # dropped_requests.  The agent must issue REROUTE_TRAFFIC to shift
            # the parent's split away from the dead child.

        elif self._task_id == "task-3":
            total_lambda = T3_INITIAL_LAMBDA
            phase = self._tick_count % T3_SURGE_CYCLE
            if self._t3_surge_start <= phase <= self._t3_surge_end:
                # Surge is modelled as a direct external burst arriving at the
                # checkout (node-1) and catalog (node-2) services from a side
                # channel that bypasses the payment gateway ingress.
                # Base λ still routes through the DAG; the surge is overlaid so
                # CRITICAL_NODE protections (no SHED_LOAD on node-1/2) still apply.
                for nid in ["node-1", "node-2"]:
                    direct_injections[nid] = T3_SURGE_MAGNITUDE

        # -------------------------------------------------------------------
        # Phase 2: DAG topological distribution
        # -------------------------------------------------------------------
        node_incoming: dict[str, float] = {n.node_id: 0.0 for n in self._nodes}
        node_map: dict[str, "NodeState"] = {n.node_id: n for n in self._nodes}

        # Seed ingress nodes with their share of external λ
        node_incoming["node-0"] = total_lambda * EXTERNAL_LAMBDA_FRACTION
        node_incoming["node-4"] = total_lambda * (1.0 - EXTERNAL_LAMBDA_FRACTION)

        # Overlay task-specific direct injections (Task-3 surge)
        for nid, extra in direct_injections.items():
            node_incoming[nid] = node_incoming.get(nid, 0.0) + extra

        # Propagate outflow through the graph in topological order
        for parent_id in _TOPOLOGICAL_ORDER:
            parent = node_map.get(parent_id)
            if parent is None:
                continue
            parent.incoming_request_rate = node_incoming[parent_id]
            # Outflow = requests the parent actually forwards downstream.
            # FAILED nodes have service_rate=0 → outflow=0 → children starved.
            outflow = min(parent.incoming_request_rate, parent.service_rate)
            parent.outflow_rate = outflow
            for child_id, split in DEFAULT_ROUTING_SPLIT.get(parent_id, {}).items():
                node_incoming[child_id] = node_incoming.get(child_id, 0.0) + outflow * split

        # -------------------------------------------------------------------
        # Phase 3: REROUTE_TRAFFIC weight corrections (post-DAG)
        # -------------------------------------------------------------------
        self._apply_reroute_weights()

        # Recalculate outflow after reroute so the agent sees accurate
        # per-node dispatch rates.  Without this, a node whose incoming was
        # halved by reroute would still report its pre-reroute outflow.
        for n in self._nodes:
            n.outflow_rate = min(n.incoming_request_rate, n.service_rate)

    def _apply_reroute_weights(self) -> None:
        """
        Apply REROUTE_TRAFFIC adjustments.

        For each node with a reroute weight w, reduce its incoming_request_rate
        by (w × its current rate) and distribute that offloaded load equally
        across healthy peer nodes.

        Important: rerouting must still work when the target is FAILED (Task 2).
        We therefore allow offload FROM failed targets. Only absorber nodes are
        constrained to non-failed nodes.

        Decays each reroute weight by 50% per tick so the effect must be
        re-issued to be maintained (forces the agent to act).
        """
        if not self._reroute_weights:
            return

        total_overflow: float = 0.0
        rerouted_ids = set(self._reroute_weights.keys())

        for n in self._nodes:
            w = self._reroute_weights.get(n.node_id, 0.0)
            if w <= 0.0:
                continue

            # Offload from the target regardless of node status.
            offload = n.incoming_request_rate * w
            if offload <= 0.0:
                continue
            n.incoming_request_rate -= offload
            total_overflow += offload

        # Spread overflow across all non-rerouted healthy nodes
        if total_overflow > 0:
            absorbers = [
                n for n in self._nodes
                if n.node_id not in rerouted_ids
                and n.status != NodeStatus.FAILED
            ]
            # If every healthy node is rerouted this tick, fall back to all
            # healthy non-rerouted nodes first, then finally all healthy nodes.
            # CRITICAL: A FAILED node should NEVER be an absorber.
            if not absorbers:
                absorbers = [n for n in self._nodes if n.status != NodeStatus.FAILED and n.node_id not in rerouted_ids]
            if not absorbers:
                absorbers = [n for n in self._nodes if n.status != NodeStatus.FAILED]
            
            if absorbers:
                share = total_overflow / len(absorbers)
                for n in absorbers:
                    n.incoming_request_rate += share

        # Decay weights — agent must keep re-issuing to maintain effect
        # *= 0.5 retains 50% per tick (fast decay).
        for nid in list(self._reroute_weights.keys()):
            self._reroute_weights[nid] *= 0.5
            if self._reroute_weights[nid] < 0.05:
                del self._reroute_weights[nid]

    def _apply_backpressure(self) -> None:
        """Reduce parent service rate when children are overloaded."""
        for parent_id, children in CLUSTER_TOPOLOGY.items():
            parent = next((n for n in self._nodes if n.node_id == parent_id), None)
            if not children or not parent or parent.status == NodeStatus.FAILED:
                continue
            
            # Compute pressure from overloaded children
            total_pressure = 0.0
            for child_id in children:
                child = next((n for n in self._nodes if n.node_id == child_id), None)
                if child:
                    excess = max(0.0, child.queue_depth - BACKPRESSURE_THRESHOLD)
                    total_pressure += excess / FATAL_FAIL_THRESHOLD  # normalise to [0, 1]
            
            # Reduce parent's effective capacity proportionally
            pressure_factor = min(BACKPRESSURE_MAX_FACTOR, total_pressure * 0.6)
            parent.capacity = max(1.0, parent.capacity * (1.0 - pressure_factor))

    def _update_queues(self) -> None:
        """
        Fluid-queue update for all nodes.

            Q_i(t+1) = max(Q_i(t) + λ_eff_i(t) − μ_i(t), 0)

        where λ_eff = λ_incoming * (1 - shed_fraction).
        """
        for node in self._nodes:
            if node.status == NodeStatus.FAILED:
                node.queue_depth = 0.0
                node.dropped_requests = node.incoming_request_rate
                continue

            # Apply shedding to incoming traffic
            lambda_eff = node.incoming_request_rate * (1.0 - node.shed_fraction)
            node.dropped_requests = node.incoming_request_rate - lambda_eff

            excess = lambda_eff - node.service_rate
            node.queue_depth = max(0.0, node.queue_depth + excess)

    def _update_derived_metrics(self) -> None:
        """Compute CPU and Latency based on current queue and status."""
        for n in self._nodes:
            if n.status == NodeStatus.FAILED:
                n.cpu_utilization = 0.0
                n.latency_ms = float("inf")
                continue

            # Utilization = Ratio of λ to μ
            service_rate = n.service_rate
            n.cpu_utilization = n.incoming_request_rate / service_rate if service_rate > 0 else 1.0

            # Latency: Hybrid M/M/1 + backlog term
            # M/M/1 gives exponential blow-up as utilization->1 (the "hockey stick")
            # Backlog term ensures queue_depth still contributes signal even when
            # utilization is capped at 0.99, preventing the flattening problem.
            utilization = min(0.99, n.cpu_utilization)  # cap to prevent infinity
            mm1_latency = BASE_LATENCY_MS / (1.0 - utilization)
            backlog_latency = n.queue_depth * LATENCY_STEEPNESS
            n.latency_ms = mm1_latency + backlog_latency

    def _update_statuses(self) -> None:
        """Transition node health based on queue boundaries.

        Recovery rules:
        - Scripted failures (Task 2 forced node kill): permanent, never auto-recover.
          Marked by is_scripted_failure=True, recovery_timer=0.
        - Overload failures (queue > FATAL_FAIL_THRESHOLD): auto-recover after
          NODE_RECOVERY_TICKS. The agent can learn to reroute away and let the
          node heal.
        """
        for n in self._nodes:
            # Scripted (task-forced) failures are permanent
            if n.is_scripted_failure:
                n.status = NodeStatus.FAILED
                n.recovery_timer = 0
                continue

            if n.queue_depth > FATAL_FAIL_THRESHOLD:
                if n.status != NodeStatus.FAILED:
                    n.status = NodeStatus.FAILED
                    n.recovery_timer = NODE_RECOVERY_TICKS
                    n.capacity = 0.5   # starts at half capacity when recovery begins
                    self._cascade_triggered = True  # Signal cascade detection
            elif n.queue_depth > OVERLOAD_THRESHOLD:
                n.status = NodeStatus.DEGRADED
            elif n.status == NodeStatus.DEGRADED and n.queue_depth < (OVERLOAD_THRESHOLD / 2):
                n.status = NodeStatus.HEALTHY

    def _cascade_failures(self) -> None:
        """Detect cascading failure: if a peer node's queue exceeds a heightened
        threshold within CASCADE_WINDOW_TICKS of a *new* failure, degrade it.

        Guardrails:
        - Only triggers when a NEW failure occurred this tick (not any failed node).
        - Graph-bounded: cascade only propagates along edges (parents or children).
        - Scripted failures (Task 2) do not trigger cascades.
        """
        if not self._cascade_triggered:
            self._cascade_tick = 0
            return

        self._cascade_tick += 1
        if self._cascade_tick > CASCADE_WINDOW_TICKS:
            self._cascade_triggered = False
            self._cascade_tick = 0
            return

        # Find all currently failed nodes
        failed_ids = {n.node_id for n in self._nodes if n.status == NodeStatus.FAILED}
        
        # Build set of nodes adjacent to any failed node (upstream or downstream)
        at_risk = set()
        for failed_id in failed_ids:
            # Downstream children of the failed node
            at_risk.update(CLUSTER_TOPOLOGY.get(failed_id, []))
            # Upstream parents of the failed node
            for parent_id, children in CLUSTER_TOPOLOGY.items():
                if failed_id in children:
                    at_risk.add(parent_id)

        cascade_threshold = FATAL_FAIL_THRESHOLD * CASCADE_QUEUE_MULTIPLIER
        for n in self._nodes:
            if n.node_id not in at_risk:
                continue   # Not adjacent to failure — cannot cascade
            if n.status == NodeStatus.FAILED or n.is_scripted_failure:
                continue
            if n.queue_depth > cascade_threshold:
                n.status = NodeStatus.DEGRADED

    def _process_recovery(self) -> None:
        """Count down recovery timers and bring FAILED nodes back online.

        Only overload-failed nodes (recovery_timer > 0) can recover.
        Scripted failures (is_scripted_failure=True) are excluded.
        """
        RECOVERY_RAMP_PER_TICK: float = 0.5   # capacity added per tick during recovery
        
        for n in self._nodes:
            if n.is_scripted_failure:
                continue
            # Check recovery_timer > 0, not status == FAILED: the first recovery
            # tick transitions the node to DEGRADED, but the timer must keep
            # counting until it reaches 0 and the node becomes HEALTHY.
            if n.recovery_timer > 0:
                n.recovery_timer -= 1
                if n.recovery_timer <= 0:
                    n.status = NodeStatus.HEALTHY
                    n.queue_depth = 0.0
                    n.latency_ms = BASE_LATENCY_MS
                    n.cpu_utilization = 0.0
                    # capacity stays at whatever it ramped to
                else:
                    # Still in recovery: ramp capacity up, stay DEGRADED
                    n.capacity = min(DEFAULT_CAPACITY, n.capacity + RECOVERY_RAMP_PER_TICK)
                    n.status = NodeStatus.DEGRADED   # not HEALTHY until fully ramped

    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()