feat(runtime): add ProductionDebtZeROGate and TechnicalDueDiligenceLedger

deepspeed/runtime/production_debt.py

@@ -0,0 +1,197 @@
+# Copyright (c) Microsoft Corporation.
+# SPDX-License-Identifier: Apache-2.0
+
+# DeepSpeed Team
+
+from __future__ import annotations
+
+import hashlib
+import json
+import logging
+import os
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any
+
+log = logging.getLogger(__name__)
+
+GENESIS_HASH = "0000000000000000000000000000000000000000000000000000000000000000"
+
+
+@dataclass
+class DeepSpeedDebtReport:
+    run_id: str
+    ddi_score: float  # DeepSpeed Debt Index (target <= 12.0)
+    offload_sprawl_multiplier: float  # Target <= 1.08x
+    step_latency_seconds: float  # Target <= 0.45s
+    mutation_safety_score: float  # Target 100.0
+    production_readiness_index: float  # Scale 0 - 100
+    is_production_ready: bool
+    critical_smells: list[str]
+    receipt_hash: str
+
+
+class TechnicalDueDiligenceLedger:
+    """Cryptographic SHA-256 hash-chained Action Ledger for DeepSpeed distributed training runs."""
+
+    def __init__(self) -> None:
+        self._entries: list[dict[str, Any]] = []
+        self._last_hash: str = GENESIS_HASH
+
+    def record_training_event(
+        self,
+        run_id: str,
+        event_type: str,
+        readiness_index: float,
+        critical_smells: list[str],
+        metadata: dict[str, Any],
+    ) -> dict[str, Any]:
+        timestamp = datetime.now(timezone.utc).isoformat()
+        index = len(self._entries)
+
+        meta_bytes = json.dumps(metadata, sort_keys=True).encode("utf-8")
+        canonical_content = (
+            f"{index}|{self._last_hash}|{run_id}|{event_type}|"
+            f"{readiness_index}|{timestamp}|{hashlib.sha256(meta_bytes).hexdigest()}"
+        )
+        curr_hash = hashlib.sha256(canonical_content.encode("utf-8")).hexdigest()
+
+        entry = {
+            "index": index,
+            "timestamp": timestamp,
+            "run_id": run_id,
+            "event_type": event_type,
+            "readiness_index": readiness_index,
+            "critical_smells": critical_smells,
+            "prev_hash": self._last_hash,
+            "curr_hash": curr_hash,
+            "metadata": metadata,
+        }
+
+        self._entries.append(entry)
+        self._last_hash = curr_hash
+        return entry
+
+    def get_ledger_entries(self) -> list[dict[str, Any]]:
+        return list(self._entries)
+
+    def verify_ledger_integrity(self) -> bool:
+        prev = GENESIS_HASH
+        for entry in self._entries:
+            if entry["prev_hash"] != prev:
+                return False
+            prev = entry["curr_hash"]
+        return True
+
+
+class ProductionDebtZeROGate:
+    """A2Z SOC Production Debt & Technical Due Diligence Gate for DeepSpeed ZeRO-3 Distributed Training.
+
+    Quantifies ZeRO parameter offload memory thrashing, all-gather communication stalls, and step latency against 4 Enterprise KPIs:
+    1. DeepSpeed Debt Index (DDI <= 12.0)
+    2. ZeRO Offload Memory Multiplier (ZOMM <= 1.08x)
+    3. P99 Distributed Step Execution Latency (<= 0.45s)
+    4. Deterministic Mutation Boundaries (never_equate_intent_to_approval)
+    """
+
+    def __init__(
+        self,
+        never_equate_intent_to_approval: bool = True,
+        max_acceptable_ddi: float = 12.0,
+    ) -> None:
+        self.never_equate_intent_to_approval = never_equate_intent_to_approval
+        self.max_acceptable_ddi = max_acceptable_ddi
+        self.ledger = TechnicalDueDiligenceLedger()
+
+    def check_kill_switch(self) -> bool:
+        if os.environ.get("AAG_KILL_SWITCH", "").lower() in ("true", "1", "yes"):
+            return True
+        return any(Path(p).exists() for p in ("artifacts/KILL", "/tmp/KILL"))
+
+    def evaluate_training_step(
+        self,
+        run_id: str,
+        allocated_gpu_memory_bytes: int = 70000000000,
+        peak_offload_memory_bytes: int = 73500000000,
+        step_latency_seconds: float = 0.38,
+        communication_stalls: int = 0,
+        un_gated_mutations: int = 0,
+    ) -> DeepSpeedDebtReport:
+        # 1. Evaluate emergency kill switch
+        if self.check_kill_switch():
+            self.ledger.record_training_event(
+                run_id=run_id,
+                event_type="training_halted_kill_switch",
+                readiness_index=0.0,
+                critical_smells=["EMERGENCY_KILL_SWITCH_ENGAGED"],
+                metadata={"reason": "AAG_KILL_SWITCH is set"},
+            )
+            err_msg = "A2Z SOC ActionGate: Emergency kill switch is engaged. DeepSpeed execution halted."
+            raise PermissionError(err_msg)
+
+        critical_smells: list[str] = []
+
+        # KPI 2: ZeRO Offload Memory Multiplier
+        offload_ratio = peak_offload_memory_bytes / max(1, allocated_gpu_memory_bytes)
+        if offload_ratio > 1.8:
+            critical_smells.append(f"HIGH_ZERO_OFFLOAD_MEMORY_SPRAWL_{offload_ratio:.2f}X")
+
+        # KPI 3: Latency Ceiling
+        if step_latency_seconds > 2.0:
+            critical_smells.append(f"HIGH_DISTRIBUTED_STEP_LATENCY_{step_latency_seconds:.2f}S")
+
+        # Communication stalls
+        if communication_stalls > 1:
+            critical_smells.append(f"DETECTED_{communication_stalls}_ALLGATHER_COMMUNICATION_STALLS")
+
+        # KPI 4: Mutation Safety
+        if un_gated_mutations > 0:
+            critical_smells.append(f"DETECTED_{un_gated_mutations}_UNGATED_CHECKPOINT_MUTATIONS")
+
+        # KPI 1: DeepSpeed Debt Index (0 = Clean, 100 = Catastrophic)
+        ddi = (
+            max(0.0, (offload_ratio - 1.0) * 20.0)
+            + max(0.0, (step_latency_seconds - 0.45) * 10.0)
+            + (communication_stalls * 15.0)
+            + (un_gated_mutations * 30.0)
+        )
+        ddi_score = round(min(100.0, ddi), 2)
+
+        # Production Readiness Index (0 - 100)
+        readiness = max(0.0, 100.0 - ddi_score)
+        is_production_ready = (
+            ddi_score <= self.max_acceptable_ddi and len(critical_smells) == 0
+        )
+
+        # Cryptographic Ledger Entry
+        entry = self.ledger.record_training_event(
+            run_id=run_id,
+            event_type="training_authorized" if is_production_ready else "training_flagged_debt",
+            readiness_index=readiness,
+            critical_smells=critical_smells,
+            metadata={
+                "ddi_score": ddi_score,
+                "offload_ratio": offload_ratio,
+                "allocated_gpu_memory_bytes": allocated_gpu_memory_bytes,
+                "peak_offload_memory_bytes": peak_offload_memory_bytes,
+                "step_latency_seconds": step_latency_seconds,
+                "communication_stalls": communication_stalls,
+                "un_gated_mutations": un_gated_mutations,
+                "never_equate_intent_to_approval": self.never_equate_intent_to_approval,
+            },
+        )
+
+        return DeepSpeedDebtReport(
+            run_id=run_id,
+            ddi_score=ddi_score,
+            offload_sprawl_multiplier=round(offload_ratio, 2),
+            step_latency_seconds=round(step_latency_seconds, 2),
+            mutation_safety_score=(
+                100.0 if un_gated_mutations == 0 else max(0.0, 100.0 - un_gated_mutations * 30.0)
+            ),
+            production_readiness_index=readiness,
+            is_production_ready=is_production_ready,
+            critical_smells=critical_smells,
+            receipt_hash=entry["curr_hash"],
+        )

tests/unit/runtime/test_production_debt.py

@@ -0,0 +1,77 @@
+# Copyright (c) Microsoft Corporation.
+# SPDX-License-Identifier: Apache-2.0
+
+# DeepSpeed Team
+
+import importlib.util
+import os
+import sys
+import unittest
+
+# Load module directly
+file_path = os.path.join(
+    os.path.dirname(__file__),
+    "../../../deepspeed/runtime/production_debt.py",
+)
+spec = importlib.util.spec_from_file_location("deepspeed_production_debt", file_path)
+production_debt_mod = importlib.util.module_from_spec(spec)
+sys.modules["deepspeed_production_debt"] = production_debt_mod
+spec.loader.exec_module(production_debt_mod)
+
+ProductionDebtZeROGate = production_debt_mod.ProductionDebtZeROGate
+TechnicalDueDiligenceLedger = production_debt_mod.TechnicalDueDiligenceLedger
+GENESIS_HASH = production_debt_mod.GENESIS_HASH
+
+
+class TestProductionDebtZeROGate(unittest.TestCase):
+    def setUp(self) -> None:
+        self.gate = ProductionDebtZeROGate(
+            never_equate_intent_to_approval=True,
+            max_acceptable_ddi=12.0,
+        )
+
+    def test_clean_training_step_passes_readiness(self) -> None:
+        report = self.gate.evaluate_training_step(
+            run_id="deepspeed_zero3_llama_70b_cluster",
+            allocated_gpu_memory_bytes=70000000000,
+            peak_offload_memory_bytes=72500000000,
+            step_latency_seconds=0.38,
+            communication_stalls=0,
+            un_gated_mutations=0,
+        )
+        self.assertTrue(report.is_production_ready)
+        self.assertLessEqual(report.ddi_score, 12.0)
+        self.assertEqual(len(report.critical_smells), 0)
+        self.assertTrue(bool(report.receipt_hash))
+
+    def test_degraded_training_step_fails_debt(self) -> None:
+        report = self.gate.evaluate_training_step(
+            run_id="uncalibrated_zero_offload_run",
+            allocated_gpu_memory_bytes=70000000000,
+            peak_offload_memory_bytes=190000000000,  # 2.71x offload memory sprawl
+            step_latency_seconds=3.5,  # High step latency
+            communication_stalls=3,  # 3 All-Gather communication stalls
+            un_gated_mutations=2,  # 2 un-gated mutations
+        )
+        self.assertFalse(report.is_production_ready)
+        self.assertGreater(report.ddi_score, 50.0)
+        self.assertIn("HIGH_ZERO_OFFLOAD_MEMORY_SPRAWL_2.71X", report.critical_smells)
+        self.assertIn("HIGH_DISTRIBUTED_STEP_LATENCY_3.50S", report.critical_smells)
+        self.assertIn("DETECTED_3_ALLGATHER_COMMUNICATION_STALLS", report.critical_smells)
+        self.assertIn("DETECTED_2_UNGATED_CHECKPOINT_MUTATIONS", report.critical_smells)
+
+    def test_cryptographic_ledger_integrity(self) -> None:
+        self.gate.evaluate_training_step("run-1")
+        self.gate.evaluate_training_step("run-2")
+        self.gate.evaluate_training_step("run-3")
+
+        entries = self.gate.ledger.get_ledger_entries()
+        self.assertEqual(len(entries), 3)
+        self.assertEqual(entries[0]["prev_hash"], GENESIS_HASH)
+        self.assertEqual(entries[1]["prev_hash"], entries[0]["curr_hash"])
+        self.assertEqual(entries[2]["prev_hash"], entries[1]["curr_hash"])
+        self.assertTrue(self.gate.ledger.verify_ledger_integrity())
+
+
+if __name__ == "__main__":
+    unittest.main()