feat: AST-level Python analysis + MI300X benchmark suite

- AST Transformer: Uses Python's ast module for tree-level analysis
- Detects device assignments, CUDA call sites, embedded kernels
- Identifies env mutations and inline kernel strings structurally
- Not regex — real compiler-level understanding of code structure
- MI300X Benchmark Suite: Ready to run on AMD Developer Cloud
- Device info, memory bandwidth, and GEMM TFLOPS tests
- Saves JSON results as proof of AMD GPU usage
- Wired AST findings into analyzer, API, and orchestrator

agents/analyzer.py

@@ -19,6 +19,7 @@ from knowledge.cuda_mappings import (
     HARDWARE_AWARE_MAPPINGS,
     IMPLICIT_CUDA_PATTERNS,
 )
+from agents.ast_transformer import ASTTransformer
 
 
 @dataclass
@@ -54,6 +55,7 @@ class AnalysisResult:
     migration_level: str = "Easy"
     code_type: str = "python"  # python, cpp, dockerfile, requirements
     summary: Dict = field(default_factory=dict)
+    ast_findings: List[Dict] = field(default_factory=list)  # AST-level analysis results
     trace_log: List[str] = field(default_factory=list)
 
 
@@ -114,6 +116,10 @@ class AnalyzerAgent:
         self._scan_implicit_assumptions(lines, code, result)
         result.trace_log.append(f"🧪 Exploration Scan → Found {len(result.implicit_assumptions)} implicit CUDA assumptions")
         
+        # AST-level analysis (Python only — real compiler-level understanding)
+        if result.code_type == "python":
+            self._run_ast_analysis(code, result)
+        
         # Build saliency map (per-line risk scoring)
         self._build_saliency_map(lines, result)
         result.trace_log.append(f"🎯 Saliency Map → {sum(1 for v in result.saliency_map.values() if v == 'critical')} critical lines identified")
@@ -500,6 +506,7 @@ class AnalyzerAgent:
             "known_issues": len(result.known_issues),
             "hardware_issues": len(result.hardware_issues),
             "implicit_assumptions": len(result.implicit_assumptions),
+            "ast_findings": len(result.ast_findings),
             "critical_lines": sum(1 for v in result.saliency_map.values() if v == "critical"),
             "migration_score": result.migration_score,
             "migration_health": result.migration_health,
@@ -509,3 +516,19 @@ class AnalyzerAgent:
             "errors": sum(1 for p in result.detected_patterns if p.severity == "error"),
             "compatible": sum(1 for p in result.detected_patterns if p.severity == "info"),
         }
+
+    def _run_ast_analysis(self, code: str, result: AnalysisResult):
+        """
+        Run AST-level analysis on Python code using Python's ast module.
+        This is a real compiler-level pass — not regex.
+        """
+        transformer = ASTTransformer()
+        findings, trace = transformer.analyze(code)
+        result.ast_findings = findings
+        result.trace_log.extend(trace)
+
+        # Count critical AST findings
+        critical = sum(1 for f in findings if f.get("severity") == "critical")
+        if critical > 0:
+            result.trace_log.append(f"🌳 AST: {critical} critical finding(s) — embedded kernels require manual review")
+

agents/ast_transformer.py

@@ -0,0 +1,237 @@
+"""
+ROCm Forge — AST-Level Python Transformer
+Uses Python's ast module to perform tree-level code transformation,
+going beyond regex to understand code STRUCTURE.
+
+This is the critical differentiator from hipify and other string-based tools.
+"""
+import ast
+import re
+from typing import List, Tuple, Dict
+
+
+class CUDANodeVisitor(ast.NodeVisitor):
+    """
+    AST visitor that walks the Python syntax tree and detects
+    CUDA-specific constructs at the structural level.
+    """
+
+    def __init__(self):
+        self.findings: List[Dict] = []
+        self.device_vars: Dict[str, int] = {}  # var_name -> line_no
+        self.cuda_call_sites: List[Dict] = []
+        self.env_mutations: List[Dict] = []
+        self.string_literals_with_cuda: List[Dict] = []
+
+    def visit_Assign(self, node: ast.Assign):
+        """Detect device = torch.device('cuda:0') assignments."""
+        if isinstance(node.value, ast.Call):
+            func = node.value
+            func_name = self._get_func_name(func)
+
+            # torch.device("cuda:X") pattern
+            if func_name == "torch.device" and func.args:
+                arg = func.args[0]
+                if isinstance(arg, ast.Constant) and isinstance(arg.value, str):
+                    if "cuda" in arg.value:
+                        for target in node.targets:
+                            if isinstance(target, ast.Name):
+                                self.device_vars[target.id] = node.lineno
+                        self.findings.append({
+                            "line": node.lineno,
+                            "type": "device_assignment",
+                            "original": f'torch.device("{arg.value}")',
+                            "suggestion": f'torch.device("{arg.value}")  # Works on ROCm — PyTorch ROCm uses cuda namespace',
+                            "severity": "info",
+                        })
+
+        # os.environ["CUDA_VISIBLE_DEVICES"] = ... pattern
+        if isinstance(node.value, ast.Constant) and len(node.targets) == 1:
+            target = node.targets[0]
+            if isinstance(target, ast.Subscript):
+                if isinstance(target.slice, ast.Constant) and isinstance(target.slice.value, str):
+                    if "CUDA" in target.slice.value:
+                        self.env_mutations.append({
+                            "line": node.lineno,
+                            "var": target.slice.value,
+                            "type": "env_mutation",
+                        })
+
+        self.generic_visit(node)
+
+    def visit_Call(self, node: ast.Call):
+        """Detect CUDA API calls: .cuda(), torch.cuda.*, etc."""
+        func_name = self._get_func_name(node)
+
+        # .cuda() method calls
+        if func_name and func_name.endswith(".cuda"):
+            self.cuda_call_sites.append({
+                "line": node.lineno,
+                "call": func_name,
+                "type": "cuda_method",
+                "note": ".cuda() works on ROCm — maps to HIP backend",
+            })
+
+        # torch.cuda.is_available(), torch.cuda.device_count(), etc.
+        if func_name and "torch.cuda" in func_name:
+            self.cuda_call_sites.append({
+                "line": node.lineno,
+                "call": func_name,
+                "type": "torch_cuda_api",
+                "note": "torch.cuda.* APIs work on ROCm via HIP backend",
+            })
+
+        # load_inline / cpp_extension calls
+        if func_name and ("load_inline" in func_name or "load" in func_name):
+            for kw in node.keywords:
+                if kw.arg == "cuda_sources":
+                    self.findings.append({
+                        "line": node.lineno,
+                        "type": "inline_kernel",
+                        "original": "cuda_sources=[...]",
+                        "suggestion": "hip_sources=[...] (rename parameter for clarity)",
+                        "severity": "review",
+                    })
+
+        self.generic_visit(node)
+
+    def visit_Constant(self, node: ast.Constant):
+        """Detect string literals containing CUDA code."""
+        if isinstance(node.value, str) and len(node.value) > 50:
+            cuda_indicators = [
+                "cuda_runtime.h", "cudaMalloc", "cudaFree",
+                "__global__", "cudaMemcpy", "cudaDeviceSynchronize",
+                "<<<", ">>>",
+            ]
+            if any(ind in node.value for ind in cuda_indicators):
+                self.string_literals_with_cuda.append({
+                    "line": node.lineno,
+                    "length": len(node.value),
+                    "type": "embedded_cuda_kernel",
+                    "note": "Embedded CUDA C/C++ kernel detected in string literal",
+                })
+        self.generic_visit(node)
+
+    def visit_If(self, node: ast.If):
+        """Detect `if not torch.cuda.is_available()` guards."""
+        test_src = ast.dump(node.test)
+        if "torch" in test_src and "cuda" in test_src:
+            self.findings.append({
+                "line": node.lineno,
+                "type": "cuda_guard",
+                "original": "CUDA availability check",
+                "suggestion": "Works on ROCm — torch.cuda.is_available() returns True with HIP backend",
+                "severity": "info",
+            })
+        self.generic_visit(node)
+
+    def visit_ImportFrom(self, node: ast.ImportFrom):
+        """Detect imports from CUDA-specific modules."""
+        if node.module and "cuda" in node.module.lower():
+            self.findings.append({
+                "line": node.lineno,
+                "type": "cuda_import",
+                "original": f"from {node.module} import ...",
+                "suggestion": "Check ROCm compatibility for this import",
+                "severity": "review",
+            })
+        self.generic_visit(node)
+
+    def _get_func_name(self, node: ast.Call) -> str:
+        """Extract dotted function name from a Call node."""
+        if isinstance(node.func, ast.Name):
+            return node.func.id
+        elif isinstance(node.func, ast.Attribute):
+            parts = []
+            current = node.func
+            while isinstance(current, ast.Attribute):
+                parts.append(current.attr)
+                current = current.value
+            if isinstance(current, ast.Name):
+                parts.append(current.id)
+            return ".".join(reversed(parts))
+        return ""
+
+
+class ASTTransformer:
+    """
+    AST-level Python code transformer.
+    Performs structural analysis beyond regex — understands scope, assignment,
+    function calls, and string literal contexts.
+    """
+
+    def __init__(self):
+        self.trace_log: List[str] = []
+        self.ast_findings: List[Dict] = []
+
+    def analyze(self, code: str) -> Tuple[List[Dict], List[str]]:
+        """
+        Perform AST-level analysis on Python code.
+        Returns (findings, trace_log).
+        Falls back gracefully if code can't be parsed.
+        """
+        self.trace_log = []
+        self.ast_findings = []
+
+        try:
+            tree = ast.parse(code)
+            self.trace_log.append("🌳 AST parse successful — performing tree-level analysis")
+        except SyntaxError as e:
+            self.trace_log.append(f"🌳 AST parse failed (line {e.lineno}) — code contains non-Python syntax, using regex fallback")
+            return [], self.trace_log
+
+        visitor = CUDANodeVisitor()
+        visitor.visit(tree)
+
+        # Aggregate findings
+        self.ast_findings = visitor.findings.copy()
+
+        # Report device variables
+        for var, line in visitor.device_vars.items():
+            self.ast_findings.append({
+                "line": line,
+                "type": "device_variable",
+                "original": f'{var} = torch.device("cuda:...")',
+                "suggestion": "Device variable tracked — all downstream .to(device) calls are AMD-safe",
+                "severity": "info",
+            })
+
+        # Report embedded kernels
+        for kernel in visitor.string_literals_with_cuda:
+            self.ast_findings.append({
+                "line": kernel["line"],
+                "type": "embedded_kernel",
+                "original": f"Inline CUDA kernel ({kernel['length']} chars)",
+                "suggestion": "Embedded kernel requires hipify transformation of string contents",
+                "severity": "critical",
+            })
+
+        # Report CUDA call sites
+        for call in visitor.cuda_call_sites:
+            self.ast_findings.append({
+                "line": call["line"],
+                "type": call["type"],
+                "original": call["call"],
+                "suggestion": call["note"],
+                "severity": "info",
+            })
+
+        # Report env mutations
+        for env in visitor.env_mutations:
+            self.ast_findings.append({
+                "line": env["line"],
+                "type": "env_mutation",
+                "original": f'os.environ["{env["var"]}"]',
+                "suggestion": f'Needs migration to ROCm equivalent env var',
+                "severity": "warning",
+            })
+
+        # Summary
+        self.trace_log.append(
+            f"🌳 AST Analysis: {len(visitor.device_vars)} device vars, "
+            f"{len(visitor.cuda_call_sites)} CUDA calls, "
+            f"{len(visitor.string_literals_with_cuda)} embedded kernels, "
+            f"{len(visitor.env_mutations)} env mutations"
+        )
+
+        return self.ast_findings, self.trace_log

api.py

@@ -71,6 +71,7 @@ async def migrate_code(request: MigrationRequest):
                     } for p in result.analysis.hardware_issues
                 ],
                 "implicit_assumptions": result.analysis.implicit_assumptions,
+                "ast_findings": result.analysis.ast_findings,
                 "saliency_map": result.analysis.saliency_map,
                 "detected_patterns": [
                     {

benchmark/rocm_benchmark.py

@@ -0,0 +1,246 @@
+"""
+ROCm Forge — MI300X Benchmark Suite
+Ready to run on AMD Developer Cloud when credits arrive.
+
+Usage:
+    python benchmark/rocm_benchmark.py --all
+    python benchmark/rocm_benchmark.py --device-info
+    python benchmark/rocm_benchmark.py --memory
+    python benchmark/rocm_benchmark.py --compute
+"""
+import argparse
+import time
+import json
+import os
+import sys
+from datetime import datetime
+
+
+def check_rocm_available():
+    """Check if ROCm/HIP runtime is available."""
+    try:
+        import torch
+        if not torch.cuda.is_available():
+            print("❌ No GPU detected. ROCm/HIP runtime not available.")
+            print("   Make sure you're running on an AMD GPU instance with ROCm installed.")
+            return False
+        device_name = torch.cuda.get_device_name(0)
+        print(f"✅ GPU detected: {device_name}")
+        print(f"   PyTorch version: {torch.__version__}")
+        hip_version = getattr(torch.version, 'hip', None)
+        cuda_version = getattr(torch.version, 'cuda', None)
+        if hip_version:
+            print(f"   HIP version: {hip_version}")
+        elif cuda_version:
+            print(f"   CUDA version: {cuda_version}")
+        return True
+    except Exception as e:
+        print(f"❌ Error checking GPU: {e}")
+        return False
+
+
+def device_info_benchmark():
+    """Collect detailed device information for the hackathon submission."""
+    import torch
+
+    results = {
+        "timestamp": datetime.now().isoformat(),
+        "pytorch_version": torch.__version__,
+        "hip_version": getattr(torch.version, 'hip', 'N/A'),
+        "gpu_count": torch.cuda.device_count(),
+        "devices": [],
+    }
+
+    for i in range(torch.cuda.device_count()):
+        props = torch.cuda.get_device_properties(i)
+        device_info = {
+            "index": i,
+            "name": props.name,
+            "total_memory_gb": round(props.total_mem / (1024**3), 2),
+            "multi_processor_count": props.multi_processor_count,
+            "major": props.major,
+            "minor": props.minor,
+        }
+        results["devices"].append(device_info)
+
+    print("\n📊 Device Information")
+    print("=" * 50)
+    print(json.dumps(results, indent=2))
+    return results
+
+
+def memory_benchmark():
+    """Test GPU memory allocation and bandwidth."""
+    import torch
+
+    device = torch.device("cuda:0")
+    results = {"tests": []}
+
+    sizes_mb = [256, 512, 1024, 2048, 4096]
+    print("\n💾 Memory Benchmark")
+    print("=" * 50)
+
+    for size_mb in sizes_mb:
+        n_elements = (size_mb * 1024 * 1024) // 4  # float32
+        torch.cuda.synchronize()
+
+        # Allocation
+        start = time.perf_counter()
+        tensor = torch.zeros(n_elements, dtype=torch.float32, device=device)
+        torch.cuda.synchronize()
+        alloc_time = (time.perf_counter() - start) * 1000
+
+        # Fill (bandwidth test)
+        start = time.perf_counter()
+        tensor.fill_(1.0)
+        torch.cuda.synchronize()
+        fill_time = (time.perf_counter() - start) * 1000
+        bandwidth_gbps = (size_mb / 1024) / (fill_time / 1000) if fill_time > 0 else 0
+
+        # Copy D2D
+        start = time.perf_counter()
+        tensor2 = tensor.clone()
+        torch.cuda.synchronize()
+        copy_time = (time.perf_counter() - start) * 1000
+
+        result = {
+            "size_mb": size_mb,
+            "alloc_ms": round(alloc_time, 2),
+            "fill_ms": round(fill_time, 2),
+            "bandwidth_gbps": round(bandwidth_gbps, 2),
+            "copy_d2d_ms": round(copy_time, 2),
+        }
+        results["tests"].append(result)
+        print(f"  {size_mb:>5} MB → alloc: {alloc_time:6.2f}ms  fill: {fill_time:6.2f}ms  bandwidth: {bandwidth_gbps:7.2f} GB/s  D2D: {copy_time:6.2f}ms")
+
+        del tensor, tensor2
+        torch.cuda.empty_cache()
+
+    return results
+
+
+def compute_benchmark():
+    """Test raw compute performance with matrix operations."""
+    import torch
+
+    device = torch.device("cuda:0")
+    results = {"tests": []}
+
+    sizes = [1024, 2048, 4096, 8192]
+    print("\n⚡ Compute Benchmark (GEMM)")
+    print("=" * 50)
+
+    for n in sizes:
+        a = torch.randn(n, n, device=device, dtype=torch.float32)
+        b = torch.randn(n, n, device=device, dtype=torch.float32)
+
+        # Warmup
+        for _ in range(3):
+            torch.mm(a, b)
+        torch.cuda.synchronize()
+
+        # Benchmark
+        num_runs = 10
+        start = time.perf_counter()
+        for _ in range(num_runs):
+            torch.mm(a, b)
+        torch.cuda.synchronize()
+        elapsed = (time.perf_counter() - start) / num_runs * 1000
+
+        # TFLOPS = 2 * N^3 / time
+        tflops = (2 * n**3) / (elapsed / 1000) / 1e12
+
+        result = {
+            "matrix_size": n,
+            "avg_ms": round(elapsed, 2),
+            "tflops": round(tflops, 2),
+        }
+        results["tests"].append(result)
+        print(f"  {n:>5}x{n} → {elapsed:8.2f} ms  ({tflops:6.2f} TFLOPS)")
+
+        del a, b
+        torch.cuda.empty_cache()
+
+    # FP16 test
+    print("\n⚡ Compute Benchmark (GEMM FP16)")
+    print("=" * 50)
+    for n in [4096, 8192]:
+        a = torch.randn(n, n, device=device, dtype=torch.float16)
+        b = torch.randn(n, n, device=device, dtype=torch.float16)
+
+        for _ in range(3):
+            torch.mm(a, b)
+        torch.cuda.synchronize()
+
+        num_runs = 10
+        start = time.perf_counter()
+        for _ in range(num_runs):
+            torch.mm(a, b)
+        torch.cuda.synchronize()
+        elapsed = (time.perf_counter() - start) / num_runs * 1000
+        tflops = (2 * n**3) / (elapsed / 1000) / 1e12
+
+        result = {
+            "matrix_size": f"{n}_fp16",
+            "avg_ms": round(elapsed, 2),
+            "tflops": round(tflops, 2),
+        }
+        results["tests"].append(result)
+        print(f"  {n:>5}x{n} FP16 → {elapsed:8.2f} ms  ({tflops:6.2f} TFLOPS)")
+
+        del a, b
+        torch.cuda.empty_cache()
+
+    return results
+
+
+def run_all_benchmarks():
+    """Run all benchmarks and save results."""
+    print("🔥 ROCm Forge — MI300X Benchmark Suite")
+    print("=" * 50)
+
+    if not check_rocm_available():
+        print("\n⚠️  Run this on AMD Developer Cloud with MI300X GPU.")
+        print("   python benchmark/rocm_benchmark.py --all")
+        return
+
+    results = {
+        "benchmark_version": "1.0",
+        "timestamp": datetime.now().isoformat(),
+        "device_info": device_info_benchmark(),
+        "memory": memory_benchmark(),
+        "compute": compute_benchmark(),
+    }
+
+    # Save results
+    os.makedirs("benchmark/results", exist_ok=True)
+    output_file = f"benchmark/results/benchmark_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
+    with open(output_file, "w") as f:
+        json.dump(results, f, indent=2)
+
+    print(f"\n✅ Results saved to {output_file}")
+    print("📸 Use these results as proof of AMD GPU usage in your hackathon submission!")
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(description="ROCm Forge MI300X Benchmark Suite")
+    parser.add_argument("--all", action="store_true", help="Run all benchmarks")
+    parser.add_argument("--device-info", action="store_true", help="Show device information")
+    parser.add_argument("--memory", action="store_true", help="Run memory benchmarks")
+    parser.add_argument("--compute", action="store_true", help="Run compute benchmarks")
+    args = parser.parse_args()
+
+    if args.all:
+        run_all_benchmarks()
+    elif args.device_info:
+        if check_rocm_available():
+            device_info_benchmark()
+    elif args.memory:
+        if check_rocm_available():
+            memory_benchmark()
+    elif args.compute:
+        if check_rocm_available():
+            compute_benchmark()
+    else:
+        parser.print_help()
+        print("\n💡 Quick start: python benchmark/rocm_benchmark.py --all")