@@ -1212,9 +1212,15 @@ def __init__(
         threads: int,
         device_slot_selection: bool,
         kimi_topk: bool,
+        row_blocks: int = 1,
     ) -> None:
         super().__init__(world_size, rank, threads, device_slot_selection)
         self._kimi_topk = bool(kimi_topk)
+        if type(row_blocks) is not int or not 1 <= row_blocks <= _MAX_BLOCKS:
+            raise ValueError("paired gather row blocks must fit the signal plane")
+        if kimi_topk and row_blocks != 1:
+            raise ValueError("fused single-row expert selection requires one CTA")
+        self._row_blocks = row_blocks
 
     @cute.jit
     def __call__(
@@ -1261,6 +1267,7 @@ def __call__(
         first_packs: Int32,
         second_packs: Int32,
         slot_delta_256b: Int32,
+        grid_x: Int32,
         stream: cuda.CUstream,
     ) -> None:
         self.kernel(
@@ -1307,7 +1314,7 @@ def __call__(
             second_packs,
             slot_delta_256b,
         ).launch(
-            grid=(1, 1, 1),
+            grid=(grid_x, 1, 1),
             block=(self._threads, 1, 1),
             max_number_threads=(512, 1, 1),
             min_blocks_per_mp=1,
@@ -1398,6 +1405,14 @@ def kernel(
             signal15,
         )
         tidx, _, _ = cute.arch.thread_idx()
+        row_group, grid = Int32(0), Int32(1)
+        if cutlass.const_expr(self._row_blocks != 1):
+            row_group, _, _ = cute.arch.block_idx()
+            grid, _, _ = cute.arch.grid_dim()
+        # Matching CTAs on every rank publish the same complete rows. A CTA
+        # never reads a row staged by a different CTA, so its peer barrier
+        # establishes all input dependencies without a local grid barrier.
+        owned_rows = (batch - row_group + grid - 1) // grid
         slot_offset = Int64(0)
         if cutlass.const_expr(self._device_slot_selection):
             generation = ld_relaxed_gpu_u32(
@@ -1409,21 +1424,29 @@ def kernel(
         local_stage = self._staging_words(staging[self._rank] + slot_offset)
 
         linear = Int32(tidx)
-        while linear < batch * first_packs:
-            batch_index = linear // first_packs
-            pack = linear - batch_index * first_packs
+        while linear < owned_rows * first_packs:
+            local_row = linear // first_packs
+            batch_index = local_row * grid + row_group
+            pack = linear - local_row * first_packs
+            source_pack = Int64(linear)
+            if cutlass.const_expr(self._row_blocks != 1):
+                source_pack = Int64(batch_index) * Int64(first_packs) + Int64(pack)
             _copy_16b(
-                local_first + Int64(linear) * Int64(4),
+                local_first + source_pack * Int64(4),
                 local_stage
                 + (Int64(batch_index) * Int64(combined_packs) + Int64(pack)) * Int64(4),
             )
             linear += Int32(self._threads)
         linear = Int32(tidx)
-        while linear < batch * second_packs:
-            batch_index = linear // second_packs
-            pack = linear - batch_index * second_packs
+        while linear < owned_rows * second_packs:
+            local_row = linear // second_packs
+            batch_index = local_row * grid + row_group
+            pack = linear - local_row * second_packs
+            source_pack = Int64(linear)
+            if cutlass.const_expr(self._row_blocks != 1):
+                source_pack = Int64(batch_index) * Int64(second_packs) + Int64(pack)
             _copy_16b(
-                local_second + Int64(linear) * Int64(4),
+                local_second + source_pack * Int64(4),
                 local_stage
                 + (
                     Int64(batch_index) * Int64(combined_packs)
@@ -1442,11 +1465,12 @@ def kernel(
             max_blocks=_MAX_BLOCKS,
         )
 
-        first_output_packs = batch * Int32(self._world_size) * first_packs
+        first_output_packs = owned_rows * Int32(self._world_size) * first_packs
         linear = Int32(tidx)
         while linear < first_output_packs:
             batch_index = linear // (Int32(self._world_size) * first_packs)
             row_pack = linear - (batch_index * Int32(self._world_size) * first_packs)
+            batch_index = batch_index * grid + row_group
             source_rank = row_pack // first_packs
             pack = row_pack - source_rank * first_packs
             # Default to this rank's own slice so the address is always
@@ -1471,20 +1495,26 @@ def kernel(
                     source_address = Int64(
                         (source_words + (source_base + Int64(pack)) * Int64(4)).toint()
                     )
+            output_pack = Int64(linear)
+            if cutlass.const_expr(self._row_blocks != 1):
+                output_pack = Int64(batch_index) * Int64(self._world_size) * Int64(
+                    first_packs
+                ) + Int64(row_pack)
             _copy_16b_addr(
                 source_address,
-                Int64((output_first + Int64(linear) * Int64(4)).toint()),
+                Int64((output_first + output_pack * Int64(4)).toint()),
             )
             linear += Int32(self._threads)
 
         if cutlass.const_expr(not self._kimi_topk):
-            second_output_packs = batch * Int32(self._world_size) * second_packs
+            second_output_packs = owned_rows * Int32(self._world_size) * second_packs
             linear = Int32(tidx)
             while linear < second_output_packs:
                 batch_index = linear // (Int32(self._world_size) * second_packs)
                 row_pack = linear - (
                     batch_index * Int32(self._world_size) * second_packs
                 )
+                batch_index = batch_index * grid + row_group
                 source_rank = row_pack // second_packs
                 pack = row_pack - source_rank * second_packs
                 source_address = Int64(
@@ -1511,9 +1541,14 @@ def kernel(
                                 source_words + (source_base + Int64(pack)) * Int64(4)
                             ).toint()
                         )
+                output_pack = Int64(linear)
+                if cutlass.const_expr(self._row_blocks != 1):
+                    output_pack = Int64(batch_index) * Int64(self._world_size) * Int64(
+                        second_packs
+                    ) + Int64(row_pack)
                 _copy_16b_addr(
                     source_address,
-                    Int64((output_second + Int64(linear) * Int64(4)).toint()),
+                    Int64((output_second + output_pack * Int64(4)).toint()),
                 )
                 linear += Int32(self._threads)
         else:
@@ -1694,7 +1729,7 @@ class SharedStorage:
                 _a2a_graph_epoch_arrive(
                     (self_signal + Int64(_GRAPH_EPOCH_INDEX)).toint(),
                     (self_signal + Int64(_GRAPH_ARRIVED_INDEX)).toint(),
-                    Uint32(1),
+                    Uint32(grid),
                 )
 
 
@@ -2164,6 +2199,7 @@ def _pair_launcher_key(
     threads: int,
     device_slot_selection: bool,
     kimi_topk: bool,
+    row_blocks: int = 1,
 ) -> tuple[object, ...]:
     """Return the cache key for a paired all-gather launcher."""
     return (
@@ -2172,6 +2208,7 @@ def _pair_launcher_key(
         int(threads),
         bool(device_slot_selection),
         bool(kimi_topk),
+        int(row_blocks),
     )
 
 
@@ -2181,6 +2218,7 @@ def is_all_gather_pair_prepared(
     threads: int,
     device_slot_selection: bool,
     kimi_topk: bool = False,
+    row_blocks: int = 1,
 ) -> bool:
     return (
         _pair_launcher_key(
@@ -2189,6 +2227,7 @@ def is_all_gather_pair_prepared(
             threads,
             device_slot_selection,
             kimi_topk,
+            row_blocks,
         )
         in _PREPARED_PAIR_LAUNCHERS
     )
@@ -2201,6 +2240,7 @@ def _get_compiled_all_gather_pair(
     threads: int,
     device_slot_selection: bool,
     kimi_topk: bool = False,
+    row_blocks: int = 1,
 ) -> Callable:
     """Compile and retain one paired all-gather launcher specialization."""
     launch = _AllGatherPairLaunch(
@@ -2209,13 +2249,15 @@ def _get_compiled_all_gather_pair(
         threads,
         device_slot_selection,
         kimi_topk,
+        row_blocks,
     )
     key = _pair_launcher_key(
         world_size,
         rank,
         threads,
         device_slot_selection,
         kimi_topk,
+        row_blocks,
     )
     raise_if_kernel_resolution_frozen("cute.compile", target=launch, cache_key=key)
     p_u32 = _u32_ptr(16, align=4)
@@ -2236,6 +2278,7 @@ def _get_compiled_all_gather_pair(
         1,
         1,
         1,
+        1,
         current_cuda_stream(),
         compile_spec=KernelCompileSpec.from_key(
             (
@@ -2251,6 +2294,7 @@ def _get_compiled_all_gather_pair(
                 "threads",
                 "device_slot_selection",
                 "kimi_topk",
+                "row_blocks",
             ),
         ),
     )
@@ -2285,6 +2329,7 @@ def run(
             int(first_packs),
             int(second_packs),
             int(slot_delta_256b),
+            min(int(batch), row_blocks),
             current_cuda_stream(),
         )
 
 
@@ -135,6 +135,8 @@ def query_from_runtime(runtime, *, surface, call) -> PcieQuery:
             and (out is None or out.data_ptr() % 16 == 0),
         )
     elif surface.endswith("all_gather_pair"):
+        from .pcie_dcp_a2a import _pair_outputs_disjoint
+
         first, second = (
             _call_value(call, "local_first"),
             _call_value(call, "local_second"),
@@ -144,6 +146,10 @@ def query_from_runtime(runtime, *, surface, call) -> PcieQuery:
             second_dtype=str(second.dtype),
             first_shape=tuple(first.shape),
             second_shape=tuple(second.shape),
+            row_blocks=call.get("row_blocks", 1),
+            disjoint_output=_pair_outputs_disjoint(
+                first, second, call.get("out_first"), call.get("out_second")
+            ),
         )
     elif surface.endswith("all_gather_pair_kimi_topk"):
         metadata["threads"] = 512
@@ -227,7 +233,12 @@ def compile_dcp_surface(query_payload, ordinal):
         if surface.endswith("all_gather_pair"):
             return {
                 slot: cute._get_compiled_all_gather_pair(
-                    query.world_size, query.rank, int(call["threads"]), slot, False
+                    query.world_size,
+                    query.rank,
+                    int(call["threads"]),
+                    slot,
+                    False,
+                    int(call.get("row_blocks", 1)),
                 )
                 for slot in (False, True)
             }
 
@@ -140,6 +140,12 @@ def _validate_query(query: PcieQuery, device) -> None:
                 raise ValueError(
                     "word exchange requires disjoint aligned data and owned TP2/TP4 state"
                 )
+    if query.surface.endswith(".all_gather_pair"):
+        blocks = query.call.get("row_blocks", 1)
+        if type(blocks) is not int or not 1 <= blocks <= 64:
+            raise ValueError("paired gather row blocks must fit the signal plane")
+        if blocks > 1 and not query.call.get("disjoint_output", False):
+            raise ValueError("parallel paired gather requires disjoint outputs")
     if query.surface == "kimi_topk16":
         if query.topology != "local":
             raise ValueError("stateless kimi_topk16 uses one local device")
 
@@ -282,6 +282,28 @@ def lse_reduce_scatter_reference(
     return (outputs * weights.unsqueeze(-1)).sum(dim=0).to(partial_outputs.dtype)
 
 
+def _pair_outputs_disjoint(first, second, out_first, out_second):
+    """Check raw contiguous spans; omitted outputs are freshly allocated."""
+
+    def overlap(a, b):
+        return (
+            a is not None
+            and b is not None
+            and a.device == b.device
+            and a.data_ptr() < b.data_ptr() + b.nbytes
+            and b.data_ptr() < a.data_ptr() + a.nbytes
+        )
+
+    return not (
+        overlap(out_first, out_second)
+        or any(
+            overlap(source, out)
+            for source in (first, second)
+            for out in (out_first, out_second)
+        )
+    )
+
+
 class PCIeDCPA2A:
     """One ordered IPC channel for DCP attention collectives."""
 
@@ -1251,6 +1273,13 @@ def _all_gather_pair_on_device(
             if not output.is_contiguous():
                 raise ValueError(f"{name} output must be contiguous")
         threads, _ = self._resolve_launch_config(threads=threads, block_limit=1)
+        row_blocks = state.query.call.get("row_blocks", 1)
+        if row_blocks > 1 and not _pair_outputs_disjoint(
+            local_first, local_second, out_first, out_second
+        ):
+            raise ValueError(
+                "parallel paired gather outputs must not overlap its inputs or each other"
+            )
         capturing = _is_current_stream_capturing(self.device)
         if capturing:
             from ._dcp_a2a_cute import is_all_gather_pair_prepared
@@ -1261,6 +1290,7 @@ def _all_gather_pair_on_device(
                 threads,
                 True,
                 False,
+                row_blocks,
             ):
                 raise RuntimeError(
                     "cold PCIe DCP paired gather CUDA graph capture is not "