fix(ci): resolve ruff lint errors blocking tests workflow

Fixes F401 unused imports and F821 undefined names in:

- backend.py: remove unused UnionFindDecoder import, add cast() for mypy

- license.py: remove unused ed25519 module import, narrow to Ed25519PublicKey

- stripe_integration.py: define STRIPE_WEBHOOK_SECRET, fix typing

This unblocks the tests / ruff-and-mypy CI job which was failing after 8s.

Author: qectorlab

generate_license_keys.py

@@ -63,6 +63,14 @@ def create_license_token(
     if private_key_pem is None:
         private_key_pem = _load_production_private_key_pem()
     priv_key = serialization.load_pem_private_key(private_key_pem, password=None)
+    if not isinstance(priv_key, ed25519.Ed25519PrivateKey):
+        # load_pem_private_key() returns a union of every key type cryptography
+        # supports (RSA, DSA, EC, X25519, ML-DSA, ML-KEM, ...). This project only
+        # ever signs with Ed25519, so fail loudly if anything else is provided.
+        raise TypeError(
+            "QECTOR license tokens require an Ed25519 private key; "
+            f"got {type(priv_key).__name__}."
+        )
     email_clean = customer_email.strip().lower()
     payload = f"{receipt_id}:{email_clean}".encode("utf-8")
     sig = priv_key.sign(payload)

python/qector_decoder_v3/__init__.py

@@ -138,11 +138,7 @@ def _emit_startup_notice() -> None:
     if any(var in _os_mod.environ for var in ci_vars):
         return
 
-    is_interactive = (
-        hasattr(_sys_mod, "ps1")
-        or "ipykernel" in _sys_mod.modules
-        or "IPython" in _sys_mod.modules
-    )
+    is_interactive = hasattr(_sys_mod, "ps1") or "ipykernel" in _sys_mod.modules or "IPython" in _sys_mod.modules
 
     if is_interactive:
         _sys_mod.stderr.write(

python/qector_decoder_v3/backend.py

@@ -1,7 +1,7 @@
 """
 qector_decoder_v3.backend — Workload-Aware AutoDecoder with Self-Auto-Debug & Multi-Tier Fallback.
 
-Provides automatic hardware routing, real-time performance calibration, and a 
+Provides automatic hardware routing, real-time performance calibration, and a
 7-tier fault-tolerant self-debugging fallback engine for quantum error correction decoding.
 """
 
@@ -20,7 +20,6 @@
     CPUBatchDecoder,
     CUDABatchDecoder,
     FastUnionFindDecoder,
-    UnionFindDecoder,
     BlossomDecoder,
     SparseBlossomDecoder,
     LookupTableDecoder,
@@ -128,7 +127,7 @@ def __init__(self, check_to_qubits, n_qubits=None, config: Optional[BackendConfi
         self._cuda_ok = bool(self.config.allow_gpu and cuda_is_available())
         opencl_auto = os.environ.get("QECTOR_ENABLE_OPENCL_AUTO", "").lower() in {"1", "true", "yes", "on"}
         self._opencl_ok = bool(self.config.allow_gpu and opencl_auto and opencl_is_available())
-        
+
         if self.config.allow_gpu and not opencl_auto and opencl_is_available():
             self._diag.warnings.append("OpenCL auto-routing disabled; set QECTOR_ENABLE_OPENCL_AUTO=1 to enable it")
 
@@ -173,7 +172,11 @@ def _get_decoder(self, backend: str) -> Any:
         if backend in self._decoders:
             return self._decoders[backend]
 
-        dec = None
+        # `dec` holds a heterogeneous set of decoder types (one per backend
+        # branch below). Typing it as Any avoids mypy flagging every elif
+        # branch as a type-incompatible assignment while remaining accurate
+        # at runtime.
+        dec: Any = None
         if backend == Backend.CPU_SINGLE:
             dec = FastUnionFindDecoder(self._c2q, self._nq)
         elif backend == Backend.CPU_RAYON:
@@ -211,16 +214,16 @@ def _get_decoder(self, backend: str) -> Any:
         return dec
 
     def _get_cpu_single(self) -> FastUnionFindDecoder:
-        return self._get_decoder(Backend.CPU_SINGLE)
+        return cast(FastUnionFindDecoder, self._get_decoder(Backend.CPU_SINGLE))
 
     def _get_cpu_rayon(self) -> BatchDecoder:
-        return self._get_decoder(Backend.CPU_RAYON)
+        return cast(BatchDecoder, self._get_decoder(Backend.CPU_RAYON))
 
     def _get_cuda(self) -> Optional[CUDABatchDecoder]:
-        return self._get_decoder(Backend.CUDA)
+        return cast(Optional[CUDABatchDecoder], self._get_decoder(Backend.CUDA))
 
     def _get_opencl(self) -> Optional[OpenCLBatchDecoder]:
-        return self._get_decoder(Backend.OPENCL)
+        return cast(Optional[OpenCLBatchDecoder], self._get_decoder(Backend.OPENCL))
 
     # -- selection ---------------------------------------------------------
     def select(self, batch_size: int) -> str:
@@ -229,7 +232,11 @@ def select(self, batch_size: int) -> str:
             return self.config.force
 
         if self.config.allow_gpu and batch_size >= self.config.gpu_threshold:
-            if self.config.prefer == Backend.OPENCL and self._opencl_ok and self._diag.backend_health.get(Backend.OPENCL, True):
+            if (
+                self.config.prefer == Backend.OPENCL
+                and self._opencl_ok
+                and self._diag.backend_health.get(Backend.OPENCL, True)
+            ):
                 return Backend.OPENCL
             if self._cuda_ok and self._diag.backend_health.get(Backend.CUDA, True):
                 return Backend.CUDA
@@ -283,7 +290,14 @@ def batch_decode(self, syndromes) -> np.ndarray:
 
         # Robust multi-tier self-debugging execution chain
         fallback_chain = [chosen]
-        for t in [Backend.CUDA, Backend.OPENCL, Backend.CPU_RAYON, Backend.CPU_BATCH, Backend.CPU_SINGLE, Backend.BLOSSOM]:
+        for t in [
+            Backend.CUDA,
+            Backend.OPENCL,
+            Backend.CPU_RAYON,
+            Backend.CPU_BATCH,
+            Backend.CPU_SINGLE,
+            Backend.BLOSSOM,
+        ]:
             if t not in fallback_chain:
                 fallback_chain.append(t)
 
@@ -362,12 +376,14 @@ def _record_auto_debug_failure(self, backend: str, exc: Exception, context: str)
         self._diag.backend_health[backend] = False
         msg = f"AutoDebug caught error on {backend} [{context}]: {exc}"
         self._diag.warnings.append(msg)
-        self._diag.debug_log.append({
-            "timestamp": time.time(),
-            "backend": backend,
-            "context": context,
-            "error": str(exc),
-        })
+        self._diag.debug_log.append(
+            {
+                "timestamp": time.time(),
+                "backend": backend,
+                "context": context,
+                "error": str(exc),
+            }
+        )
         logger.warning(msg)
 
     def _python_fallback_decode(self, syndrome: np.ndarray) -> np.ndarray:

python/qector_decoder_v3/decoder_cache.py

@@ -93,6 +93,7 @@ def _build_decoder_pool(
     n_workers: Optional[int],
 ):
     from .decoder_pool import DecoderPool
+
     checks = [list(c) for c in checks_tuple]
     return DecoderPool(
         checks,

python/qector_decoder_v3/decoder_pool.py

@@ -133,7 +133,6 @@ def __init__(
         self._decoder_type = str(decoder_type)
         self._pool: Optional["_mp.pool.Pool"] = None
 
-
     def decode(self, syndromes) -> np.ndarray:
         """Decode a batch of syndromes.
 

python/qector_decoder_v3/license.py

@@ -1,18 +1,34 @@
 from __future__ import annotations
 import base64
-from cryptography.hazmat.primitives.asymmetric import ed25519
+from typing import Optional
 from cryptography.hazmat.primitives import serialization
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey
 from cryptography.exceptions import InvalidSignature
 
 # Embedded Public Key — Production Ed25519 Key (rotated 2026-07-22)
 PUBLIC_KEY_PEM = b"""-----BEGIN PUBLIC KEY-----
 MCowBQYDK2VwAyEAQh9t19EZ4KWZEYjY3EwHCUzUIehZBlovaMtrpLQXeGA=
 -----END PUBLIC KEY-----"""
 
-try:
-    _PUBLIC_KEY = serialization.load_pem_public_key(PUBLIC_KEY_PEM)
-except Exception:
-    _PUBLIC_KEY = None
+
+def _load_ed25519_public_key() -> Optional[Ed25519PublicKey]:
+    """Loads the embedded PEM and narrows it to Ed25519PublicKey.
+
+    load_pem_public_key() returns a union of every key type cryptography
+    supports (RSA, DSA, EC, X25519, ML-DSA, ML-KEM, ...). This project only
+    ever signs/verifies with Ed25519, so we assert that narrowly here rather
+    than propagating the full union to every call site.
+    """
+    try:
+        key = serialization.load_pem_public_key(PUBLIC_KEY_PEM)
+    except Exception:
+        return None
+    if not isinstance(key, Ed25519PublicKey):
+        return None
+    return key
+
+
+_PUBLIC_KEY: Optional[Ed25519PublicKey] = _load_ed25519_public_key()
 
 
 def verify_license_token(token: str, customer_email: str = "") -> bool:
@@ -38,11 +54,11 @@ def verify_license_token(token: str, customer_email: str = "") -> bool:
             if missing_pad:
                 email_b64 += "=" * (4 - missing_pad)
             embedded_email = base64.urlsafe_b64decode(email_b64).decode("utf-8").lower()
-            
+
             # If caller provided explicit email check, ensure match
             if customer_email and customer_email.strip().lower() != embedded_email:
                 return False
-            
+
             target_email = embedded_email
         except Exception:
             return False

python/qector_decoder_v3/predecoder.py

@@ -102,14 +102,9 @@ def __init__(self, check_to_qubits, n_qubits=None, backend: str = "blossom"):
         if backend in _backend_map:
             backend = _backend_map[backend]
         if backend not in _valid_backends:
-            raise ValueError(
-                "backend must be one of "
-                "['blossom', 'union_find', 'sparse_blossom', 'fast_union_find']"
-            )
+            raise ValueError("backend must be one of ['blossom', 'union_find', 'sparse_blossom', 'fast_union_find']")
         self.backend = backend
-        residual: Union[
-            BlossomDecoder, UnionFindDecoder, SparseBlossomDecoder, FastUnionFindDecoder
-        ]
+        residual: Union[BlossomDecoder, UnionFindDecoder, SparseBlossomDecoder, FastUnionFindDecoder]
         if backend == "blossom":
             residual = BlossomDecoder(self._c2q, self.n_qubits)
         elif backend == "union_find":
@@ -118,9 +113,7 @@ def __init__(self, check_to_qubits, n_qubits=None, backend: str = "blossom"):
             residual = FastUnionFindDecoder(self._c2q, self.n_qubits)
         else:
             residual = SparseBlossomDecoder(self._c2q, self.n_qubits)
-        self._residual: Union[
-            BlossomDecoder, UnionFindDecoder, SparseBlossomDecoder, FastUnionFindDecoder
-        ] = residual
+        self._residual: Union[BlossomDecoder, UnionFindDecoder, SparseBlossomDecoder, FastUnionFindDecoder] = residual
         self.last_predecoded = 0  # number of defects resolved by the predecoder
 
     def _predecode(self, syndrome: np.ndarray) -> Tuple[np.ndarray, np.ndarray]:

python/qector_decoder_v3/stripe_integration.py

@@ -6,11 +6,12 @@
 import os
 import json
 import logging
-from typing import Dict, Any, Optional
+from typing import Any, Dict, Optional
 from pathlib import Path
 
 try:
     from dotenv import load_dotenv
+
     # Load environment variables from .env at repo root or current working dir
     env_path = Path(__file__).resolve().parents[2] / ".env"
     if env_path.exists():
@@ -26,18 +27,21 @@
 logger = logging.getLogger("qector_decoder_v3.stripe")
 
 # Retrieve keys from environment
-STRIPE_SECRET_KEY: str = os.getenv("STRIPE_SECRET_KEY") or os.getenv("STRIPE_SECRET", "")
-STRIPE_PUBLISHABLE_KEY: str = os.getenv("STRIPE_PUBLISHABLE_KEY") or os.getenv("STRIPE_PUBLISHABLE", "")
+# Use the default-argument form of os.getenv so the static type is `str` rather
+# than `str | None`; the `or os.getenv(..., "")` pattern confuses mypy.
+STRIPE_SECRET_KEY: str = os.getenv("STRIPE_SECRET_KEY", "") or os.getenv("STRIPE_SECRET", "")
+STRIPE_PUBLISHABLE_KEY: str = os.getenv("STRIPE_PUBLISHABLE_KEY", "") or os.getenv("STRIPE_PUBLISHABLE", "")
+STRIPE_WEBHOOK_SECRET: str = os.getenv("STRIPE_WEBHOOK_SECRET", "")
 STRIPE_PURCHASE_LINK: str = os.getenv(
     "STRIPE_PURCHASE_LINK",
-    "https://buy.stripe.com/7sY9AVdwlgoyfse9bYeUU00?locale=en&__embed_source=buy_btn_1TsoKxRsa9cg9l8A7ExMmc77"
+    "https://buy.stripe.com/7sY9AVdwlgoyfse9bYeUU00?locale=en&__embed_source=buy_btn_1TsoKxRsa9cg9l8A7ExMmc77",
 )
 
 if STRIPE_SECRET_KEY:
     stripe.api_key = STRIPE_SECRET_KEY
 
 
-def get_stripe_keys() -> Dict[str, str]:
+def get_stripe_keys() -> Dict[str, Any]:
     """Returns configured Stripe keys snapshot."""
     return {
         "publishable_key": STRIPE_PUBLISHABLE_KEY,
@@ -94,9 +98,7 @@ def create_checkout_session(
 
 
 def handle_stripe_webhook_payload(
-    payload: bytes,
-    sig_header: str = "",
-    webhook_secret: Optional[str] = None
+    payload: bytes, sig_header: str = "", webhook_secret: Optional[str] = None
 ) -> Dict[str, Any]:
     """
     Parses and verifies Stripe webhook events, issuing a signed Ed25519 license token
@@ -114,14 +116,13 @@ def handle_stripe_webhook_payload(
         payload_text = payload.decode("utf-8") if isinstance(payload, bytes) else str(payload)
         event = json.loads(payload_text)
 
-
     event_type = event.get("type", "")
     response_data = {"event_type": event_type, "license_token": None, "issued": False}
 
     if event_type in ("checkout.session.completed", "payment_intent.succeeded"):
         session_obj = event.get("data", {}).get("object", {})
         receipt_id = session_obj.get("id") or session_obj.get("payment_intent") or "rec_stripe_live"
-        
+
         customer_details = session_obj.get("customer_details") or {}
         customer_email = (
             session_obj.get("customer_email")
@@ -130,7 +131,7 @@ def handle_stripe_webhook_payload(
         )
 
         token = create_license_token(receipt_id=receipt_id, customer_email=customer_email)
-        
+
         # Save issued license record locally
         _save_issued_license(receipt_id, customer_email, token)
 
@@ -151,9 +152,11 @@ def _save_issued_license(receipt_id: str, email: str, token: str) -> None:
             records = json.loads(record_file.read_text(encoding="utf-8"))
         except Exception:
             records = []
-    records.append({
-        "receipt_id": receipt_id,
-        "customer_email": email,
-        "license_token": token,
-    })
+    records.append(
+        {
+            "receipt_id": receipt_id,
+            "customer_email": email,
+            "license_token": token,
+        }
+    )
     record_file.write_text(json.dumps(records, indent=2), encoding="utf-8")

python/tests/test_auto_debug_fallbacks.py

@@ -14,21 +14,21 @@
 def test_auto_debug_fallback_recovery_on_exception():
     code = codes.rotated_surface_code(5)
     H = code.parity_check_matrix()
-    
+
     cfg = BackendConfig(force=Backend.CUDA, allow_gpu=True, enable_auto_debug=True)
     ad = AutoDecoder(code.check_to_qubits, code.n_qubits, cfg)
-    
+
     # Mock CUDA decoder to raise a simulated hardware/driver exception
     mock_cuda = ad._get_cuda()
     if mock_cuda is not None:
         with patch.object(mock_cuda, "batch_decode", side_effect=RuntimeError("Simulated CUDA OOM Error")):
             syns = (np.random.default_rng(42).random((32, code.n_checks)) < 0.08).astype(np.uint8)
-            
+
             # The auto-debug engine should catch the CUDA exception and gracefully fall back to CPU_RAYON / CPU_SINGLE
             out = ad.batch_decode(syns)
             assert out.shape == (32, code.n_qubits)
             assert all(np.array_equal((H @ out[i]) & 1, syns[i]) for i in range(32))
-            
+
             diag = ad.diagnostics()
             assert diag["backend_health"][Backend.CUDA] is False
             assert len(diag["debug_log"]) >= 1
@@ -39,6 +39,6 @@ def test_reset_backend_health():
     code = codes.repetition_code(7)
     ad = AutoDecoder(code.check_to_qubits, code.n_qubits)
     ad._diag.backend_health[Backend.CUDA] = False
-    
+
     ad.reset_backend_health()
     assert ad._diag.backend_health[Backend.CUDA] is True

python/tests/test_license_included.py

@@ -41,7 +41,8 @@ def test_license_file_exists_and_non_empty():
     # Section 3 — accept any of the well-known phrasings used across revisions.
     assert (
         "commercial use requires a paid license" in text_lower
-        or "commercial use" in text_lower and "license" in text_lower
+        or "commercial use" in text_lower
+        and "license" in text_lower
     )