Flag repairs that pass by deleting a hardware guard (dataset quality)

CHANGELOG.md

@@ -10,6 +10,14 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
 ## [Unreleased]
 
 ### Added
+- **Anti-cheat flags repairs that pass by deleting a hardware guard** (dataset-quality signal):
+  `anti_cheat.detect_removed_hardware_guard` + `episodes.episode_removed_guard` detect the
+  validator-hack seen in the merged data — a repair that "passes" by removing a device / SM /
+  compute-capability / architecture assertion the failed attempt had (e.g. an SM90 guard that
+  failed on the real SM120 GPU) instead of fixing the kernel. The release gate records a
+  **non-blocking** `repair_guard_removed_rows` count (+ task ids) in `dataset_manifest.json` — these
+  kernels still passed hardware validation, so they are surfaced for review / reward down-weighting,
+  not dropped. Complements the portable-prompt nudge (which prevents the guard being added at all).
 - **Correctness preference pairs for DPO** (`triton_dataset/benchmark_pairs.py`,
   `dpo_export.py`, `sparkproof-export-dpo --pair-type`): the DPO exporter previously built
   only *optimization* pairs (faster-vs-slower, both passing, requires monitored timings).

sparkproof/triton_dataset/anti_cheat.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import ast
+import re
 from typing import Any
 
 FORBIDDEN_LAUNCHER_CALLS = frozenset(
@@ -242,6 +243,40 @@ def detect_timing_manipulation(code: str) -> list[str]:
     return issues
 
 
+# A repair can "pass" by deleting a device/architecture guard the prior attempt
+# added (e.g. an SM90 assertion that failed on the real SM120 validation GPU)
+# rather than fixing the kernel. Flag that so it can be reviewed / down-weighted.
+_HW_GUARD_RE = re.compile(
+    r"get_device_capability|compute[_ ]?capab|get_device_name|capability\s*\(|"
+    r"targets?\s+(?:Hopper|Blackwell|Ampere|SM\d)|"
+    r"(?:assert|raise|RuntimeError|ValueError).{0,90}(?:capab|architect|Hopper|Blackwell|Ampere|\bSM\d)",
+    re.I,
+)
+
+
+def _hardware_guard_lines(code: str) -> list[str]:
+    return [ln.strip() for ln in (code or "").splitlines() if _HW_GUARD_RE.search(ln)]
+
+
+def detect_removed_hardware_guard(previous_code: str, repaired_code: str) -> str | None:
+    """Flag a repair that dropped a device/architecture guard the prior attempt had.
+
+    Returns a short reason when ``previous_code`` contained hardware/compute-capability/
+    SM/architecture guards and ``repaired_code`` has strictly fewer of them (i.e. a guard
+    was deleted rather than the kernel actually fixed). Conservative: only fires on a net
+    decrease in guard lines, so a repair that keeps or strengthens its checks is not flagged.
+    """
+    prev = _hardware_guard_lines(previous_code)
+    if not prev:
+        return None
+    new = _hardware_guard_lines(repaired_code)
+    if len(new) < len(prev):
+        removed = [g for g in prev if g not in new]
+        if removed:
+            return f"repair removed hardware/architecture guard: {removed[0][:100]}"
+    return None
+
+
 def analyze_anti_cheat(code: str) -> dict[str, Any]:
     issues = detect_torch_fallbacks(code) + detect_timing_manipulation(code)
     return {"passed": not issues, "issues": issues}

sparkproof/triton_dataset/episodes.py

@@ -9,6 +9,8 @@
 
 from typing import Any
 
+from sparkproof.triton_dataset.anti_cheat import detect_removed_hardware_guard
+from sparkproof.triton_dataset.code_extract import extract_fenced_python
 from sparkproof.triton_dataset.training_cot import (
     normalize_training_reasoning,
     prose_rationale_from_response,
@@ -158,3 +160,31 @@ def episode_to_messages(episode: dict[str, Any]) -> list[dict[str, str]]:
 def trajectory_has_episode(trajectory: dict[str, Any]) -> bool:
     episode = (trajectory.get("metadata") or {}).get("episode")
     return isinstance(episode, dict) and bool(episode.get("turns"))
+
+
+def episode_removed_guard(episode: dict[str, Any]) -> str | None:
+    """Flag an episode whose repair passed by deleting a hardware/architecture guard.
+
+    Scans each failed assistant turn against the following assistant turn: if the later
+    code dropped a device/SM/compute-capability guard the failed attempt had, return the
+    reason. Non-blocking signal — the resulting kernel still passed hardware validation, so
+    this surfaces a suspicious repair for review rather than proving incorrectness.
+    """
+    assistants = [t for t in (episode.get("turns") or []) if t.get("role") == "assistant"]
+    for prev, nxt in zip(assistants, assistants[1:]):
+        if prev.get("passed") is False:
+            reason = detect_removed_hardware_guard(
+                extract_fenced_python(prev.get("content") or ""),
+                extract_fenced_python(nxt.get("content") or ""),
+            )
+            if reason:
+                return reason
+    return None
+
+
+def trajectory_removed_guard(trajectory: dict[str, Any]) -> str | None:
+    """Guard-removal flag for a published trajectory (via its episode), or None."""
+    episode = (trajectory.get("metadata") or {}).get("episode")
+    if isinstance(episode, dict):
+        return episode_removed_guard(episode)
+    return None

sparkproof/triton_dataset/release_gate.py

@@ -9,6 +9,7 @@
 from typing import Any
 
 from sparkproof.triton_dataset.decontaminate import TritonDecontaminator, extract_python_from_response
+from sparkproof.triton_dataset.episodes import trajectory_removed_guard
 from sparkproof.triton_dataset.novelty import NoveltyRegistry, compute_novelty_report
 from sparkproof.triton_dataset.task_policy import FORBIDDEN_TRAINING_ORIGINS
 
@@ -187,9 +188,21 @@ def run_release_gate(
     novelty_report = compute_novelty_report(verified_rows, registry).to_dict()
     (bundle_dir / "novelty_report.json").write_text(json.dumps(novelty_report, indent=2))
 
+    # Non-blocking quality signal: verified rows whose repair passed by deleting a
+    # hardware/architecture guard. These kernels still passed hardware validation (so they
+    # are not blocked), but the pattern is surfaced for review / reward down-weighting.
+    guard_removed = [
+        (traj.get("metadata") or {}).get("prompt_meta", {}).get("task_id")
+        for traj in verified_rows
+        if trajectory_removed_guard(traj)
+    ]
+
     manifest = build_manifest(trajectories=trajectories, dataset_version=dataset_version, bundle_dir=bundle_dir)
     manifest["blocked_rows"] = len(blocked)
     manifest["passed"] = len(blocked) == 0
+    manifest["repair_guard_removed_rows"] = len(guard_removed)
+    if guard_removed:
+        manifest["repair_guard_removed_task_ids"] = [t for t in guard_removed if t][:50]
     # Duplicates don't fail the gate — decontamination blocks eval leakage, novelty
     # only feeds reward accounting (novel_verified_rows), per issue #9's design.
     manifest["novelty"] = novelty_report

tests/test_guard_removal.py

@@ -0,0 +1,61 @@
+"""Flag repairs that pass by deleting a hardware/architecture guard (dataset quality)."""
+
+from __future__ import annotations
+
+from sparkproof.triton_dataset.anti_cheat import detect_removed_hardware_guard
+from sparkproof.triton_dataset.episodes import (
+    episode_removed_guard,
+    make_turn,
+    trajectory_removed_guard,
+)
+
+GUARDED = (
+    "import torch\n"
+    'assert torch.cuda.get_device_capability()[0] == 9, "This test targets Hopper SM90"\n'
+    "def launch(x):\n    return x\n"
+)
+GUARD_REMOVED = "import torch\ndef launch(x):\n    return x\n"
+
+
+def test_detects_removed_guard():
+    assert detect_removed_hardware_guard(GUARDED, GUARD_REMOVED) is not None
+
+
+def test_no_flag_when_guard_kept():
+    assert detect_removed_hardware_guard(GUARDED, GUARDED) is None
+
+
+def test_no_flag_when_no_guard_present():
+    assert detect_removed_hardware_guard("def f():\n    return 1\n", "def f():\n    return 2\n") is None
+
+
+def _episode(prev_code: str, prev_passed: bool, new_code: str) -> dict:
+    return {
+        "turns": [
+            make_turn(role="user", kind="task", content="write a kernel"),
+            make_turn(role="assistant", kind="attempt", content=f"```python\n{prev_code}```", passed=prev_passed),
+            make_turn(role="user", kind="validator_feedback", content="FAILED (compile_execute_failed)"),
+            make_turn(role="assistant", kind="repair", content=f"```python\n{new_code}```", passed=True),
+        ]
+    }
+
+
+def test_episode_flags_guard_removal_after_failure():
+    ep = _episode(GUARDED, prev_passed=False, new_code=GUARD_REMOVED)
+    assert episode_removed_guard(ep) is not None
+    assert trajectory_removed_guard({"metadata": {"episode": ep}}) is not None
+
+
+def test_episode_not_flagged_when_prior_attempt_passed():
+    # Guard removed but the prior attempt already passed → not a fix-by-deletion.
+    ep = _episode(GUARDED, prev_passed=True, new_code=GUARD_REMOVED)
+    assert episode_removed_guard(ep) is None
+
+
+def test_episode_not_flagged_when_guard_kept():
+    ep = _episode(GUARDED, prev_passed=False, new_code=GUARDED)
+    assert episode_removed_guard(ep) is None
+
+
+def test_no_episode_no_flag():
+    assert trajectory_removed_guard({"metadata": {}}) is None