@@ -482,6 +482,8 @@ def add(name, function, kind, rows, tokens, fields, *, enabled=None):
                 forward = PreparedVisionGeneration(encoder, vision_inputs, vision_state, completion, frame, sequence)
                 add(name, forward, "vision", 1, capacity, (*vision_inputs.buffers(), "vision_feature_slot"))
                 vision_owners.extend((vision_metadata, encoder, forward))
+        prefill_feedback = (torch.empty((config.prefill_chunk_size, 4096), dtype=torch.bfloat16, device=device)
+                            if config.mtp else None)
         for enabled in graph_modes:
             suffix = "-no-mtp" if config.mtp and not enabled else ""
             mode_inputs = tuple(field for field in forward_inputs if enabled or field != "draft_page_tables")
@@ -493,7 +495,7 @@ def add(name, function, kind, rows, tokens, fields, *, enabled=None):
                     metadata[name] = DecoderMetadata.allocate(capacity, rows, blocks, device)
                     storage = prefill_storage if mode == "prefill" else decoder_storage
                     scratch[name] = storage.rows(capacity, mode, sample_rows=rows)
-                    feedback = (torch.empty((capacity, 4096), dtype=torch.bfloat16, device=device)
+                    feedback = (prefill_feedback[:capacity]
                                 if enabled and kind != "scored_prefill" else None)
                     decoders[name] = PreparedDecoder(weights, scratch[name], pool.decoder, metadata[name], mode=mode,
                         scoring=scoring if kind == "scored_prefill" else None,
 
@@ -78,6 +78,9 @@ def buffer(*shape):
         # Fused layer-input projection: query 1536, latent 512, index key 128 and index head weights 32.
         projected = buffer(2208)
         prefill = context_ranks > 1 and mode == "prefill"
+        query = buffer(heads, 256)
+        # CP prefill consumes each query slice before expanding values into the same rows.
+        expanded = query if prefill else buffer(heads, 256)
         exchange = min(tokens, SLICE_ROWS) if context_ranks > 1 and not prefill else 0
         # Context-parallel rows absorb, attend and expand one slice at a time; otherwise every row at once.
         absorbed = exchange or (min(tokens, PREFILL_SLICE_ROWS) if prefill else tokens)
@@ -90,7 +93,7 @@ def buffer(*shape):
 
         window = min(WINDOW_RECORDS, blocks * 256) if remote and prefill else 0
         return cls(projected, projected[:, :1536], buffer(1536), projected[:, 1536:2048], buffer(512),
-            buffer(heads, 256), zeros(absorbed, heads, 512), zeros(absorbed, heads, 512), buffer(heads, 256),
+            query, zeros(absorbed, heads, 512), zeros(absorbed, heads, 512), expanded,
             buffer(4096), zeros(tokens, dtype=torch.int32), index,
             zeros(exchange, heads * context_ranks, 512), zeros(exchange, heads * context_ranks, 512),
             zeros(exchange, heads * context_ranks, dtype=torch.float32), zeros(positions, 528, dtype=torch.uint8),
 
@@ -126,13 +126,18 @@ def _draft_metadata(Lengths, Starts, Slots, Tokens, Hidden, Tables, StateRows, R
 
 
 class DraftMetadata:
-    def __init__(self, tokens, rows, blocks, device, *, shards=None):
+    def __init__(self, tokens, rows, blocks, device, *, shards=None, hidden=None):
         from .dcp import ContextShards
 
         self.shards = ContextShards() if shards is None else shards
         self.indexed = IndexMetadata.allocate(tokens, rows, blocks, device)
         self.token_ids = torch.zeros(tokens, dtype=torch.int64, device=device)
-        self.hidden = torch.zeros((tokens, 4096), dtype=torch.bfloat16, device=device)
+        if hidden is None:
+            hidden = torch.zeros((tokens, 4096), dtype=torch.bfloat16, device=device)
+        if (hidden.shape != (tokens, 4096) or hidden.dtype != torch.bfloat16
+                or hidden.device != self.token_ids.device or not hidden.is_contiguous()):
+            raise ValueError("GLM draft hidden storage must be contiguous BF16 [tokens, 4096] on the metadata device")
+        self.hidden = hidden
         self.indices = torch.arange(tokens, device=device)
         self.output_indices = torch.zeros(rows, dtype=torch.int64, device=device)
         self.num_tokens = torch.zeros(1, dtype=torch.int32, device=device)
 
@@ -133,7 +133,9 @@ def __init__(self, owner, generation, feedback, name):
         self.owner, self.generation, self.feedback = owner, generation, feedback
         tokens = feedback.shape[0]
         device = feedback.device
-        self.metadata = DraftMetadata(tokens, 1, owner.inputs.page_tables.shape[1], device, shards=owner.shards)
+        # Target generation finishes before draft metadata overwrites its normalized attention input.
+        self.metadata = DraftMetadata(tokens, 1, owner.inputs.page_tables.shape[1], device, shards=owner.shards,
+                                      hidden=generation.decoder.scratch.attention_hc.normalized)
         self.absorb = owner.draft(self.metadata, f"{name}.absorb", cache_only=True, mode="prefill")
         self.previous = torch.zeros((1, 4096), dtype=torch.bfloat16, device=device)
         self.indices = torch.arange(tokens, device=device)
 
@@ -23,7 +23,8 @@ def test_fused_draft_metadata_matches_reference(world, capacity, requests):
     generator = torch.Generator().manual_seed(world * 10 + capacity)
     blocks, slots_capacity = 40, 8
     shards = ContextShards.on(1, world, device) if world > 1 else ContextShards()
-    fused = DraftMetadata(requests * capacity, requests, blocks, device, shards=shards)
+    storage = torch.full((requests * capacity, 4096), float("nan"), dtype=torch.bfloat16, device=device)
+    fused = DraftMetadata(requests * capacity, requests, blocks, device, shards=shards, hidden=storage)
     reference = DraftMetadata(requests * capacity, requests, blocks, device, shards=shards)
     slots = torch.tensor([3, 0, 5, 6][:requests], dtype=torch.int64, device=device)
     lengths = torch.randint(0, capacity + 1, (requests,), generator=generator).to(device)
 
@@ -6,14 +6,18 @@
 
 import os
 import time
+from dataclasses import replace
 from pathlib import Path
+from types import SimpleNamespace
 
 import pytest
 import torch
 from b12x.attention import sparse_mla
 from cuda.bindings import driver as cuda
 
 from ignition.components.parallel import TensorParallelShard
+from ignition.components.linear import BatchedQuantizedProjection, BatchedUnquantizedProjection
+from ignition.components.sparse_mla import PreparedSparseMLA
 from ignition.preparation.checkpoint import inspect_indexed_checkpoint
 from ignition.preparation.compilation import compile_forward
 from ignition.preparation.execution import GraphSignature, PreparedExecution
@@ -235,3 +239,86 @@ def forward():
             if worker is not None:
                 worker.close()
             graph.reset()
+
+
+@pytest.mark.parametrize("model_weights", [2], indirect=True)
+@torch.no_grad()
+def test_cp_prefill_query_reuse_matches_separate_values_across_slices(model_weights):
+    from b12x.attention._shared.mla.reference import pack_mla_kv_cache_reference
+    from ignition.recipes.glm53.dcp import ContextShards, PREFILL_SLICE_ROWS
+
+    module, weights, world = model_weights
+    device, capacity, blocks, heads = torch.device("cuda:0"), 8192, 32, 64 // world
+    torch.manual_seed(53)
+    caches = [pack_mla_kv_cache_reference(torch.randn(blocks * 256, 512, device=device)).view(-1, 64, 528)
+              for _ in range(world)]
+    peers = torch.tensor([cache.data_ptr() for cache in caches], dtype=torch.int64, device=device)
+    storage = MLAScratch.allocate(capacity, heads, blocks, "prefill", device, context_ranks=world)
+    queries = torch.randn(storage.query.shape, dtype=torch.bfloat16, device=device) * .1
+    arena, operations, graphs, compiled = ScratchArena(device), {}, {}, {}
+    kv = weights["kv_b_proj"]
+    projection = BatchedUnquantizedProjection if isinstance(kv, torch.Tensor) else BatchedQuantizedProjection
+    if isinstance(kv, torch.Tensor):
+        values = kv.view(heads, 512, 512)
+        absorb_weights, expand_weights = values[:, :256], values[:, 256:]
+    else:
+        values, scales = kv.weight.values.view(heads, 512, 512), kv.weight.scale_rows.view(heads, 512, 16)
+        absorb_weights, expand_weights = (values[:, :256], scales[:, :256]), (values[:, 256:], scales[:, 256:])
+    with RecordedPreparationSession(device=device, checkpoint="glm53-cp-scratch",
+            workload_schema="glm-cp-scratch-v1", numerical_policy="default") as session:
+        session.register_scratch(arena)
+        for rows in (capacity, 2049):
+            for alias in (False, True):
+                key = rows, alias
+                s = storage.rows(rows, "prefill")
+                if not alias:
+                    s = replace(s, expanded=torch.empty_like(s.expanded))
+                metadata = IndexMetadata.allocate(rows, 1, blocks, device)
+                metadata.positions.copy_(torch.arange(rows, device=device) + 8)
+                metadata.block_table[0].copy_(torch.arange(blocks, device=device))
+                s.index.physical.fill_(-1)
+                s.index.physical[:, :4].copy_(torch.arange(4, device=device))
+                s.index.selected_lengths.fill_(4)
+                # Isolate the production CP slice loop with real projections, attention and peer-cache reads.
+                op = PreparedGlmMLA.__new__(PreparedGlmMLA)
+                op.scratch, op.metadata, op.state = s, metadata, SimpleNamespace(peers=peers)
+                op.context = SimpleNamespace(shards=ContextShards.on(0, world, device))
+                op.windows, op.slices = (), []
+                caps = sparse_mla.Caps(device=device, num_q_heads=heads, max_q_rows=PREFILL_SLICE_ROWS, max_width=2051,
+                    softmax_scale=module.scaling, max_batch=PREFILL_SLICE_ROWS, max_kv_rows=s.slab.shape[0],
+                    kv_dtype=torch.uint8, head_dim=512, v_head_dim=512, page_size=64,
+                    model_type=sparse_mla.ModelType.GLM_NEXT, cache_record_bytes=528, mode="extend")
+                for start in range(0, rows, PREFILL_SLICE_ROWS):
+                    end = min(start + PREFILL_SLICE_ROWS, rows)
+                    absorbed, attended = s.absorbed_query[:end - start], s.attended[:end - start]
+                    name = f"cp{rows}.{alias}.{start}"
+                    absorb = projection(s.query[start:end].transpose(0, 1), absorb_weights,
+                        absorbed.transpose(0, 1), major="n", session=session, name=name + ".absorb")
+                    attention = PreparedSparseMLA(absorbed, s.slab.view(-1, 64, 528), s.index.physical[start:end],
+                        s.visible[start:end], s.index.selected_lengths[start:end], attended, caps=caps,
+                        session=session, name=name + ".attention", workspace=arena)
+                    expand = projection(attended.transpose(0, 1), expand_weights,
+                        s.expanded[start:end].transpose(0, 1), major="k", session=session, name=name + ".expand")
+                    op.slices.append((start, end, absorb, attention, expand))
+                operations[key] = op
+        session.tune()
+        for key, op in operations.items():
+            def forward(op=op):
+                op.scratch.query.copy_(queries[:op.scratch.query.shape[0]])
+                op.attend_gathered_context()
+                arena.tensor.fill_(255)
+            compiled[key] = compile_forward(forward)
+            compiled[key]()
+            graphs[key] = torch.cuda.CUDAGraph()
+            with session.capture(), torch.cuda.graph(graphs[key]):
+                compiled[key]()
+        session.freeze()
+        for rows in (capacity, 2049, capacity):
+            graphs[rows, False].replay()
+            expected = operations[rows, False].scratch.expanded.clone()
+            storage.query.fill_(float("nan"))
+            graphs[rows, True].replay()
+            torch.cuda.synchronize(device)
+            assert torch.equal(operations[rows, True].scratch.expanded, expected)
+        for graph in graphs.values():
+            graph.reset()
@@ -17,6 +17,8 @@
 from ignition.recipes.glm53.dcp import ContextShards
 from ignition.recipes.glm53.speculative import PreparedRound
 from ignition.recipes.glm53.mtp import DraftMetadata
+from ignition.recipes.glm53.stack import DecoderScratch
+from ignition.components.parallel import TensorParallelShard
 from ignition.recipes.glm53.state import DecoderMetadata
 from ignition.components.recurrent import DeltaMetadata
 
@@ -76,8 +78,12 @@ def test_verification_packs_unequal_lengths_and_retains_request_output_rows(
     assert op.metadata.recurrent.state_indices.tolist() == state_indices
 
 
-def test_draft_rows_exclude_exhausted_requests_and_restore_output_order():
-    metadata = DraftMetadata(3, 3, 2, "cpu")
+@pytest.mark.parametrize("reuse", [False, True])
+def test_draft_rows_exclude_exhausted_requests_and_restore_output_order(reuse):
+    scratch = DecoderScratch.allocate(3, 2, "prefill", "cpu", shard=TensorParallelShard(0, 2), sample_rows=1)
+    storage = scratch.attention_hc.normalized if reuse else None
+    metadata = DraftMetadata(3, 3, 2, "cpu", hidden=storage)
+    metadata.hidden.fill_(float("nan"))
     slots = torch.tensor([2, 0, 1])
     state_rows = torch.arange(1, 4)[:, None].expand(3, 8)
     tables = torch.tensor([[10, 11], [20, 21], [30, 31]])
@@ -89,13 +95,25 @@ def test_draft_rows_exclude_exhausted_requests_and_restore_output_order():
     assert metadata.indexed.token_slots.tolist() == [10 * 256 + 255, 21 * 256 + 254, -1]
     assert metadata.token_ids[:2].tolist() == [8, 9]
     assert metadata.hidden[metadata.output_indices][1:, 0].tolist() == [1, 2]
+    assert hidden[:, 0].tolist() == [0, 1, 2]
+    if reuse:
+        assert torch.equal(storage, metadata.hidden)
+        assert storage.data_ptr() == metadata.hidden.data_ptr()
     metadata.set(slots, torch.zeros(3, dtype=torch.int64), torch.full((3,), 512),
                  torch.tensor([7, 8, 9]), hidden, tables, capacity=1, state_rows=state_rows)
     assert metadata.num_tokens.item() == 0
+    assert not metadata.hidden.count_nonzero()
     assert metadata.indexed.positions.tolist() == [-1, -1, -1]
     assert metadata.indexed.token_slots.tolist() == [-1, -1, -1]
 
 
+@pytest.mark.parametrize("hidden", [torch.empty((2, 4096)), torch.empty((1, 4096), dtype=torch.bfloat16),
+    torch.empty((4096, 2), dtype=torch.bfloat16).t(), torch.empty((2, 4096), device="meta", dtype=torch.bfloat16)])
+def test_draft_hidden_storage_rejects_incompatible_views(hidden):
+    with pytest.raises(ValueError, match="draft hidden storage"):
+        DraftMetadata(2, 1, 2, "cpu", hidden=hidden)
+
+
 @pytest.fixture(autouse=True)
 def records_only_cpu_commit(monkeypatch):
     # These state-machine tests run without CUDA; reduction and graph ordering
 
@@ -59,6 +59,22 @@ def test_scratch_views_reject_unavailable_token_sample_and_request_capacity():
             base.rows(tokens, mode, sample_rows=rows)
 
 
+def test_cp_prefill_query_value_alias_survives_layer_overlay():
+    base = DecoderScratch.allocate(65, 2, "prefill", "cpu", shard=TensorParallelShard(0, 2),
+        sample_rows=1, context_ranks=2, overlay_layers=True)
+    for rows in (65, 16, 65):
+        view = base.rows(rows, "prefill", sample_rows=1).mla
+        view.query.fill_(float("nan"))
+        view.expanded.fill_(rows)
+        assert torch.all(view.query == rows)
+    for mode, ranks in (("prefill", 1), ("decode", 2)):
+        separate = DecoderScratch.allocate(2, 2, mode, "cpu", shard=TensorParallelShard(0, 2),
+            sample_rows=1, context_ranks=ranks).mla
+        separate.query.fill_(7)
+        separate.expanded.zero_()
+        assert torch.all(separate.query == 7)
+
+
 @pytest.mark.parametrize("world,rank", [(2, 0), (4, 3)])
 def test_canonical_layers_share_intermediates_and_own_distinct_state(world, rank):
     snapshot = Path(os.environ.get("FLARE_GLM53_SNAPSHOT",