refactor: replace boolean Path C fusion gate with a detailed audit dataclass and add matrix profile receipt validation logic

Author: apstenku123

cppmega_mlx/runtime/path_c_fusion.py

@@ -39,10 +39,12 @@
     "PathCFusionRegionBuilder",
     "PathCModelBrick",
     "PathCModelShapeEnv",
+    "PathCPlanDefaultEligibilityAudit",
     "SparseMLAFp8PrepareSpec",
     "WarmCacheAudit",
     "Z3SyncSpec",
     "audit_fusion_cache_key",
+    "audit_fused_path_c_plan_default_eligibility",
     "audit_warm_cache_reuse",
     "build_path_c_aot_autograd_region",
     "build_path_c_fusion_region",
@@ -2966,6 +2968,16 @@ def c_over_b(self) -> float:
         return self.path_c_warm_tok_sec / self.path_b_tok_sec
 
 
+@dataclass(frozen=True)
+class PathCPlanDefaultEligibilityAudit:
+    """Structured result for the fused Path C default-eligibility gate."""
+
+    eligible: bool
+    status: str
+    reason: str
+    schedule_id: str = ""
+
+
 def path_b_baseline_is_clean(
     row: BenchmarkAcceptanceRow,
     *,
@@ -2995,34 +3007,149 @@ def fused_path_c_default_eligible(
     return row.path_c_peak_delta_gib <= max_peak_delta_gib
 
 
-def fused_path_c_plan_default_eligible(
+def audit_fused_path_c_plan_default_eligibility(
     plan: FusionCompilePlan,
     row: BenchmarkAcceptanceRow,
     *,
     min_c_over_b: float = DEFAULT_MIN_C_OVER_B,
     max_peak_delta_gib: float = DEFAULT_MAX_PEAK_DELTA_GIB,
-) -> bool:
-    """Gate default Path C fusion on both the plan and clean benchmark row."""
+    trusted_schedule_ids: Sequence[str] | None = None,
+) -> PathCPlanDefaultEligibilityAudit:
+    """Explain whether a fused Path C plan can be enabled by default."""
 
     if not plan.single_kernel_fused:
-        return False
+        return PathCPlanDefaultEligibilityAudit(
+            eligible=False,
+            status="not_single_kernel_fused",
+            reason="plan did not lower to one fused kernel",
+        )
     if plan.schedule_status != "ready":
-        return False
+        return PathCPlanDefaultEligibilityAudit(
+            eligible=False,
+            status="schedule_not_ready",
+            reason=f"schedule status is {plan.schedule_status!r}, not 'ready'",
+        )
     if plan.autograd_status != "ready" or plan.autograd_missing_backward_nodes:
-        return False
+        return PathCPlanDefaultEligibilityAudit(
+            eligible=False,
+            status="autograd_not_ready",
+            reason="AOT autograd backward coverage is not complete",
+        )
     if plan.semantic_blockers:
-        return False
-    if plan.schedule_contract is None or plan.schedule_contract.status != "verified":
-        return False
-    if not plan.schedule_contract.declared_required_real_abi_inputs:
-        return False
-    if plan.schedule_contract.missing_real_abi_inputs:
-        return False
-    return fused_path_c_default_eligible(
+        return PathCPlanDefaultEligibilityAudit(
+            eligible=False,
+            status="semantic_blockers",
+            reason="semantic fusion blockers remain",
+        )
+    if plan.schedule_contract is None:
+        return PathCPlanDefaultEligibilityAudit(
+            eligible=False,
+            status="missing_schedule_contract",
+            reason="plan has no verified schedule contract",
+        )
+    contract = plan.schedule_contract
+    if contract.status != "verified":
+        return PathCPlanDefaultEligibilityAudit(
+            eligible=False,
+            status="unverified_schedule_contract",
+            reason=contract.reason,
+            schedule_id=contract.declared_schedule_id,
+        )
+    if contract.declared_implementation_kind != "production":
+        return PathCPlanDefaultEligibilityAudit(
+            eligible=False,
+            status="non_production_schedule",
+            reason=(
+                "declared implementation kind is "
+                f"{contract.declared_implementation_kind!r}, not 'production'"
+            ),
+            schedule_id=contract.declared_schedule_id,
+        )
+    if not contract.declared_schedule_id:
+        return PathCPlanDefaultEligibilityAudit(
+            eligible=False,
+            status="missing_schedule_id",
+            reason="verified production contract did not declare a schedule ID",
+        )
+    trusted_ids = (
+        trusted_path_c_production_schedule_ids()
+        if trusted_schedule_ids is None
+        else frozenset(str(schedule_id) for schedule_id in trusted_schedule_ids)
+    )
+    if contract.declared_schedule_id not in trusted_ids:
+        return PathCPlanDefaultEligibilityAudit(
+            eligible=False,
+            status="untrusted_schedule_id",
+            reason=(
+                f"production schedule ID {contract.declared_schedule_id!r} "
+                "is not trusted by this build"
+            ),
+            schedule_id=contract.declared_schedule_id,
+        )
+    if not contract.declared_required_real_abi_inputs:
+        return PathCPlanDefaultEligibilityAudit(
+            eligible=False,
+            status="undeclared_real_abi",
+            reason="verified production contract declared no real ABI inputs",
+            schedule_id=contract.declared_schedule_id,
+        )
+    if contract.missing_real_abi_inputs:
+        missing = ", ".join(contract.missing_real_abi_inputs)
+        return PathCPlanDefaultEligibilityAudit(
+            eligible=False,
+            status="incomplete_real_abi",
+            reason=f"verified production contract is missing real ABI inputs: {missing}",
+            schedule_id=contract.declared_schedule_id,
+        )
+    if not path_b_baseline_is_clean(row):
+        return PathCPlanDefaultEligibilityAudit(
+            eligible=False,
+            status="unclean_path_b_baseline",
+            reason="Path B baseline row is not clean enough for default gating",
+            schedule_id=contract.declared_schedule_id,
+        )
+    if row.c_over_b < min_c_over_b:
+        return PathCPlanDefaultEligibilityAudit(
+            eligible=False,
+            status="insufficient_path_c_speedup",
+            reason=f"Path C/B ratio {row.c_over_b:.6g} is below {min_c_over_b:.6g}",
+            schedule_id=contract.declared_schedule_id,
+        )
+    if row.path_c_peak_delta_gib > max_peak_delta_gib:
+        return PathCPlanDefaultEligibilityAudit(
+            eligible=False,
+            status="path_c_peak_delta_too_high",
+            reason=(
+                f"Path C peak delta {row.path_c_peak_delta_gib:.6g} GiB "
+                f"exceeds {max_peak_delta_gib:.6g} GiB"
+            ),
+            schedule_id=contract.declared_schedule_id,
+        )
+    return PathCPlanDefaultEligibilityAudit(
+        eligible=True,
+        status="eligible",
+        reason="plan and benchmark row satisfy fused Path C default gating",
+        schedule_id=contract.declared_schedule_id,
+    )
+
+
+def fused_path_c_plan_default_eligible(
+    plan: FusionCompilePlan,
+    row: BenchmarkAcceptanceRow,
+    *,
+    min_c_over_b: float = DEFAULT_MIN_C_OVER_B,
+    max_peak_delta_gib: float = DEFAULT_MAX_PEAK_DELTA_GIB,
+    trusted_schedule_ids: Sequence[str] | None = None,
+) -> bool:
+    """Gate default Path C fusion on both the plan and clean benchmark row."""
+
+    return audit_fused_path_c_plan_default_eligibility(
+        plan,
         row,
         min_c_over_b=min_c_over_b,
         max_peak_delta_gib=max_peak_delta_gib,
-    )
+        trusted_schedule_ids=trusted_schedule_ids,
+    ).eligible
 
 
 @dataclass(frozen=True)

scripts/m04_train_step.py

@@ -228,6 +228,12 @@
 PATH_C_FUSION_COMPILE_RECEIPT_PATH = (
     ROOT / "reports" / "path_c_fusion_compile_receipt.json"
 )
+PATH_C_FUSION_MATRIX_PROFILE_RECEIPT_ENV = (
+    "CPPMEGA_PATH_C_FUSION_MATRIX_PROFILE_RECEIPT"
+)
+PATH_C_FUSION_MATRIX_PROFILE_RECEIPT_PATH = (
+    ROOT / "reports" / "path_c_fusion_matrix_profile_receipt.json"
+)
 FP8_PATH_C_BRIDGE_TARGET = "native_mlx_tvm_ffi_graph_bridge"
 FP8_PATH_C_BRIDGE_STATUS = "m04_wired_for_native_tvm_ffi_graph_outputs"
 FP8_PATH_C_CARRIER_DTYPE = "bfloat16"
@@ -5767,10 +5773,114 @@ def _path_c_fusion_compile_receipt_payload(
     }
 
 
+def _path_c_fusion_matrix_profile_receipt_path() -> Path:
+    override = os.environ.get(PATH_C_FUSION_MATRIX_PROFILE_RECEIPT_ENV)
+    if override:
+        return Path(override).expanduser()
+    return PATH_C_FUSION_MATRIX_PROFILE_RECEIPT_PATH
+
+
+def _path_c_fusion_matrix_profile_payload(
+    *,
+    schedule_spec: Any,
+    receipt_path: str | Path | None = None,
+) -> dict[str, Any]:
+    """Validate 1B matrix/profile/cache evidence for a production schedule."""
+
+    receipt_path = (
+        Path(receipt_path).expanduser()
+        if receipt_path is not None
+        else _path_c_fusion_matrix_profile_receipt_path()
+    )
+    expected_schedule_id = str(schedule_spec.schedule_id)
+    expected_schedule_name = str(schedule_spec.schedule_name)
+    base: dict[str, Any] = {
+        "path": str(receipt_path),
+        "status": "missing",
+        "verified": False,
+        "schedule_id": None,
+        "schedule_name": None,
+        "failed_checks": [],
+        "reason": "production fused schedule 1B matrix/profile receipt is missing",
+    }
+    if not receipt_path.exists():
+        return base
+    try:
+        receipt = json.loads(receipt_path.read_text(encoding="utf-8"))
+    except (OSError, json.JSONDecodeError) as exc:
+        return {
+            **base,
+            "status": "unreadable",
+            "reason": f"{type(exc).__name__}: {exc}",
+        }
+
+    matrix_rows = receipt.get("matrix_rows", ())
+    checks = {
+        "receipt_kind_ok": (
+            receipt.get("kind")
+            == "cppmega_path_c_fusion_matrix_profile_receipt"
+        ),
+        "receipt_status_ok": receipt.get("status") == "ok",
+        "model_profile_matches": (
+            receipt.get("model_profile") == REQUIRED_MODEL_PROFILE
+        ),
+        "schedule_id_matches": (
+            str(receipt.get("schedule_id", "")) == expected_schedule_id
+        ),
+        "schedule_name_matches": (
+            str(receipt.get("schedule_name", "")) == expected_schedule_name
+        ),
+        "full_1b_matrix_captured": (
+            receipt.get("full_1b_matrix_captured") is True
+        ),
+        "profiling_traces_captured": (
+            receipt.get("profiling_traces_captured") is True
+        ),
+        "memory_non_regression_ok": (
+            receipt.get("memory_non_regression_ok") is True
+        ),
+        "cache_receipts_captured": (
+            receipt.get("cache_receipts_captured") is True
+        ),
+        "path_b_baselines_clean": (
+            receipt.get("path_b_baselines_clean") is True
+        ),
+        "path_c_default_gate_rows_passed": (
+            receipt.get("path_c_default_gate_rows_passed") is True
+        ),
+        "path_c_peak_memory_non_regression": (
+            receipt.get("path_c_peak_memory_non_regression") is True
+        ),
+        "path_c_warm_cache_hit_observed": (
+            receipt.get("path_c_warm_cache_hit_observed") is True
+        ),
+        "matrix_rows_present": isinstance(matrix_rows, list) and bool(matrix_rows),
+    }
+    failed_checks = [name for name, ok in checks.items() if not ok]
+    verified = not failed_checks
+    return {
+        **base,
+        "status": "verified" if verified else "mismatch",
+        "verified": verified,
+        "schedule_id": receipt.get("schedule_id"),
+        "schedule_name": receipt.get("schedule_name"),
+        "checks": checks,
+        "failed_checks": failed_checks,
+        "matrix_row_count": len(matrix_rows) if isinstance(matrix_rows, list) else 0,
+        "reason": (
+            "production fused schedule has matching full 1B matrix, profiling, "
+            "memory non-regression, and cache receipt evidence"
+            if verified
+            else "production fused schedule matrix/profile receipt does not match this route"
+        ),
+    }
+
+
 def path_c_fusion_payload(
     *,
     model: Any | None = None,
     compile_receipt_path: str | Path | None = None,
+    matrix_profile_receipt_path: str | Path | None = None,
     bank_buffers: Mapping[str, Any] | None = None,
     bank_buffer_owner: str | None = None,
     bank_owner: Any | None = None,
@@ -5825,6 +5935,13 @@ def path_c_fusion_payload(
         receipt_path=compile_receipt_path,
     )
     production_compile_verified = bool(compile_receipt.get("verified"))
+    matrix_profile_receipt = _path_c_fusion_matrix_profile_payload(
+        schedule_spec=schedule_spec,
+        receipt_path=matrix_profile_receipt_path,
+    )
+    production_matrix_profile_verified = bool(
+        matrix_profile_receipt.get("verified")
+    )
     expected_bank_buffer_owner = (
         bank_buffer_owner
         if bank_buffer_owner is not None
@@ -6109,6 +6226,7 @@ def path_c_fusion_payload(
             fused_train_block_training_contract
         ),
         "production_compile_receipt": compile_receipt,
+        "production_matrix_profile_receipt": matrix_profile_receipt,
         "direct_chained_fusion": {
             **_path_c_direct_chain_plan_payload(direct_chain),
             "construction": direct_chain_construction,
@@ -6264,16 +6382,31 @@ def path_c_fusion_payload(
                 if not runtime_fused_route_bound
                 else []
             ),
-            {
-                "kind": "production_1b_matrix_profile_missing",
-                "schedule_id": schedule_spec.schedule_id,
-                "schedule_name": schedule_spec.schedule_name,
-                "reason": (
-                    "the full 1B Path B/Path C matrix, profiling traces, memory "
-                    "non-regression evidence, and cache receipts have not been "
-                    "captured for this production schedule"
-                ),
-            },
+            *(
+                [
+                    {
+                        "kind": "production_1b_matrix_profile_missing",
+                        "schedule_id": schedule_spec.schedule_id,
+                        "schedule_name": schedule_spec.schedule_name,
+                        "matrix_profile_receipt_status": (
+                            matrix_profile_receipt.get("status")
+                        ),
+                        "matrix_profile_receipt_path": (
+                            matrix_profile_receipt.get("path")
+                        ),
+                        "failed_checks": list(
+                            matrix_profile_receipt.get("failed_checks", ())
+                        ),
+                        "reason": (
+                            "the full 1B Path B/Path C matrix, profiling traces, "
+                            "memory non-regression evidence, and cache receipts "
+                            "have not been captured for this production schedule"
+                        ),
+                    }
+                ]
+                if not production_matrix_profile_verified
+                else []
+            ),
             *(
                 [
                     {
@@ -6352,6 +6485,9 @@ def path_c_fusion_payload(
             "requires_ready_fusion_plan": True,
             "requires_compile_verified_single_kernel": True,
             "current_compile_receipt_verified": production_compile_verified,
+            "current_matrix_profile_verified": (
+                production_matrix_profile_verified
+            ),
             "requires_verified_schedule_contract": True,
             "requires_complete_real_abi_contract": True,
             "current_plan_default_eligible": False,