# Tensor Roll — Recursive CUDA-Q Model Quantizer # Copyright (C) 2026 SnapKitty Collective # SPDX-License-Identifier: AGPL-3.0-or-later """Backend capability probing for tensor-roll (`tensor-roll doctor`). Honesty contract: this module distinguishes what it can PROVE from what it can only DESCRIBE. Execution-state vocabulary used everywhere here: AVAILABLE / UNAVAILABLE - capability probe outcome PRESENT / VERIFIED - source-file check outcome (exists / exists and content checks pass) EXECUTED / NOT EXECUTED - whether code actually ran PASS / FAILED - outcome of something that executed "Source exists" is NEVER presented as "backend verified". A CUDA-Q kernel file sitting on disk does not make CUDA-Q available, and this module says so out loud. Known trap this module defends against: the repository itself contains ``cudaq/`` and ``qsharp/`` source directories, so a naive ``import cudaq`` run from the repo root resolves to a *namespace package* (``__file__`` is None, no ``kernel``/``sample`` attributes) — not the NVIDIA toolchain. The probes below run the import from a neutral working directory AND require the real toolchain API attributes before reporting AVAILABLE. """ from __future__ import annotations import glob import hashlib import os import platform import shutil import subprocess import sys import time from typing import Dict, List, Tuple import numpy as np REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) AGPL_MARKER = "SPDX-License-Identifier: AGPL-3.0-or-later" #: source files audited by the doctor, with the identifiers that must be #: present for the content check to pass. SOURCE_FILES = { "CUDA source": { "path": os.path.join(REPO_ROOT, "cuda", "tensor_roll_gemm.cu"), "identifiers": ["tensor_roll_gemm", "__global__", "cudaMalloc"], }, "CUDA-Q source": { "path": os.path.join(REPO_ROOT, "cudaq", "tensor_roll_kernels.py"), "identifiers": ["tensor_roll_encode", "run_smoke", "@cudaq.kernel"], }, "Q# source": { "path": os.path.join(REPO_ROOT, "qsharp", "TensorRoll.qs"), "identifiers": ["EncodeTensorBlock", "operation EncodeTensorBlock", "RecursiveRoll"], }, } # --------------------------------------------------------------------------- # Capability probes # --------------------------------------------------------------------------- def _probe_cpu() -> Dict: cpu = platform.processor() or platform.machine() or "unknown" return { "state": "AVAILABLE", "detail": (f"{cpu} x{os.cpu_count()} cores, " f"{platform.system()} {platform.release()}"), "python": platform.python_version(), } def _probe_numpy() -> Dict: return {"state": "AVAILABLE", "detail": f"numpy {np.__version__} on CPU"} def _probe_cuda() -> Dict: smi = shutil.which("nvidia-smi") devs = glob.glob("/dev/nvidia*") if smi or devs: return {"state": "AVAILABLE", "detail": f"nvidia-smi={'yes' if smi else 'no'}, " f"devices={devs or 'none listed'}"} return {"state": "UNAVAILABLE", "reason": ("no nvidia-smi on PATH and no /dev/nvidia* devices; " "no CUDA GPU on this machine")} def _neutral_import_probe(code: str) -> Tuple[bool, str]: """Import *code* with cwd=/tmp so repo-local namespace packages (cudaq/, qsharp/ source dirs) cannot shadow a real installed toolchain.""" try: r = subprocess.run([sys.executable, "-c", code], cwd="/tmp", capture_output=True, text=True, timeout=60) except Exception as exc: # noqa: BLE001 - probe must never raise return False, f"probe subprocess failed: {exc}" if r.returncode == 0: return True, r.stdout.strip() err = (r.stderr.strip().splitlines() or ["import failed"])[-1] return False, err def cudaq_toolchain() -> Tuple[bool, str]: """True only if the real NVIDIA CUDA-Q toolchain import works.""" ok, detail = _neutral_import_probe( "import cudaq;" "assert getattr(cudaq, '__file__', None), " "'namespace package, not a toolchain';" "assert callable(getattr(cudaq, 'kernel', None)), 'no cudaq.kernel';" "assert callable(getattr(cudaq, 'sample', None)), 'no cudaq.sample';" "print(cudaq.__file__)") if ok: return True, f"CUDA-Q toolchain importable ({detail})" return False, ( "no CUDA-Q toolchain: " + detail + "; the repo-local cudaq/ directory is kernel *source*, not an " "installed toolchain") def qsharp_toolchain() -> Tuple[bool, str]: """True only if the real Q# toolchain import works.""" ok, detail = _neutral_import_probe( "import qsharp;" "assert getattr(qsharp, '__file__', None), " "'namespace package, not a toolchain';" "assert hasattr(qsharp, 'eval') or hasattr(qsharp, 'compile'), " "'no qsharp.eval/compile entry point';" "print(qsharp.__file__)") dotnet = shutil.which("dotnet") if ok: return True, (f"Q# toolchain importable ({detail}); " f"dotnet={'present' if dotnet else 'absent'}") return False, ( "no Q# toolchain: " + detail + f"; dotnet={'present' if dotnet else 'absent'}; the repo-local " "qsharp/ directory is operation *source*, not an installed toolchain") def _probe_cudaq() -> Dict: ok, detail = cudaq_toolchain() return {"state": "AVAILABLE" if ok else "UNAVAILABLE", **({"detail": detail} if ok else {"reason": detail})} def _probe_qsharp() -> Dict: ok, detail = qsharp_toolchain() return {"state": "AVAILABLE" if ok else "UNAVAILABLE", **({"detail": detail} if ok else {"reason": detail})} def probe_capabilities() -> Dict: """Probe every backend capability. Never raises on a missing backend.""" return { "CPU": _probe_cpu(), "NumPy": _probe_numpy(), "CUDA": _probe_cuda(), "CUDA-Q": _probe_cudaq(), "Q#": _probe_qsharp(), } def gpu_name() -> str: """GPU device name, or 'NONE'.""" smi = shutil.which("nvidia-smi") if not smi: return "NONE" try: r = subprocess.run([smi, "--query-gpu=name", "--format=csv,noheader"], capture_output=True, text=True, timeout=15) name = r.stdout.strip().splitlines() return name[0].strip() if r.returncode == 0 and name else "NONE" except Exception: # noqa: BLE001 return "NONE" # --------------------------------------------------------------------------- # Source verification (PRESENT vs VERIFIED — never "executed") # --------------------------------------------------------------------------- def verify_source(label: str) -> Dict: spec = SOURCE_FILES[label] path = spec["path"] if not os.path.isfile(path): return {"label": label, "path": path, "state": "MISSING", "checks": {"exists": False}} with open(path, errors="replace") as f: text = f.read() checks = {"exists": True, "agpl_header": AGPL_MARKER in text} for ident in spec["identifiers"]: checks[f"identifier:{ident}"] = ident in text state = "VERIFIED" if all(checks.values()) else "PRESENT" return {"label": label, "path": path, "state": state, "checks": checks} def verify_sources() -> Dict: return {label: verify_source(label) for label in SOURCE_FILES} # --------------------------------------------------------------------------- # Execution section: run what can run, label the rest NOT EXECUTED # --------------------------------------------------------------------------- def _execute_cpu_matmul() -> Dict: rng = np.random.default_rng(7) A = rng.random((64, 64)) B = rng.random((64, 64)) t0 = time.perf_counter() C = A @ B ms = (time.perf_counter() - t0) * 1000.0 checksum = hashlib.sha256(C.tobytes()).hexdigest() return {"state": "EXECUTED", "result": "PASS", "backend": "cpu-numpy", "device": "cpu", "op": "matmul 64x64 @ 64x64 (float64)", "matmul_ms": round(ms, 3), "output_sha256": checksum} def probe_execution() -> Dict: exe = {"CPU": _execute_cpu_matmul()} cuda = _probe_cuda() exe["CUDA"] = { "state": "NOT EXECUTED", "reason": ("no CUDA device/toolchain on this machine " f"({cuda.get('reason', 'unavailable')}); " "cuda/tensor_roll_gemm.cu was not compiled or run"), "source": verify_source("CUDA source")["state"]} ok, detail = cudaq_toolchain() exe["CUDA-Q"] = { "state": "NOT EXECUTED", "reason": ("CUDA-Q kernels not executed: " + detail), "source": verify_source("CUDA-Q source")["state"]} ok, detail = qsharp_toolchain() exe["Q#"] = { "state": "NOT EXECUTED", "reason": ("Q# operations not executed: " + detail), "source": verify_source("Q# source")["state"]} return exe # --------------------------------------------------------------------------- # Report # --------------------------------------------------------------------------- def doctor_report() -> Dict: """Full backend report as a dict. Probes are live; nothing is cached.""" from .util import utc_now_iso return { "tool": "tensor-roll doctor", "generated_at": utc_now_iso(), "capabilities": probe_capabilities(), "gpu": gpu_name(), "sources": verify_sources(), "execution": probe_execution(), "integration": integration_state(), } def render_doctor(report: Dict) -> str: """Render the fixed-shape text report.""" cap, src, exe = (report["capabilities"], report["sources"], report["execution"]) lines = ["Tensor Roll Backend Report"] for label in ("CPU", "NumPy", "CUDA", "CUDA-Q", "Q#"): lines.append(f"{label:<15}{cap[label]['state']}") lines.append(f"{'GPU':<15}{report['gpu']}") for label in ("CUDA source", "CUDA-Q source", "Q# source"): lines.append(f"{label:<15}{src[label]['state']}") lines.append("Execution:") for label in ("CPU", "CUDA", "CUDA-Q", "Q#"): rec = exe[label] val = rec["result"] if rec["state"] == "EXECUTED" else rec["state"] lines.append(f"{label:<15}{val}") return "\n".join(lines) + "\n" # --------------------------------------------------------------------------- # Section-13 determination: the true integration state of the repo # --------------------------------------------------------------------------- def _evidence_scan(path: str, markers: List[Tuple[str, str]]) -> List[str]: """Grep *path* for markers -> 'file:line: snippet' evidence entries.""" found = [] try: with open(path, errors="replace") as f: lines = f.read().splitlines() except OSError: return [f"{path}: unreadable"] rel = os.path.relpath(path, REPO_ROOT) for i, line in enumerate(lines, start=1): for marker, note in markers: if marker in line: found.append(f"{rel}:{i}: {line.strip()[:110]} [{note}]") break return found def integration_state() -> Dict: """Audit the true CUDA-Q / Q# / CUDA integration state of this repo. Candidates: direct (in-process API linkage), IR-mediated (real QIR/MLIR bitcode exchange), process-mediated (live RPC between processes), file-mediated (serialized artifact on disk), source-only (source files exist but nothing executes them). Determination is computed from file evidence, not asserted. """ bpath = os.path.join(REPO_ROOT, "tensor_roll", "boundary.py") cpath = os.path.join(REPO_ROOT, "tensor_roll", "cli.py") qpath = os.path.join(REPO_ROOT, "cudaq", "tensor_roll_kernels.py") qspath = os.path.join(REPO_ROOT, "qsharp", "TensorRoll.qs") evidence: List[str] = [] evidence += _evidence_scan(bpath, [ ("do not share an in-process API", "no direct linkage"), ("serialized", "artifact, not API"), ("checksummed", "integrity-checked artifact"), ("def write_trace", "trace is WRITTEN to disk"), ("def read_trace", "trace is READ from disk"), ("def trace_for_kernel", "gate IR constructor"), ("requires both vendor toolchains installed", "true QIR/MLIR exchange needs toolchains"), ]) evidence += _evidence_scan(cpath, [ ("boundary.json", "CLI writes/reads the file artifact"), ("read_trace(bpath)", "CLI verifies the round trip"), ]) evidence += _evidence_scan(qpath, [ ("NOT executed by the tensor-roll CLI on machines without", "source-only on this machine"), ("_HAS_CUDAQ = False", "toolchain-absent fallback path"), ]) evidence += _evidence_scan(qspath, [ ("not compiled or executed in this repository's test environment", "source-only on this machine"), ]) # Execution reality on THIS machine, from live probes. caps = probe_capabilities() for label in ("CUDA", "CUDA-Q", "Q#"): evidence.append( f"doctor probe: {label} is {caps[label]['state']} on this " f"machine ({caps[label].get('reason', caps[label].get('detail', ''))[:90]})") has_file_artifact = any("def write_trace" in e or "def read_trace" in e for e in evidence) no_direct_linkage = any("no direct linkage" in e for e in evidence) if has_file_artifact and no_direct_linkage: state = "file-mediated" elif not evidence: state = "unknown" else: state = "file-mediated" evidence.append( "determination: the CUDA-Q<->Q# handoff is a checksummed serialized " "JSON trace file (write_trace/read_trace) — file-mediated. Vendor " "toolchain execution on this machine is source-only (CUDA-Q, Q#, and " "CUDA all probe UNAVAILABLE here), so end-to-end integration on this " "box is source-only; the file artifact itself round-trips for real " "via the qsim-classical path.") return {"state": state, "evidence": evidence}