flagos-ai · zhzhcookie · Jun 23, 2026 · Jun 18, 2026 · Jun 18, 2026 · Jun 18, 2026
@@ -16,18 +16,12 @@ def _triton_root() -> str | None:
 def _get_spec_module():
     global _spec_module
     from ._flagtree_backend import FLAGTREE_BACKEND
-    if not FLAGTREE_BACKEND:
-        return None
-    triton_root = _triton_root()
-    if triton_root is None:
-        return None
-    spec_dir = os.path.join(triton_root, "backends", FLAGTREE_BACKEND, "spec")
-    if not os.path.isdir(spec_dir):
-        return None
     if _spec_module is not None:
         return _spec_module
+    if not FLAGTREE_BACKEND:
+        return None
     try:
-        _spec_module = importlib.import_module(f"triton.backends.{FLAGTREE_BACKEND}.spec")
+        _spec_module = importlib.import_module(f"triton.spec.{FLAGTREE_BACKEND}")
     except ImportError:
         return None
     return _spec_module
@@ -45,7 +39,7 @@ def spec_path(path_list: list):
         return
     triton_root = current_path[:idx + len(marker)]
     rel_path = current_path[idx + 1 + len(marker):]
-    backend_path = os.path.join(triton_root, "backends", FLAGTREE_BACKEND, "spec", "triton", rel_path)
+    backend_path = os.path.join(triton_root, "spec", FLAGTREE_BACKEND, rel_path)
     if os.path.isdir(backend_path) and backend_path not in path_list:
         path_list.insert(0, backend_path)
 

@@ -0,0 +1 @@
+from .language import language_extend_globals
@@ -13,7 +13,6 @@
 from typing import Any, Callable, Dict, Optional, Tuple, Type, Union, Iterable, List
 
 import triton.language.extra.cann.extension as extension
-from triton.extension.buffer.language import core as bl
 from triton.extension.buffer.language.builder import setup_unified_builder_with_buffer_builder
 
 from .. import knobs, language
@@ -39,8 +38,8 @@ def check_identifier_legality(name, type):
 WITH_DISPATCH = {}
 
 # Import and register Ascend extension dispatch handlers
-from triton.language.extra.cann.extension.dispatch import ASCEND_WITH_DISPATCH
-from triton.language.extra.cann.extension.builder import setup_unified_builder
+from triton.language.extra.cann.extension.dispatch import ASCEND_WITH_DISPATCH  # noqa: E402
+from triton.language.extra.cann.extension.builder import setup_unified_builder  # noqa: E402
 
 WITH_DISPATCH.update(ASCEND_WITH_DISPATCH)
 
@@ -1206,7 +1205,7 @@ def visit_For(self, node):
             warp_specialize = iterator.warp_specialize
             disable_licm = iterator.disable_licm
             if (IteratorClass is extension.parallel):
-                bind_sub_block = iterator.bind_sub_block
+                bind_sub_block = iterator.bind_sub_block  # noqa: F841
         elif IteratorClass is range:
             # visit iterator arguments
             # note: only `range` iterator is supported now

@@ -8,7 +8,7 @@
 from ..backends.compiler import BaseBackend, GPUTarget
 from .. import __version__, knobs
 from ..runtime.autotuner import OutOfResources
-from ..runtime.cache import get_cache_manager, get_dump_manager, get_override_manager, get_cache_key, triton_key
+from ..runtime.cache import get_cache_manager, get_dump_manager, get_override_manager, get_cache_key
 from ..runtime.driver import driver
 from ..tools.disasm import get_sass
 from .errors import MLIRCompilationError

@@ -259,8 +259,8 @@ def check_subview(src, offsets, sizes, strides):
     length = len(strides)
     src_strides = [1] * length
     if length == 1:
-        if offset[0] % base_byte != 0:
-            raise TypeError(f"all strides should be 1 and the offset value should be 32-bytes aligned.")
+        if offset[0] % base_byte != 0:  # noqa: F821
+            raise TypeError("all strides should be 1 and the offset value should be 32-bytes aligned.")
         return
     for i in range(length - 2, -1, -1):
         src_strides[i] = src_strides[i + 1] * src.shape[i + 1]
@@ -276,7 +276,7 @@ def check_subview(src, offsets, sizes, strides):
     stride_1 = all(s == 1 for s in strides)
     is_unaligned = result_offset % base_byte != 0 or is_unaligned or not stride_1
     if is_unaligned:
-        raise TypeError(f"all strides should be 1 and the offset value should be 32-bytes aligned.")
+        raise TypeError("all strides should be 1 and the offset value should be 32-bytes aligned.")
 
 
 @builtin

@@ -61,7 +61,7 @@ def to_buffer(
     if isinstance(bind_buffer, bl.buffer):
         builder.create_bind_buffer(tensor.handle, bind_buffer.handle)
         return bind_buffer
-    if not (bind_buffer is None):
+    if bind_buffer is not None:
         raise ValueError("bind_buffer must be a buffer or None")
     address_space = tl._unwrap_if_constexpr(address_space)
     addr_space_attr = (address_space.to_ir(builder) if address_space else builder.get_null_attr())

@@ -515,7 +515,7 @@ def minus(self, input: TensorTy) -> TensorTy:
         input_sca_ty = input.type.scalar
         if hasattr(input, 'was_bool_to_int8'):
             if input.type.scalar.is_int8():
-                raise TypeError(f"unexpected type bool")
+                raise TypeError("unexpected type bool")
         if input_sca_ty.is_ptr():
             raise ValueError("wrong type argument to unary minus (" + input_sca_ty.__repr__() + ")")
         _0 = self.tensor(self.builder.get_null_value(input_sca_ty.to_ir(self.builder)), input_sca_ty)
@@ -1091,7 +1091,7 @@ def _load_legacy(self, ptr, mask, other, boundary_check, padding, cache, evictio
                              "pointers or loading a scalar. Because the compiler does not know the boundary; please "
                              "use block pointers (defined by `make_block_ptr`) instead")
 
-        if mask is not None and other is None and care_padding == True:
+        if mask is not None and other is None and care_padding == True:  # noqa: E712
             # Get element type to determine default padding value
             elt_ty = ptr.type.scalar.element_ty
             # Use 0.0 for floating point types, 0 for integer types
@@ -1583,7 +1583,7 @@ def dot(self, lhs: TensorTy, rhs: TensorTy, acc: TensorTy, input_precision: Opti
 
         M = lhs.type.shape[-2]
         N = rhs.type.shape[-1]
-        K = lhs.type.shape[-1]
+        K = lhs.type.shape[-1]  # noqa: F841
         B = lhs.type.shape[0] if lhs_rank == 3 else None
         ret_ty = tl.block_type(ret_scalar_ty, [B, M, N] if B else [M, N])
         if acc is None:
@@ -1633,13 +1633,13 @@ def dot_scaled(self, lhs: TensorTy, lhs_scale: TensorTy, lhs_format: str, rhs: T
         assert lhs.type.is_block() and rhs.type.is_block()
         if is_compile_on_910_95:
             assert lhs.dtype in [tl.float16, tl.bfloat16, tl.uint8, tl.float8e5,
-                                 tl.float8e4nv], f"lhs matrix dtype must be in [bf16, fp16, uint8, e5m2, e4m3]"
+                                 tl.float8e4nv], "lhs matrix dtype must be in [bf16, fp16, uint8, e5m2, e4m3]"
             assert rhs.dtype in [tl.float16, tl.bfloat16, tl.uint8, tl.float8e5,
-                                 tl.float8e4nv], f"rhs matrix dtype must be in [bf16, fp16, uint8, e5m2, e4m3]"
+                                 tl.float8e4nv], "rhs matrix dtype must be in [bf16, fp16, uint8, e5m2, e4m3]"
         else:
-            assert lhs.dtype == tl.bfloat16 or lhs.dtype == tl.float16, f"lhs matrix dtype must be bf16 or fp16"
-            assert rhs.dtype == tl.bfloat16 or lhs.dtype == tl.float16, f"rhs matrix dtype must be bf16 or fp16"
-        assert lhs.dtype == rhs.dtype, f"lhs rhs matrix must get same dtype"
+            assert lhs.dtype == tl.bfloat16 or lhs.dtype == tl.float16, "lhs matrix dtype must be bf16 or fp16"
+            assert rhs.dtype == tl.bfloat16 or lhs.dtype == tl.float16, "rhs matrix dtype must be bf16 or fp16"
+        assert lhs.dtype == rhs.dtype, "lhs rhs matrix must get same dtype"
         #TODO: validate types.
         lhs_rank = len(lhs.shape)
         rhs_rank = len(rhs.shape)
@@ -1657,10 +1657,10 @@ def dot_scaled(self, lhs: TensorTy, lhs_scale: TensorTy, lhs_format: str, rhs: T
         rhs_scale_is_none = rhs_scale is None or (isinstance(rhs_scale, tl.constexpr) and rhs_scale.value is None)
         lhs_scale_is_none = lhs_scale is None or (isinstance(lhs_scale, tl.constexpr) and lhs_scale.value is None)
         assert isinstance(lhs_scale, tl.tensor) and (lhs_scale.dtype == tl.int8 or lhs_scale.dtype
-                                                     == tl.uint8), f"lhs_scale must be int8 or uint8 tensor"
+                                                     == tl.uint8), "lhs_scale must be int8 or uint8 tensor"
         if not rhs_scale_is_none:
             assert isinstance(rhs_scale, tl.tensor) and (rhs_scale.dtype == tl.int8 or rhs_scale.dtype
-                                                         == tl.uint8), f"rhs_scale must be int8 or uint8 tensor"
+                                                         == tl.uint8), "rhs_scale must be int8 or uint8 tensor"
         lhs = self._bitcast_to_fp_type(lhs, lhs_format)
         rhs = self._bitcast_to_fp_type(rhs, rhs_format)
         assert lhs_k_pack or lhs_format == "e2m1", "only mxfp4 inputs can be packed along a dimension different than K"

@@ -534,7 +534,7 @@ def create_gather_out_to_ub(self, src_ptr, index_tensor, index_boundary, dim, sr
         # Convert src_stride, start_offset, end_offset to integers
         src_stride_vals = [self.to_int_val(s) for s in src_stride]
         start_offset_vals = [self.to_int_val(s) for s in start_offset]
-        end_offset_vals = [self.to_int_val(s) for s in end_offset]
+        end_offset_vals = [self.to_int_val(s) for s in end_offset]  # noqa: F841
 
         # Element type
         dtype_tt = src_ptr.get_element_ty()
@@ -604,7 +604,7 @@ def create_scatter_ub_to_out(self, dst_ptr, value_tensor, index_tensor, index_bo
         # Convert dst_stride, start_offset, end_offset to integers
         dst_stride_vals = [self.to_int_val(s) for s in dst_stride]
         start_offset_vals = [self.to_int_val(s) for s in start_offset]
-        end_offset_vals = [self.to_int_val(s) for s in end_offset]
+        end_offset_vals = [self.to_int_val(s) for s in end_offset]  # noqa: F841
 
         # Element type
         dtype_tt = dst_ptr.get_element_ty()
@@ -616,7 +616,7 @@ def create_scatter_ub_to_out(self, dst_ptr, value_tensor, index_tensor, index_bo
         index_rank = len(index_shape)
         total_elements = np.prod(index_shape)
         flat_values = value_tensor.data.flatten()
-        flat_indices = index_tensor.data.flatten()
+        flat_indices = index_tensor.data.flatten()  # noqa: F841
 
         # Generate  coordinates
         all_coords = []
@@ -730,7 +730,7 @@ def get_i64_array_attr(self, val):
 
     def create_annotation_mark(self, ptr_data, hint_name: str, hint_val):
         if hint_name == "overflow_mode":
-            raise ValueError(f"overflow_mode is not supported in interpreter mode, may have accuracy issues")
+            raise ValueError("overflow_mode is not supported in interpreter mode, may have accuracy issues")
         else:
             warnings.warn(f"compile_hint '{hint_name}' is not supported in interpreter mode, just pass it", UserWarning,
                           stacklevel=2)
@@ -6,13 +6,12 @@
 import hashlib
 import json
 from functools import cached_property
-import itertools
 
-from typing import Any, Dict, Tuple, List, Optional, List
+from typing import Dict, Tuple, Optional, List
 
 from .. import knobs
 from .jit import KernelInterface, JITFunction
-from .errors import OutOfResources, PTXASError
+from .errors import OutOfResources
 from .driver import driver
 from .cache import get_cache_manager, triton_key
 from triton._C.libtriton import get_cache_invalidating_env_vars

@@ -30,10 +30,10 @@ def _try_import_ascend():
         from . import ascend_interpreter
         AscendInterpreterBuilder = ascend_interpreter.AscendInterpreterBuilder
         _has_ascend_support = True
-    except ImportError as e:
+    except ImportError:
         _has_ascend_support = False
         AscendInterpreterBuilder = None
-    except Exception as e:
+    except Exception:
         # Catch other exceptions (like circular import) and log them
         _has_ascend_support = False
         AscendInterpreterBuilder = None

@@ -33,27 +33,17 @@
 import ast
 from collections import OrderedDict
 from typing import (
-    Any,
-    Callable,
     Dict,
-    Final,
-    Iterator,
-    List,
     Optional,
-    Tuple,
-    Type,
-    Union,
-    overload,
 )
 
 import triton
-from triton._C import libentryC
 import torch
-import torch_npu
+import torch_npu  # noqa: F401
 
 torch_device_fn = torch.npu
 
-from .code_cache import config_cache_dir
+from .code_cache import config_cache_dir  # noqa: E402
 
 DEVICE_COUNT = torch_device_fn.device_count()
 version = triton.__version__.split(".")

@@ -38,8 +38,8 @@ def check_npu_smi_device():
             output = result.stdout.lower()
             return "ascend910_95" in output or "ascend950" in output or "910_958b" in output
         return False
-    except Exception as e:
-        logger.warning(f"can not use command: npu-smi info")
+    except Exception:
+        logger.warning("can not use command: npu-smi info")
         return False