diff --git a/src/maxtext/layers/attention_compressed.py b/src/maxtext/layers/attention_compressed.py
index 6530a8b0fb..f4f9a71e13 100644
--- a/src/maxtext/layers/attention_compressed.py
+++ b/src/maxtext/layers/attention_compressed.py
@@ -1608,9 +1608,9 @@ def __call__(
     if self.query_pre_attn_scalar and self.query_pre_attn_scalar != 1.0:
       q = q * self.query_pre_attn_scalar
 
-    # Build indexer mask explicitly for tokamax splash kernel
+    # Build indexer mask explicitly for tokamax splash kernel (CSA dynamic path)
     indexer_mask = None
-    if self.attention_kernel == "flash" and compressed_mask is not None:
+    if self.attention_kernel == "flash" and compressed_mask is not None and self.compress_ratio == 4:
       indexer_mask = self.attention_op.generate_attention_mask(
           q,
           unpadded_kv,
@@ -1640,6 +1640,7 @@ def __call__(
         cached_values=current_kv_cache,
         indexer_mask=indexer_mask,
         decoder_segment_ids_kv=decoder_segment_ids_kv,
+        pad_kv_total=pad_kv_total,
     )
 
     # Reverse RoPE on Values
diff --git a/src/maxtext/layers/attention_op.py b/src/maxtext/layers/attention_op.py
index 003aed30fa..e82dcc4b3f 100644
--- a/src/maxtext/layers/attention_op.py
+++ b/src/maxtext/layers/attention_op.py
@@ -216,6 +216,64 @@ def __hash__(self):
     )
 
 
+class HCAStaticMask(splash_attention_mask._ComputableMask):  # pylint: disable=protected-access
+  """Static mask for DeepSeek-V4 Heavily Compressed Attention (HCA).
+
+  Local tokens attend causally (and with sliding window if set).
+  Compressed tokens attend fully without masking.
+  Tile padding tokens between local and compressed tokens are masked out.
+  """
+
+  def __init__(
+      self,
+      shape: tuple[int, int],
+      local_kv_len: int,
+      compressed_kv_len: int,
+      pad_kv_total: int = 0,
+      sliding_window_size: int | None = None,
+      shard_count: int = 1,
+  ):
+    compress_ratio = max(1, local_kv_len // max(1, compressed_kv_len))
+    is_power_of_2 = (compress_ratio & (compress_ratio - 1)) == 0
+    compress_shift = int(np.log2(compress_ratio)) if is_power_of_2 else None
+    comp_start = local_kv_len + pad_kv_total
+
+    def hca_mask_fn(q_ids, kv_ids):
+      if q_ids.size == 0 or kv_ids.size == 0:
+        return np.empty((q_ids.shape[0], kv_ids.shape[1]), dtype=np.bool_)
+
+      q_valid = q_ids < local_kv_len
+      is_local = (kv_ids < local_kv_len) & q_valid
+      causal = q_ids >= kv_ids
+      if sliding_window_size is not None:
+        local_valid = causal & ((q_ids - kv_ids) < sliding_window_size)
+      else:
+        local_valid = causal
+
+      c_idx = kv_ids - comp_start
+      if compress_shift is not None:
+        c_thresh = (q_ids + 1) >> compress_shift
+      else:
+        c_thresh = (q_ids + 1) // compress_ratio
+
+      compressed_valid = q_valid & (c_idx >= 0) & (c_idx < c_thresh) & (c_idx < compressed_kv_len)
+      return (is_local & local_valid) | compressed_valid
+
+    super().__init__(
+        shape=shape,
+        mask_function=hca_mask_fn,
+        shard_count=shard_count,
+    )
+
+  def __eq__(self, other: object):
+    if not isinstance(other, type(self)):
+      return NotImplemented
+    return self.shape == other.shape and np.array_equal(self.q_sequence, other.q_sequence)
+
+  def __hash__(self):
+    return hash((type(self), self.shape, self.q_sequence.tobytes() if self.q_sequence is not None else None))
+
+
 class BlockCausalMask(splash_attention_mask._ComputableMask):  # pylint: disable=protected-access,abstract-method
   """Lazy mask with bidirectional attention within causal blocks."""
 
@@ -1368,6 +1426,7 @@ def apply_attention(
       compressed_mask: Optional[Array] = None,
       record_max_logits: bool = False,
       decoder_segment_ids_kv: Optional[Array] = None,
+      pad_kv_total: int = 0,
       *,
       qk_product_einsum: Callable[..., Array],
       wv_product_einsum: Callable[..., Array],
@@ -1463,6 +1522,7 @@ def apply_attention(
             use_ragged_attention=use_ragged_attention,
             record_max_logits=record_max_logits,
             decoder_segment_ids_kv=decoder_segment_ids_kv,
+            pad_kv_total=pad_kv_total,
         )
         if max_logits is not None:
           self.max_logits = nnx.Intermediate(max_logits)
@@ -1636,6 +1696,7 @@ def tpu_flash_attention(
       use_ragged_attention: bool = False,
       record_max_logits: bool = False,
       decoder_segment_ids_kv: Array | None = None,
+      pad_kv_total: int = 0,
   ) -> tuple[Array, Array]:
     """TPU Flash Attention."""
 
@@ -1853,6 +1914,16 @@ def wrap_ulysses_splash_kernel(single_head_mask):
       use_load_balanced_cp = cp_size > 1 and load_balanced_context_parallel
       if self.attention_type == AttentionType.FULL:
         mask = mask_module.FullMask(mask_shape)
+      elif self.attention_type == AttentionType.COMPRESSED and indexer_mask is None:
+        local_kv_len = query.shape[2]
+        compressed_kv_len = key.shape[2] - query.shape[2] - pad_kv_total
+        mask = HCAStaticMask(
+            shape=mask_shape,
+            local_kv_len=local_kv_len,
+            compressed_kv_len=compressed_kv_len,
+            pad_kv_total=pad_kv_total,
+            sliding_window_size=self.sliding_window_size if self.sliding_window_size else None,
+        )
       elif self.attention_type == AttentionType.BLOCK_DIFFUSION:
         mask_type = LoadBalancedBlockCausalMask if use_load_balanced_cp else BlockCausalMask
         mask_kwargs = {"cp_size": cp_size} if use_load_balanced_cp else {}
@@ -1909,6 +1980,8 @@ def wrap_ulysses_splash_kernel(single_head_mask):
       single_head_mask = mask  # tokamax now just uses a single mask and assumes broadcast to all heads
       if self.config.use_max_logit_estimate > 0:
         sa_config = dataclasses.replace(sa_config, max_logit_const=self.config.use_max_logit_estimate)
+      if self.attention_type == AttentionType.COMPRESSED and indexer_mask is None:
+        sa_config = dataclasses.replace(sa_config, dq_reduction_steps=3)
 
       # Create the splash attention kernel object separately, jit it for performance
       @partial(
@@ -2190,6 +2263,14 @@ def kernel_fn(q, k, v, d, s):
 
       return attention_output, None
 
+    # Pad query and segment IDs to mask_shape if sequence length is not aligned to block size
+    orig_q_len = query.shape[2]
+    pad_q = mask_shape[0] - query.shape[2]
+    if pad_q > 0:
+      query = jnp.pad(query, ((0, 0), (0, 0), (0, pad_q), (0, 0)))
+      if decoder_segment_ids is not None:
+        decoder_segment_ids = jnp.pad(decoder_segment_ids, ((0, 0), (0, pad_q)))
+
     query = self._maybe_shard_with_pspec(query, axis_names_q)
     key = self._maybe_shard_with_pspec(key, axis_names_kv)
     value = self._maybe_shard_with_pspec(value, axis_names_kv)
@@ -2215,6 +2296,11 @@ def kernel_fn(q, k, v, d, s):
 
     x, max_logits = ret
     x = jnp.transpose(x, axes=(0, 2, 1, 3))
+    # Slice outputs back to unpadded sequence length
+    if pad_q > 0:
+      x = x[:, :orig_q_len, :, :]
+      if record_max_logits:
+        max_logits = max_logits[:, :, :orig_q_len]
 
     if record_max_logits:
       # Max over sequence length (dim 2 of max_logits)
@@ -2771,6 +2857,7 @@ def __call__(
       slot: Optional[int] = None,
       record_max_logits: bool = False,
       decoder_segment_ids_kv: Optional[Array] = None,
+      pad_kv_total: int = 0,
   ):
     if cached_values is None:
       prefill_kv_cache, ar_kv_cache = None, None
@@ -2818,6 +2905,7 @@ def __call__(
         qk_product_einsum=self.AqtEinsum_0,
         wv_product_einsum=self.AqtEinsum_1,
         decoder_segment_ids_kv=decoder_segment_ids_kv,
+        pad_kv_total=pad_kv_total,
     )
 
     if ar_kv_cache is None: