File size: 6,207 Bytes
9e551a1 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 | # Tensor Roll — Recursive CUDA-Q Model Quantizer
# Copyright (C) 2026 SnapKitty Collective
# SPDX-License-Identifier: AGPL-3.0-or-later
"""Process/IR boundary between the CUDA-Q kernels and the Q# operations.
There is no direct CUDA-Q <-> Q# linkage: the two toolchains do not share
an in-process API. The real boundary used by tensor-roll is a serialized,
checksummed, versioned artifact containing:
1. a gate-level IR trace (S-expression-ish JSON list) describing exactly
which kernel ran, on how many qubits, with which gates, and
2. the measurement counts produced by executing that trace.
Either side of the boundary — the CUDA-Q process
(``cudaq/tensor_roll_kernels.py``), the Q# toolchain (``qsharp/TensorRoll.qs``),
or the classical ``qsim-classical`` simulator — can write this artifact and
any other side can read it. Nothing is invented: the file is real JSON on
disk with a SHA-256 checksum, and the reader refuses corrupted artifacts.
True QIR/MLIR bitcode exchange requires both vendor toolchains installed;
when they are present, the same trace maps 1:1 onto QIR gate ops
(``__quantum__qis__ry``, ``__quantum__qis__cnot__ctl``, ...).
"""
from __future__ import annotations
import hashlib
import json
import os
import time
from typing import Any, Dict, List, Optional
BOUNDARY_VERSION = 1
TRACE_SCHEMA = "tensor-roll/boundary-trace"
# ---------------------------------------------------------------------------
# Gate IR construction (mirrors both kernel families' semantics)
# ---------------------------------------------------------------------------
def _ry(q: int, theta: float) -> Dict[str, Any]:
return {"gate": "ry", "qubits": [q], "params": [float(theta)],
"qir": "__quantum__qis__ry__body"}
def _cnot(c: int, t: int) -> Dict[str, Any]:
return {"gate": "cnot", "qubits": [c, t], "params": [],
"qir": "__quantum__qis__cnot__ctl"}
def _measure(q: int) -> Dict[str, Any]:
return {"gate": "measure", "qubits": [q], "params": [],
"qir": "__quantum__qis__mz__body"}
def _reset(q: int) -> Dict[str, Any]:
return {"gate": "reset", "qubits": [q], "params": [],
"qir": "__quantum__qis__reset__body"}
def trace_for_kernel(kernel: str, n_qubits: int,
angles: Optional[List[float]] = None,
depth: int = 0) -> List[Dict[str, Any]]:
"""Build the gate IR trace for a tensor_roll kernel.
This is the exact gate sequence the CUDA-Q kernels in
cudaq/tensor_roll_kernels.py apply and the Q# operations in
qsharp/TensorRoll.qs apply — a single source of truth for the
cross-toolchain boundary.
"""
a = list(angles) if angles else [0.0] * n_qubits
gates: List[Dict[str, Any]] = []
if kernel == "tensor_roll_encode":
for q in range(n_qubits):
gates.append(_ry(q, 2.0 * a[q % len(a)]))
elif kernel == "tensor_roll_rotate":
for q in range(n_qubits):
gates.append(_ry(q, a[q % len(a)]))
elif kernel == "tensor_roll_entangle":
for q in range(n_qubits - 1):
gates.append(_cnot(q, q + 1))
elif kernel in ("tensor_roll_measure", "tensor_roll_reconstruct"):
for q in range(n_qubits):
gates.append(_measure(q))
elif kernel == "tensor_roll_recursive":
for lvl in range(max(1, depth)):
k = max(1, n_qubits - lvl)
for q in range(k):
gates.append(_ry(q, a[q % len(a)]))
for q in range(k - 1):
gates.append(_cnot(q, q + 1))
if lvl < max(1, depth) - 1:
gates.append(_measure(k - 1))
gates.append(_reset(k - 1))
for q in range(n_qubits):
gates.append(_measure(q))
else:
raise ValueError(f"unknown tensor_roll kernel: {kernel}")
return gates
# ---------------------------------------------------------------------------
# Trace artifact: write / read with checksum
# ---------------------------------------------------------------------------
def write_trace(path: str, kernel: str, n_qubits: int, shots: int,
counts: Dict[str, int],
angles: Optional[List[float]] = None,
depth: int = 0,
producer: str = "qsim-classical") -> Dict[str, Any]:
"""Write a boundary trace artifact. Returns the decoded header."""
gates = trace_for_kernel(kernel, n_qubits, angles=angles, depth=depth)
body = {
"schema": TRACE_SCHEMA,
"version": BOUNDARY_VERSION,
"kernel": kernel,
"n_qubits": n_qubits,
"shots": shots,
"gates": gates,
"n_gates": len(gates),
"counts": {k: int(v) for k, v in counts.items()},
"producer": producer,
"produced_at": time.strftime("%Y-%m-%dT%H:%M:%S%z", time.gmtime()),
"qir_mapping": "each gate lists its __quantum__qis__* QIR op name",
}
payload = json.dumps(body, sort_keys=True).encode()
digest = hashlib.sha256(payload).hexdigest()
os.makedirs(os.path.dirname(os.path.abspath(path)), exist_ok=True)
with open(path, "w") as f:
json.dump({"sha256": digest, "trace": body}, f, indent=2)
return body
def read_trace(path: str) -> Dict[str, Any]:
"""Read and checksum-verify a boundary trace artifact."""
with open(path) as f:
doc = json.load(f)
payload = json.dumps(doc["trace"], sort_keys=True).encode()
if hashlib.sha256(payload).hexdigest() != doc["sha256"]:
raise ValueError(f"boundary trace checksum mismatch: {path}")
if doc["trace"].get("schema") != TRACE_SCHEMA:
raise ValueError(f"not a tensor-roll boundary trace: {path}")
return doc["trace"]
def counts_to_amplitudes(trace: Dict[str, Any]) -> List[float]:
"""Classical reconstruction estimator shared across the boundary:
amplitude_i = sqrt(count_i / shots). Classical by construction."""
shots = trace["shots"]
dim = 1 << trace["n_qubits"]
amps = [0.0] * dim
for bits, c in trace["counts"].items():
idx = int(bits, 2) if set(bits) <= {"0", "1"} else 0
if idx < dim and shots > 0:
amps[idx] = (c / shots) ** 0.5
return amps
|