diff --git a/kt-kernel/python/__init__.py b/kt-kernel/python/__init__.py
index 194f8e4cf..e2b223b8e 100644
--- a/kt-kernel/python/__init__.py
+++ b/kt-kernel/python/__init__.py
@@ -1,96 +1,97 @@
-# KT-Kernel: High-performance kernel operations for KTransformers
-# SPDX-License-Identifier: Apache-2.0
-
-"""
-KT-Kernel provides high-performance kernel operations for KTransformers,
-including CPU-optimized MoE inference with AMX, AVX, and KML support.
-
-The package automatically detects your CPU capabilities and loads the optimal
-kernel variant (AMX, AVX512, or AVX2) at runtime.
-
-Example usage:
-    >>> from kt_kernel import KTMoEWrapper
-    >>> wrapper = KTMoEWrapper(
-    ...     layer_idx=0,
-    ...     num_experts=8,
-    ...     num_experts_per_tok=2,
-    ...     hidden_size=4096,
-    ...     moe_intermediate_size=14336,
-    ...     num_gpu_experts=2,
-    ...     cpuinfer_threads=32,
-    ...     threadpool_count=2,
-    ...     weight_path="/path/to/weights",
-    ...     chunked_prefill_size=512,
-    ...     method="AMXINT4"
-    ... )
-
-    Check which CPU variant is loaded:
-    >>> import kt_kernel
-    >>> print(kt_kernel.__cpu_variant__)  # 'amx', 'avx512', or 'avx2'
-
-Environment Variables:
-    KT_KERNEL_CPU_VARIANT: Override automatic detection ('amx', 'avx512', 'avx2')
-    KT_KERNEL_DEBUG: Enable debug output ('1' to enable)
-"""
-
-from __future__ import annotations
-
-# Detect CPU and load optimal extension variant
-from ._cpu_detect import initialize as _initialize_cpu
-
-_kt_kernel_ext, __cpu_variant__ = _initialize_cpu()
-
-# Make the extension module available to other modules in this package
-import sys
-
-sys.modules["kt_kernel_ext"] = _kt_kernel_ext
-sys.modules[f"{__name__}.kt_kernel_ext"] = _kt_kernel_ext
-
-# Also expose kt_kernel_ext as an attribute for backward compatibility
-kt_kernel_ext = _kt_kernel_ext
-
-# Import main API
-from .experts import KTMoEWrapper
-from .experts_base import generate_gpu_experts_masks
-
-def __getattr__(name):
-    if name == "AMXSFTMoEWrapper":
-        try:
-            from .sft.amx import AMXSFTMoEWrapper
-            return AMXSFTMoEWrapper
-        except (ImportError, AttributeError):
-            return None
-    raise AttributeError(f"module 'kt_kernel' has no attribute {name!r}")
-
-# Read version from package metadata (preferred) or fallback to project root
-try:
-    # Try to get version from installed package metadata (works in installed environment)
-    from importlib.metadata import version, PackageNotFoundError
-
-    try:
-        __version__ = version("kt-kernel")
-    except PackageNotFoundError:
-        # Package not installed, try to read from source tree version.py
-        import os
-
-        _root_version_file = os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(__file__))), "version.py")
-        if os.path.exists(_root_version_file):
-            _version_ns = {}
-            with open(_root_version_file, "r", encoding="utf-8") as f:
-                exec(f.read(), _version_ns)
-            __version__ = _version_ns.get("__version__", "0.6.1")
-        else:
-            __version__ = "0.6.1"
-except ImportError:
-    # Python < 3.8, fallback to pkg_resources or hardcoded version
-    try:
-        from pkg_resources import get_distribution, DistributionNotFound
-
-        try:
-            __version__ = get_distribution("kt-kernel").version
-        except DistributionNotFound:
-            __version__ = "0.6.1"
-    except ImportError:
-        __version__ = "0.6.1"
-
-__all__ = ["KTMoEWrapper", "AMXSFTMoEWrapper", "generate_gpu_experts_masks", "kt_kernel_ext", "__cpu_variant__", "__version__"]
+# KT-Kernel: High-performance kernel operations for KTransformers
+# SPDX-License-Identifier: Apache-2.0
+
+"""
+KT-Kernel provides high-performance kernel operations for KTransformers,
+including CPU-optimized MoE inference with AMX, AVX, and KML support.
+
+The package automatically detects your CPU capabilities and loads the optimal
+kernel variant (AMX, AVX512, or AVX2) at runtime.
+
+Example usage:
+    >>> from kt_kernel import KTMoEWrapper
+    >>> wrapper = KTMoEWrapper(
+    ...     layer_idx=0,
+    ...     num_experts=8,
+    ...     num_experts_per_tok=2,
+    ...     hidden_size=4096,
+    ...     moe_intermediate_size=14336,
+    ...     num_gpu_experts=2,
+    ...     cpuinfer_threads=32,
+    ...     threadpool_count=2,
+    ...     weight_path="/path/to/weights",
+    ...     chunked_prefill_size=512,
+    ...     method="AMXINT4"
+    ... )
+
+    Check which CPU variant is loaded:
+    >>> import kt_kernel
+    >>> print(kt_kernel.__cpu_variant__)  # 'amx', 'avx512', or 'avx2'
+
+Environment Variables:
+    KT_KERNEL_CPU_VARIANT: Override automatic detection ('amx', 'avx512', 'avx2')
+    KT_KERNEL_DEBUG: Enable debug output ('1' to enable)
+"""
+
+from __future__ import annotations
+
+# Detect CPU and load optimal extension variant
+from ._cpu_detect import initialize as _initialize_cpu
+
+_kt_kernel_ext, __cpu_variant__ = _initialize_cpu()
+
+# Make the extension module available to other modules in this package
+import sys
+
+sys.modules["kt_kernel_ext"] = _kt_kernel_ext
+sys.modules[f"{__name__}.kt_kernel_ext"] = _kt_kernel_ext
+
+# Also expose kt_kernel_ext as an attribute for backward compatibility
+kt_kernel_ext = _kt_kernel_ext
+
+# Import main API
+from .experts import KTMoEWrapper
+from .experts_base import generate_gpu_experts_masks
+from .expert_placement import plan_gpu_expert_placement
+
+def __getattr__(name):
+    if name == "AMXSFTMoEWrapper":
+        try:
+            from .sft.amx import AMXSFTMoEWrapper
+            return AMXSFTMoEWrapper
+        except (ImportError, AttributeError):
+            return None
+    raise AttributeError(f"module 'kt_kernel' has no attribute {name!r}")
+
+# Read version from package metadata (preferred) or fallback to project root
+try:
+    # Try to get version from installed package metadata (works in installed environment)
+    from importlib.metadata import version, PackageNotFoundError
+
+    try:
+        __version__ = version("kt-kernel")
+    except PackageNotFoundError:
+        # Package not installed, try to read from source tree version.py
+        import os
+
+        _root_version_file = os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(__file__))), "version.py")
+        if os.path.exists(_root_version_file):
+            _version_ns = {}
+            with open(_root_version_file, "r", encoding="utf-8") as f:
+                exec(f.read(), _version_ns)
+            __version__ = _version_ns.get("__version__", "0.6.1")
+        else:
+            __version__ = "0.6.1"
+except ImportError:
+    # Python < 3.8, fallback to pkg_resources or hardcoded version
+    try:
+        from pkg_resources import get_distribution, DistributionNotFound
+
+        try:
+            __version__ = get_distribution("kt-kernel").version
+        except DistributionNotFound:
+            __version__ = "0.6.1"
+    except ImportError:
+        __version__ = "0.6.1"
+
+__all__ = ["KTMoEWrapper", "AMXSFTMoEWrapper", "generate_gpu_experts_masks", "plan_gpu_expert_placement", "kt_kernel_ext", "__cpu_variant__", "__version__"]
diff --git a/kt-kernel/python/expert_placement.py b/kt-kernel/python/expert_placement.py
new file mode 100644
index 000000000..e5072f911
--- /dev/null
+++ b/kt-kernel/python/expert_placement.py
@@ -0,0 +1,310 @@
+# SPDX-License-Identifier: Apache-2.0
+"""Expert placement planning utilities for heterogeneous MoE inference.
+
+This module contains pure-Python / PyTorch helpers for turning expert
+activation statistics into GPU expert masks.
+
+The default frequency strategy intentionally preserves the existing
+generate_gpu_experts_masks behavior: select the globally hottest experts.
+
+The score_aware_layer_balanced strategy is a small, practical placement policy
+inspired by hybrid CPU-GPU MoE scheduling work: smooth activation scores across
+profiling windows and avoid concentrating all GPU experts in only a few layers.
+"""
+
+from __future__ import annotations
+
+import math
+from typing import Any, Optional
+
+import torch
+
+
+SUPPORTED_EXPERT_PLACEMENT_STRATEGIES = frozenset(
+    {
+        "frequency",
+        "score_aware_layer_balanced",
+    }
+)
+
+
+def plan_gpu_expert_placement(
+    activation_freq: torch.Tensor,
+    num_gpu_experts: int,
+    strategy: str = "frequency",
+    *,
+    min_experts_per_layer: int = 0,
+    max_experts_per_layer: Optional[int] = None,
+    previous_scores: Optional[torch.Tensor] = None,
+    alpha: float = 0.8,
+) -> tuple[torch.Tensor, dict[str, Any]]:
+    """Plan which MoE experts should be placed on GPU.
+
+    Args:
+        activation_freq:
+            Activation frequency or score table with shape
+            ``(num_layers, num_experts_per_layer)``.
+        num_gpu_experts:
+            Total number of experts to place on GPU across all layers.
+        strategy:
+            Placement strategy. ``frequency`` preserves the existing global
+            top-k behavior. ``score_aware_layer_balanced`` uses EMA-smoothed
+            scores and optional per-layer bounds.
+        min_experts_per_layer:
+            Minimum experts to place on GPU for each layer. Only meaningful for
+            ``score_aware_layer_balanced``.
+        max_experts_per_layer:
+            Maximum experts to place on GPU for each layer. Defaults to all
+            experts in a layer.
+        previous_scores:
+            Optional previous activation score table for EMA smoothing.
+        alpha:
+            EMA coefficient for current activation scores when previous_scores
+            is provided: ``alpha * current + (1 - alpha) * previous``.
+
+    Returns:
+        A tuple ``(gpu_experts_mask, report)`` where mask is a CPU bool tensor
+        with the same shape as activation_freq.
+
+    Raises:
+        ValueError:
+            If inputs are malformed or constraints are impossible.
+    """
+    scores = _prepare_scores(
+        activation_freq=activation_freq,
+        previous_scores=previous_scores,
+        alpha=alpha,
+    )
+    num_layers, num_experts_per_layer = scores.shape
+    total_experts = num_layers * num_experts_per_layer
+    budget = _clamp_int(num_gpu_experts, 0, total_experts)
+
+    if strategy not in SUPPORTED_EXPERT_PLACEMENT_STRATEGIES:
+        raise ValueError(
+            f"Unknown expert placement strategy: {strategy!r}. "
+            f"Supported strategies: {sorted(SUPPORTED_EXPERT_PLACEMENT_STRATEGIES)}"
+        )
+
+    if strategy == "frequency":
+        mask = _global_topk_mask(scores, budget)
+    else:
+        mask = _score_aware_layer_balanced_mask(
+            scores=scores,
+            budget=budget,
+            min_experts_per_layer=min_experts_per_layer,
+            max_experts_per_layer=max_experts_per_layer,
+        )
+
+    return mask, _build_report(
+        scores=scores,
+        mask=mask,
+        requested_num_gpu_experts=num_gpu_experts,
+        strategy=strategy,
+        min_experts_per_layer=min_experts_per_layer,
+        max_experts_per_layer=max_experts_per_layer,
+        alpha=alpha,
+        previous_scores_provided=previous_scores is not None,
+    )
+
+
+def _prepare_scores(
+    activation_freq: torch.Tensor,
+    previous_scores: Optional[torch.Tensor],
+    alpha: float,
+) -> torch.Tensor:
+    if not isinstance(activation_freq, torch.Tensor):
+        raise TypeError("activation_freq must be a torch.Tensor")
+
+    if activation_freq.ndim != 2:
+        raise ValueError(
+            f"activation_freq must be a 2D tensor of shape "
+            f"(num_layers, num_experts), got shape {tuple(activation_freq.shape)}"
+        )
+
+    if not 0.0 <= alpha <= 1.0:
+        raise ValueError(f"alpha must be in [0, 1], got {alpha}")
+
+    current = activation_freq.detach().to(device="cpu", dtype=torch.float32)
+
+    if previous_scores is None:
+        return current.contiguous()
+
+    if not isinstance(previous_scores, torch.Tensor):
+        raise TypeError("previous_scores must be a torch.Tensor when provided")
+
+    if previous_scores.shape != activation_freq.shape:
+        raise ValueError(
+            f"previous_scores shape {tuple(previous_scores.shape)} must match "
+            f"activation_freq shape {tuple(activation_freq.shape)}"
+        )
+
+    previous = previous_scores.detach().to(device="cpu", dtype=torch.float32)
+    return (alpha * current + (1.0 - alpha) * previous).contiguous()
+
+
+def _global_topk_mask(scores: torch.Tensor, budget: int) -> torch.Tensor:
+    num_layers, num_experts_per_layer = scores.shape
+    total_experts = num_layers * num_experts_per_layer
+
+    mask = torch.zeros(total_experts, dtype=torch.bool, device="cpu")
+    if budget == 0:
+        return mask.view(num_layers, num_experts_per_layer)
+
+    flat_scores = scores.reshape(-1)
+    _, top_indices = torch.topk(flat_scores, k=budget, largest=True, sorted=False)
+    mask[top_indices] = True
+    return mask.view(num_layers, num_experts_per_layer)
+
+
+def _score_aware_layer_balanced_mask(
+    scores: torch.Tensor,
+    budget: int,
+    min_experts_per_layer: int,
+    max_experts_per_layer: Optional[int],
+) -> torch.Tensor:
+    num_layers, num_experts_per_layer = scores.shape
+
+    if min_experts_per_layer < 0:
+        raise ValueError(
+            f"min_experts_per_layer must be non-negative, got {min_experts_per_layer}"
+        )
+
+    if max_experts_per_layer is None:
+        max_experts_per_layer = num_experts_per_layer
+
+    if max_experts_per_layer < 0:
+        raise ValueError(
+            f"max_experts_per_layer must be non-negative, got {max_experts_per_layer}"
+        )
+
+    if min_experts_per_layer > max_experts_per_layer:
+        raise ValueError(
+            f"min_experts_per_layer ({min_experts_per_layer}) cannot exceed "
+            f"max_experts_per_layer ({max_experts_per_layer})"
+        )
+
+    if max_experts_per_layer > num_experts_per_layer:
+        raise ValueError(
+            f"max_experts_per_layer ({max_experts_per_layer}) cannot exceed "
+            f"num_experts_per_layer ({num_experts_per_layer})"
+        )
+
+    min_required = min_experts_per_layer * num_layers
+    max_allowed = max_experts_per_layer * num_layers
+
+    if budget < min_required:
+        raise ValueError(
+            f"num_gpu_experts={budget} is too small for "
+            f"min_experts_per_layer={min_experts_per_layer} across "
+            f"{num_layers} layers; need at least {min_required}"
+        )
+
+    if budget > max_allowed:
+        raise ValueError(
+            f"num_gpu_experts={budget} exceeds max_experts_per_layer="
+            f"{max_experts_per_layer} across {num_layers} layers; "
+            f"at most {max_allowed} experts can be selected"
+        )
+
+    mask = torch.zeros_like(scores, dtype=torch.bool, device="cpu")
+    per_layer_count = [0 for _ in range(num_layers)]
+
+    # First satisfy the minimum layer coverage constraint.
+    if min_experts_per_layer > 0:
+        for layer_idx in range(num_layers):
+            layer_scores = scores[layer_idx]
+            _, expert_indices = torch.topk(
+                layer_scores,
+                k=min_experts_per_layer,
+                largest=True,
+                sorted=False,
+            )
+            mask[layer_idx, expert_indices] = True
+            per_layer_count[layer_idx] = min_experts_per_layer
+
+    remaining = budget - int(mask.sum().item())
+    if remaining == 0:
+        return mask
+
+    # Deterministic global ordering with layer cap enforcement.
+    # Tie-breaks by layer index then expert index so tests and profiles are stable.
+    candidates: list[tuple[float, int, int]] = []
+    for layer_idx in range(num_layers):
+        for expert_idx in range(num_experts_per_layer):
+            if not bool(mask[layer_idx, expert_idx].item()):
+                candidates.append(
+                    (float(scores[layer_idx, expert_idx].item()), layer_idx, expert_idx)
+                )
+
+    candidates.sort(key=lambda item: (-item[0], item[1], item[2]))
+
+    for _, layer_idx, expert_idx in candidates:
+        if remaining == 0:
+            break
+        if per_layer_count[layer_idx] >= max_experts_per_layer:
+            continue
+
+        mask[layer_idx, expert_idx] = True
+        per_layer_count[layer_idx] += 1
+        remaining -= 1
+
+    if remaining != 0:
+        raise RuntimeError(
+            f"Failed to allocate requested GPU expert budget; {remaining} experts "
+            f"were left unassigned. This indicates an internal placement bug."
+        )
+
+    return mask
+
+
+def _build_report(
+    scores: torch.Tensor,
+    mask: torch.Tensor,
+    requested_num_gpu_experts: int,
+    strategy: str,
+    min_experts_per_layer: int,
+    max_experts_per_layer: Optional[int],
+    alpha: float,
+    previous_scores_provided: bool,
+) -> dict[str, Any]:
+    selected_scores = scores[mask]
+    total_score = float(scores.sum().item())
+    selected_score = float(selected_scores.sum().item()) if selected_scores.numel() else 0.0
+    per_layer_counts_tensor = mask.sum(dim=1).to(dtype=torch.float32)
+    per_layer_counts = [int(v) for v in mask.sum(dim=1).tolist()]
+
+    if per_layer_counts_tensor.numel() > 1:
+        layer_std = float(per_layer_counts_tensor.std(unbiased=False).item())
+    else:
+        layer_std = 0.0
+
+    if math.isclose(total_score, 0.0):
+        expected_hit_rate = 0.0
+    else:
+        expected_hit_rate = selected_score / total_score
+
+    return {
+        "strategy": strategy,
+        "num_layers": int(scores.shape[0]),
+        "num_experts_per_layer": int(scores.shape[1]),
+        "requested_num_gpu_experts": int(requested_num_gpu_experts),
+        "actual_num_gpu_experts": int(mask.sum().item()),
+        "expected_hit_rate": expected_hit_rate,
+        "selected_score": selected_score,
+        "total_score": total_score,
+        "per_layer_counts": per_layer_counts,
+        "layer_min": int(min(per_layer_counts)) if per_layer_counts else 0,
+        "layer_max": int(max(per_layer_counts)) if per_layer_counts else 0,
+        "layer_mean": float(per_layer_counts_tensor.mean().item())
+        if per_layer_counts_tensor.numel()
+        else 0.0,
+        "layer_std": layer_std,
+        "min_experts_per_layer": int(min_experts_per_layer),
+        "max_experts_per_layer": max_experts_per_layer,
+        "alpha": float(alpha),
+        "previous_scores_provided": previous_scores_provided,
+    }
+
+
+def _clamp_int(value: int, min_value: int, max_value: int) -> int:
+    return min(max(int(value), min_value), max_value)
\ No newline at end of file
diff --git a/kt-kernel/python/experts_base.py b/kt-kernel/python/experts_base.py
index bc58b98f0..0d9a174bb 100644
--- a/kt-kernel/python/experts_base.py
+++ b/kt-kernel/python/experts_base.py
@@ -17,59 +17,63 @@
 
 from kt_kernel import kt_kernel_ext
 
+try:
+    from .expert_placement import plan_gpu_expert_placement
+except ImportError:
+    # Keep compatibility with existing direct test imports that put
+    # kt-kernel/python on sys.path and import experts_base as a top-level module.
+    from expert_placement import plan_gpu_expert_placement
 
 def generate_gpu_experts_masks(
     activation_freq: torch.Tensor,
     num_gpu_experts: int,
-) -> torch.Tensor:
+    strategy: str = "frequency",
+    *,
+    min_experts_per_layer: int = 0,
+    max_experts_per_layer: Optional[int] = None,
+    previous_scores: Optional[torch.Tensor] = None,
+    alpha: float = 0.8,
+    return_report: bool = False,
+) -> torch.Tensor | tuple[torch.Tensor, dict]:
     """
-    Generate GPU experts masks based on activation frequency.
+    Generate GPU expert masks from activation statistics.
 
-    Selects the top `num_gpu_experts` experts with highest activation frequency
-    across all layers to be placed on GPU.
+    By default, this preserves the original behavior: globally select the top
+    ``num_gpu_experts`` experts by activation frequency.
+
+    The optional ``score_aware_layer_balanced`` strategy uses EMA-smoothed
+    scores and layer coverage bounds to avoid over-concentrating GPU experts in
+    only a few MoE layers.
 
     Args:
-        activation_freq: Activation frequency table of shape (num_layers, num_experts).
-                         Higher values indicate more frequently activated experts.
-        num_gpu_experts: Total number of experts to place on GPU across all layers.
+        activation_freq: Activation frequency table of shape
+            ``(num_layers, num_experts)``.
+        num_gpu_experts: Total number of experts to place on GPU.
+        strategy: ``frequency`` or ``score_aware_layer_balanced``.
+        min_experts_per_layer: Minimum GPU experts per layer for the
+            score-aware layer-balanced strategy.
+        max_experts_per_layer: Maximum GPU experts per layer for the
+            score-aware layer-balanced strategy.
+        previous_scores: Optional previous activation scores for EMA smoothing.
+        alpha: EMA coefficient for current scores when previous_scores is set.
+        return_report: If True, return ``(mask, report)``.
 
     Returns:
-        gpu_experts_masks: Boolean mask of shape (num_layers, num_experts) on CPU.
-                           True means the expert should be on GPU.
-
-    Example:
-        >>> activation_freq = torch.tensor([
-        ...     [0.1, 0.5, 0.3, 0.8],  # layer 0
-        ...     [0.2, 0.4, 0.9, 0.1],  # layer 1
-        ... ])
-        >>> masks = generate_gpu_experts_masks(activation_freq, num_gpu_experts=3)
-        >>> # Top 3: layer0-expert3 (0.8), layer1-expert2 (0.9), layer0-expert1 (0.5)
-        >>> masks
-        tensor([[False,  True, False,  True],
-                [False, False,  True, False]])
+        Boolean mask of shape ``(num_layers, num_experts)`` on CPU, or
+        ``(mask, report)`` when return_report=True.
     """
-    num_layers, num_experts_per_layer = activation_freq.shape
-    total_experts = num_layers * num_experts_per_layer
-
-    # Clamp num_gpu_experts to valid range
-    num_gpu_experts = min(num_gpu_experts, total_experts)
-    num_gpu_experts = max(num_gpu_experts, 0)
-
-    if num_gpu_experts == 0:
-        return torch.zeros(num_layers, num_experts_per_layer, dtype=torch.bool, device="cpu")
-
-    # Flatten and find top-k indices
-    flat_freq = activation_freq.view(-1).to(device="cpu")
-    _, top_indices = torch.topk(flat_freq, k=num_gpu_experts, largest=True, sorted=False)
-
-    # Create mask
-    gpu_experts_masks = torch.zeros(total_experts, dtype=torch.bool, device="cpu")
-    gpu_experts_masks[top_indices] = True
-
-    # Reshape to (num_layers, num_experts)
-    gpu_experts_masks = gpu_experts_masks.view(num_layers, num_experts_per_layer)
-
-    return gpu_experts_masks
+    mask, report = plan_gpu_expert_placement(
+        activation_freq=activation_freq,
+        num_gpu_experts=num_gpu_experts,
+        strategy=strategy,
+        min_experts_per_layer=min_experts_per_layer,
+        max_experts_per_layer=max_experts_per_layer,
+        previous_scores=previous_scores,
+        alpha=alpha,
+    )
+    if return_report:
+        return mask, report
+    return mask
 
 
 class KExpertsCPUBuffer:
diff --git a/kt-kernel/test/per_commit/test_expert_placement.py b/kt-kernel/test/per_commit/test_expert_placement.py
new file mode 100644
index 000000000..1a15250d9
--- /dev/null
+++ b/kt-kernel/test/per_commit/test_expert_placement.py
@@ -0,0 +1,192 @@
+"""Tests for expert placement planning utilities."""
+
+import os
+import sys
+import types
+
+import pytest
+import torch
+
+# Add parent directory to path for CI registration.
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
+from ci.ci_register import register_cpu_ci
+
+# Add kt-kernel/python directly so this test does not require kt_kernel_ext.
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "python"))
+
+# experts_base imports kt_kernel.kt_kernel_ext at module import time. The tests
+# below only exercise pure-Python placement helpers, so a minimal stub is enough.
+if "kt_kernel" not in sys.modules:
+    kt_kernel_stub = types.ModuleType("kt_kernel")
+    kt_kernel_stub.kt_kernel_ext = types.SimpleNamespace()
+    sys.modules["kt_kernel"] = kt_kernel_stub
+
+from expert_placement import plan_gpu_expert_placement
+from experts_base import generate_gpu_experts_masks
+
+
+register_cpu_ci(est_time=5, suite="default")
+
+
+@pytest.mark.cpu
+def test_frequency_strategy_preserves_global_topk_behavior():
+    activation_freq = torch.tensor(
+        [
+            [0.1, 0.5, 0.3, 0.8],
+            [0.2, 0.4, 0.9, 0.1],
+        ]
+    )
+
+    mask = generate_gpu_experts_masks(activation_freq, num_gpu_experts=3)
+
+    assert mask.dtype == torch.bool
+    assert str(mask.device) == "cpu"
+    assert mask.sum().item() == 3
+    assert mask[1, 2]
+    assert mask[0, 3]
+    assert mask[0, 1]
+
+
+@pytest.mark.cpu
+def test_score_aware_layer_balanced_uses_ema_scores():
+    current = torch.tensor(
+        [
+            [10.0, 1.0, 1.0, 1.0],
+            [1.0, 9.0, 1.0, 1.0],
+        ]
+    )
+    previous = torch.tensor(
+        [
+            [0.0, 20.0, 0.0, 0.0],
+            [0.0, 0.0, 18.0, 0.0],
+        ]
+    )
+
+    mask, report = plan_gpu_expert_placement(
+        current,
+        num_gpu_experts=2,
+        strategy="score_aware_layer_balanced",
+        previous_scores=previous,
+        alpha=0.5,
+        max_experts_per_layer=1,
+    )
+
+    # EMA scores make layer0 expert1 and layer1 expert2 the hottest experts.
+    assert mask.sum().item() == 2
+    assert mask[0, 1]
+    assert mask[1, 2]
+    assert report["strategy"] == "score_aware_layer_balanced"
+    assert report["previous_scores_provided"] is True
+    assert report["layer_max"] == 1
+
+
+@pytest.mark.cpu
+def test_score_aware_layer_balanced_respects_min_and_max_layer_bounds():
+    activation_freq = torch.tensor(
+        [
+            [100.0, 90.0, 80.0, 70.0],
+            [1.0, 2.0, 3.0, 4.0],
+            [5.0, 6.0, 7.0, 8.0],
+        ]
+    )
+
+    mask, report = plan_gpu_expert_placement(
+        activation_freq,
+        num_gpu_experts=6,
+        strategy="score_aware_layer_balanced",
+        min_experts_per_layer=1,
+        max_experts_per_layer=2,
+    )
+
+    per_layer = mask.sum(dim=1).tolist()
+
+    assert mask.sum().item() == 6
+    assert per_layer == [2, 2, 2]
+    assert report["layer_min"] == 2
+    assert report["layer_max"] == 2
+
+
+@pytest.mark.cpu
+def test_score_aware_layer_balanced_is_deterministic_for_ties():
+    activation_freq = torch.ones(3, 4)
+
+    mask1 = generate_gpu_experts_masks(
+        activation_freq,
+        num_gpu_experts=5,
+        strategy="score_aware_layer_balanced",
+        max_experts_per_layer=2,
+    )
+    mask2 = generate_gpu_experts_masks(
+        activation_freq,
+        num_gpu_experts=5,
+        strategy="score_aware_layer_balanced",
+        max_experts_per_layer=2,
+    )
+
+    assert torch.equal(mask1, mask2)
+
+
+@pytest.mark.cpu
+def test_return_report_from_public_wrapper():
+    activation_freq = torch.tensor(
+        [
+            [0.1, 0.2],
+            [0.3, 0.4],
+        ]
+    )
+
+    mask, report = generate_gpu_experts_masks(
+        activation_freq,
+        num_gpu_experts=2,
+        return_report=True,
+    )
+
+    assert mask.sum().item() == 2
+    assert report["actual_num_gpu_experts"] == 2
+    assert 0.0 <= report["expected_hit_rate"] <= 1.0
+
+
+@pytest.mark.cpu
+def test_invalid_previous_scores_shape_raises():
+    activation_freq = torch.ones(2, 4)
+    previous_scores = torch.ones(2, 3)
+
+    with pytest.raises(ValueError, match="previous_scores shape"):
+        plan_gpu_expert_placement(
+            activation_freq,
+            num_gpu_experts=2,
+            strategy="score_aware_layer_balanced",
+            previous_scores=previous_scores,
+        )
+
+
+@pytest.mark.cpu
+def test_impossible_layer_constraints_raise():
+    activation_freq = torch.ones(4, 8)
+
+    with pytest.raises(ValueError, match="too small"):
+        plan_gpu_expert_placement(
+            activation_freq,
+            num_gpu_experts=2,
+            strategy="score_aware_layer_balanced",
+            min_experts_per_layer=1,
+        )
+
+    with pytest.raises(ValueError, match="exceeds max_experts_per_layer"):
+        plan_gpu_expert_placement(
+            activation_freq,
+            num_gpu_experts=12,
+            strategy="score_aware_layer_balanced",
+            max_experts_per_layer=2,
+        )
+
+
+@pytest.mark.cpu
+def test_zero_and_oversized_budget_are_clamped_for_frequency():
+    activation_freq = torch.ones(2, 3)
+
+    zero_mask = generate_gpu_experts_masks(activation_freq, num_gpu_experts=0)
+    full_mask = generate_gpu_experts_masks(activation_freq, num_gpu_experts=100)
+
+    assert zero_mask.sum().item() == 0
+    assert full_mask.sum().item() == 6
\ No newline at end of file