diff --git a/cactus-engine/src/engine.h b/cactus-engine/src/engine.h
index d592be929..0225d792f 100644
--- a/cactus-engine/src/engine.h
+++ b/cactus-engine/src/engine.h
@@ -988,7 +988,7 @@ class Model {
     size_t component_chunk_tokens(const Component& comp, const std::string& input_name) const;
     size_t component_output_tokens(const Component& comp, const std::string& output_name) const;
     ChunkedPrefillResult run_chunked_prefill(const std::vector<uint32_t>& tokens, size_t start_position,
-                                             size_t chunk_size, bool prepare_decode);
+                                             size_t chunk_size, bool prepare_decode, bool produce_logits);
     void execute_prefill_chunk(Component& chunk_comp, Component* enc_comp, size_t encoder_chunk,
                                size_t chunk_tokens, const std::vector<uint32_t>& tokens,
                                size_t processed, size_t start_position);
@@ -1049,6 +1049,8 @@ class Model {
     Component* audio_encoder_ = nullptr;
     Component* lm_encoder_media_step_ = nullptr;
     Component* decoder_prefill_chunk_ = nullptr;
+    Component* decoder_prefill_cache_chunk_ = nullptr;
+    Component* decoder_prefill_logits_head_ = nullptr;
     Component* lm_encoder_ = nullptr;
     Component* lm_encoder_text_chunk_ = nullptr;
     Component* lm_encoder_media_chunk_ = nullptr;
diff --git a/cactus-engine/src/model.cpp b/cactus-engine/src/model.cpp
index 5383c89bf..1f19daaff 100644
--- a/cactus-engine/src/model.cpp
+++ b/cactus-engine/src/model.cpp
@@ -810,6 +810,12 @@ bool Model::init(const std::string& bundle_dir, size_t context_size,
     if (!decoder_cross_kv_name.empty()) decoder_cross_kv_ = &components_.at(decoder_cross_kv_name);
     if (components_.count("decoder_prefill_chunk")) {
         decoder_prefill_chunk_ = &components_.at("decoder_prefill_chunk");
+        decoder_prefill_cache_chunk_ = components_.count("decoder_prefill_cache_chunk")
+            ? &components_.at("decoder_prefill_cache_chunk")
+            : nullptr;
+        decoder_prefill_logits_head_ = components_.count("decoder_prefill_logits_head")
+            ? &components_.at("decoder_prefill_logits_head")
+            : nullptr;
         decoder_prefill_ = components_.count("decoder_prefill_text_chunk")
             ? &components_.at("decoder_prefill_text_chunk")
             : decoder_prefill_chunk_;
@@ -846,6 +852,8 @@ bool Model::init(const std::string& bundle_dir, size_t context_size,
         decoder_cross_kv_,
         decoder_prefill_,
         decoder_prefill_chunk_,
+        decoder_prefill_cache_chunk_,
+        decoder_prefill_logits_head_,
         vision_encoder_,
         vision_projector_,
         audio_encoder_,
@@ -2154,7 +2162,7 @@ void Model::reset_prefill_stats() {
     last_prefill_tail_padding_tokens_ = 0;
 }
 
-Model::ChunkedPrefillResult Model::run_chunked_prefill(const std::vector<uint32_t>& tokens, size_t start_position, size_t chunk_size, bool prepare_decode) {
+Model::ChunkedPrefillResult Model::run_chunked_prefill(const std::vector<uint32_t>& tokens, size_t start_position, size_t chunk_size, bool prepare_decode, bool produce_logits) {
     ChunkedPrefillResult result;
     reset_prefill_stats();
     if (decode_route_ != DecodeRoute::CACHED_STEP || !encoder_ || !decoder_ || !decoder_prefill_) return result;
@@ -2188,23 +2196,41 @@ Model::ChunkedPrefillResult Model::run_chunked_prefill(const std::vector<uint32_
             }
         }
     }
-    result.component = prefill_component;
+    Component* logits_component = prefill_component;
+    Component* cache_component = decoder_prefill_cache_chunk_;
+    Component* logits_head = decoder_prefill_logits_head_;
+    if (cache_component && declared_prefill_tokens(cache_component) != declared_prefill_tokens(logits_component)) {
+        cache_component = nullptr;
+        logits_head = nullptr;
+    }
     result.encoder_component = encoder_component;
     if (start_position != 0) {
         if (std::getenv("CACTUS_DISABLE_SHARED_STATE") != nullptr || !decoder_->graph) return result;
         publish_component_cache_states(*decoder_);
     }
-    if (!load_component_graph(*prefill_component)) return result;
+    // Start with the cache-only graph whenever at least one chunk does not
+    // need logits. The final chunk switches to the logits graph only when its
+    // last row will be sampled directly.
+    const size_t declared_tokens = declared_prefill_tokens(logits_component);
+    const bool exact_chunk_tail = declared_tokens > 1 && !tokens.empty() && tokens.size() % declared_tokens == 0;
+    const bool needs_only_logits_chunk = !logits_head
+        && produce_logits && exact_chunk_tail && tokens.size() <= declared_tokens;
+    Component* active_component = cache_component && !needs_only_logits_chunk
+        ? cache_component
+        : logits_component;
+    prefill_component = active_component;
+    result.component = active_component;
+    if (!load_component_graph(*active_component)) return result;
     if (encoder_component && !load_component_graph(*encoder_component)) return result;
-    if (!cache_states_compatible(*prefill_component, *decoder_)) return result;
-    size_t component_tokens = component_chunk_tokens(*prefill_component, "inputs_embeds");
+    if (!cache_states_compatible(*active_component, *decoder_)) return result;
+    size_t component_tokens = component_chunk_tokens(*active_component, "inputs_embeds");
     if (component_tokens <= 1) return result;
     size_t effective_chunk = chunk_size > 0 ? std::min(chunk_size, component_tokens) : component_tokens;
     if (effective_chunk != component_tokens) effective_chunk = component_tokens;
     size_t whole_chunks_end = (tokens.size() / effective_chunk) * effective_chunk;
     auto any_cache_node = [&](auto predicate) {
-        if (!prefill_component->graph) return false;
-        for (const auto& state : prefill_component->cache_states) {
+        if (!active_component->graph) return false;
+        for (const auto& state : active_component->cache_states) {
             for (int node_id : {state.key_node_id, state.value_node_id}) {
                 if (node_id < 0) continue;
                 if (predicate(static_cast<size_t>(node_id))) return true;
@@ -2213,17 +2239,17 @@ Model::ChunkedPrefillResult Model::run_chunked_prefill(const std::vector<uint32_
         return false;
     };
     const bool has_recurrent_state = any_cache_node([&](size_t id) {
-        return prefill_component->graph->get_node_op_type(id) == OpType::RECURRENT_CACHE_STATE;
+        return active_component->graph->get_node_op_type(id) == OpType::RECURRENT_CACHE_STATE;
     });
     if (has_recurrent_state && whole_chunks_end > effective_chunk) {
         whole_chunks_end = effective_chunk;
     }
     const bool has_sliding_window_cache = any_cache_node([&](size_t id) {
-        return prefill_component->graph->get_node_op_type(id) == OpType::KV_CACHE_STATE
-            && prefill_component->graph->get_node_window_size(id) > 0;
+        return active_component->graph->get_node_op_type(id) == OpType::KV_CACHE_STATE
+            && active_component->graph->get_node_window_size(id) > 0;
     });
     const bool has_conv_state = any_cache_node([&](size_t id) {
-        return prefill_component->graph->get_node_op_type(id) == OpType::CONV_CACHE_STATE;
+        return active_component->graph->get_node_op_type(id) == OpType::CONV_CACHE_STATE;
     });
     const size_t tail_tokens = tokens.size() - whole_chunks_end;
     const size_t padding_cutoff = std::max<size_t>(1, effective_chunk / 16);
@@ -2232,9 +2258,9 @@ Model::ChunkedPrefillResult Model::run_chunked_prefill(const std::vector<uint32_
         && !has_sliding_window_cache
         && tail_tokens >= padding_cutoff;
     const bool padded_window_too_small = any_cache_node([&](size_t id) {
-        if (prefill_component->graph->get_node_op_type(id) != OpType::KV_CACHE_STATE) return false;
-        size_t window = prefill_component->graph->get_node_window_size(id);
-        size_t sink = prefill_component->graph->get_node_sink_size(id);
+        if (active_component->graph->get_node_op_type(id) != OpType::KV_CACHE_STATE) return false;
+        size_t window = active_component->graph->get_node_window_size(id);
+        size_t sink = active_component->graph->get_node_sink_size(id);
         return window > 0 && (window < effective_chunk * 4 || window <= effective_chunk + sink);
     });
     const bool use_padded_tail = !pad_tail && !prefill_tail_pad_disabled_
@@ -2255,9 +2281,29 @@ Model::ChunkedPrefillResult Model::run_chunked_prefill(const std::vector<uint32_
         }
     }
 
+    auto activate_component = [&](Component* target, size_t logical_current) -> bool {
+        if (!target || target == active_component) return target != nullptr;
+        if (!load_component_graph(*target)) return false;
+        if (!cache_states_compatible(*active_component, *target)) return false;
+        move_cache_states(*active_component, *target, logical_current);
+        active_component->graph->release_runtime_buffers();
+        unload_component_graph(*active_component);
+        active_component = target;
+        prefill_component = target;
+        result.component = target;
+        return true;
+    };
+
     size_t processed = 0;
     while (processed + effective_chunk <= executable_tokens) {
-        execute_prefill_chunk(*prefill_component, encoder_component, encoder_chunk,
+        const bool final_exact_chunk = produce_logits
+            && tail_tokens == 0
+            && processed + effective_chunk == executable_tokens;
+        Component* target = final_exact_chunk && !logits_head
+            ? logits_component
+            : (cache_component ? cache_component : logits_component);
+        if (!activate_component(target, start_position + processed)) return ChunkedPrefillResult{};
+        execute_prefill_chunk(*active_component, encoder_component, encoder_chunk,
                               effective_chunk, tokens, processed, start_position);
         processed += effective_chunk;
     }
@@ -2268,18 +2314,21 @@ Model::ChunkedPrefillResult Model::run_chunked_prefill(const std::vector<uint32_
         const size_t pads = effective_chunk - tail_tokens;
         const size_t kept_real = tail_tokens - 1;
         std::vector<std::pair<size_t, std::vector<uint8_t>>> backups;
-        for (const auto& state : prefill_component->cache_states) {
+        if (!activate_component(cache_component ? cache_component : logits_component, start_position + processed)) {
+            return ChunkedPrefillResult{};
+        }
+        for (const auto& state : active_component->cache_states) {
             for (int node_id : {state.key_node_id, state.value_node_id}) {
                 if (node_id < 0) continue;
                 size_t id = static_cast<size_t>(node_id);
-                if (prefill_component->graph->get_node_op_type(id) != OpType::KV_CACHE_STATE) continue;
-                backups.emplace_back(id, prefill_component->graph->snapshot_cache_padded_append(id, kept_real, pads + 1));
+                if (active_component->graph->get_node_op_type(id) != OpType::KV_CACHE_STATE) continue;
+                backups.emplace_back(id, active_component->graph->snapshot_cache_padded_append(id, kept_real, pads + 1));
             }
         }
-        execute_prefill_chunk(*prefill_component, encoder_component, encoder_chunk,
+        execute_prefill_chunk(*active_component, encoder_component, encoder_chunk,
                               effective_chunk, tokens, processed, start_position);
         for (auto& [id, backup] : backups) {
-            prefill_component->graph->rollback_cache_padded_append(id, kept_real, pads + 1, backup);
+            active_component->graph->rollback_cache_padded_append(id, kept_real, pads + 1, backup);
         }
         processed += kept_real;
         tail_executed = kept_real;
@@ -2288,10 +2337,22 @@ Model::ChunkedPrefillResult Model::run_chunked_prefill(const std::vector<uint32_
 
     result.executed_tokens = processed;
     result.logical_tokens = std::min(tokens.size(), processed);
+    result.padding_tokens = processed > tokens.size() ? processed - tokens.size() : 0;
+    if (produce_logits && logits_head && cache_component
+        && result.logical_tokens == tokens.size() && result.padding_tokens == 0) {
+        if (!load_component_graph(*logits_head)) return ChunkedPrefillResult{};
+        copy_component_outputs_to_inputs(*active_component, *logits_head);
+        logits_head->graph->execute();
+        result.component = logits_head;
+    }
     if (result.logical_tokens > 0) {
-        result.last_logit_row = (result.logical_tokens - 1) % effective_chunk;
+        const size_t logits_rows = result.component
+            ? component_output_tokens(*result.component, "logits")
+            : 0;
+        result.last_logit_row = logits_rows == 1
+            ? 0
+            : (result.logical_tokens - 1) % effective_chunk;
     }
-    result.padding_tokens = processed > tokens.size() ? processed - tokens.size() : 0;
     result.scalar_tail_tokens = tokens.size() - result.logical_tokens;
     last_prefill_padding_tokens_ = result.padding_tokens;
     last_prefill_scalar_tail_tokens_ = result.scalar_tail_tokens;
@@ -2303,10 +2364,14 @@ Model::ChunkedPrefillResult Model::run_chunked_prefill(const std::vector<uint32_
             std::fill(decoder_->input_buffers[i].begin(), decoder_->input_buffers[i].end(), 0);
         }
         auto copy_start = std::chrono::high_resolution_clock::now();
-        move_cache_states(*prefill_component, *decoder_, start_position + result.logical_tokens);
+        move_cache_states(*active_component, *decoder_, start_position + result.logical_tokens);
         auto copy_end = std::chrono::high_resolution_clock::now();
         last_prefill_cache_copy_ms_ = std::chrono::duration_cast<std::chrono::microseconds>(copy_end - copy_start).count() / 1000.0;
     }
+    if (result.component == logits_head && active_component != logits_head) {
+        active_component->graph->release_runtime_buffers();
+        unload_component_graph(*active_component);
+    }
     return result;
 }
 
@@ -3198,7 +3263,7 @@ bool Model::prefill_and_sample_first_token(const std::vector<uint32_t>& tokens,
     }
     ChunkedPrefillResult chunked;
     if (decoder_prefill_) {
-        chunked = run_chunked_prefill(tokens, cache_total_seq_len_, get_prefill_chunk_size(), true);
+        chunked = run_chunked_prefill(tokens, cache_total_seq_len_, get_prefill_chunk_size(), true, true);
         if (chunked.logical_tokens == tokens.size() && chunked.padding_tokens > 0 && tokens.size() > 0) {
             // Cache already moved into the step component; drop the last row and re-run it for step-decoder logits.
             set_cache_current_len(*decoder_, tokens.size() - 1);
@@ -3250,7 +3315,7 @@ void Model::prefill(const std::vector<uint32_t>& tokens, size_t /*chunk_size*/,
         cache_token_ids_ = context_tokens_;
         return;
     }
-    ChunkedPrefillResult chunked = run_chunked_prefill(tokens, cache_total_seq_len_, get_prefill_chunk_size(), prepare_decode);
+    ChunkedPrefillResult chunked = run_chunked_prefill(tokens, cache_total_seq_len_, get_prefill_chunk_size(), prepare_decode, false);
     cache_total_seq_len_ += chunked.logical_tokens;
     if (prepare_decode && chunked.logical_tokens > 0) {
         if (chunked.component) unload_component_graph(*chunked.component);
diff --git a/python/cactus/transpiler/CONTRIBUTING.md b/python/cactus/transpiler/CONTRIBUTING.md
index 8e1fae46a..c90e47ede 100644
--- a/python/cactus/transpiler/CONTRIBUTING.md
+++ b/python/cactus/transpiler/CONTRIBUTING.md
@@ -1,45 +1,235 @@
-# Transpiler development
+# Contributing to Transpiler (The First Triumvirate)
 
-The transpiler turns an exported PyTorch graph into a Cactus runtime bundle in four stages:
+The Cactus transpiler turns a PyTorch model into a runnable Cactus bundle. Its consists of 3 key susbsystems:
 
-1. `Converter/` exports a model and serializes the graph as a `LayerMap`, this can be visualized through JSON and serves as the intermediary representation of the model.
-2. `IR/` normalizes the exported graph and applies safe graph fusions for cactus-native operations.
-3. `Generator/` splits the graph into ModelProfile specified components, and lowers each fused operation to Cactus calls.
-4. `RuntimePlan/` and `ModelProfiles/` describe component routes, persistent state, cache bindings, and metadata files.
+1. **Crassus — Converter + Model Profiles** Captures the source model and describes its model-level contract. Generates the intermediary representation of the model through LayerMap object, which is representable through a JSON.
+2. **Pompey — Fusions + IR** Defines fusions of basic .
+3. **Caesar — Generator + Runtime Plan** lowers the simplified computation into Cactus graphs and describes how their components execute together.
 
-Keep model-specific behavior in `ModelProfiles/` or a narrowly scoped matcher. Shared lowering and graph code must remain independent of Hugging Face model names!!!
+The CLI announces these phases as it runs:
 
-## Updating Converter Logic (Crassus)
+```text
+Running converter (Crassus vX.Y.Z)
+Running IR simplifier (Pompey vX.Y.Z)
+Running Generator (Caesar vX.Y.Z)
+```
 
-`ModelProfiles/profiles.py::MODEL_ID_MAP` is the boundary between optimized and
-generic conversion. An exact, case-insensitive registered model ID always uses
-its complete profile: input strategy, modalities, export patches, cache
-contract, fusion fields, component split, and runtime plan. Generic CLI options
-must not override a registered profile.
+Their versions live in `Converter/version.py`, `IR/version.py`, and
+`Generator/version.py` respectively.
 
-An unregistered model returns no profile from `profile_for_model_id`. The CLI
-then constructs a generic contract from `--task`, `--modalities`,
-`--cache-style`, and optional `--fusion-groups`. Never select a generic
-architecture from substrings in a repository name. To add an optimized model,
-create or reuse a profile and explicitly add every supported model ID to
-`MODEL_ID_MAP`.
+Keep model-specific policy in `ModelProfiles/` or in a narrowly scoped
+structural matcher. Shared conversion, IR, fusion, lowering, and runtime-plan
+code must not select behavior from Hugging Face repository-name substrings.
 
-## Adding or changing a fusion
+## Crassus: Converter and Model Profiles
 
-- Match structure and tensor semantics, not generated node names alone.
-- Validate ranks, axes, constants, cache mode, and source-module context before replacing a subgraph.
-- Preserve every externally consumed output of the matched region.
-- Add a focused matcher regression test and regenerate both prefill and decode graphs for every affected profile.
-- Run at least two complete simplify/fusion rounds over raw IR, then continue until the serialized graph is stable. Later fusion families and cleanup can expose patterns owned by an earlier family.
-- Compare the simplified operation census before and after the change. An unexpected remaining `bmm`, `softmax`, cache copy, or transpose usually means a pattern only partially matched.
+Crassus answers two questions:
 
-## Changing a model profile or runtime plan
+1. What model and inputs should PyTorch export?
+2. What capabilities and model-specific contracts should the remaining phases
+   apply?
+
+### Converter responsibilities
+
+`Converter/` loads the source model, prepares representative inputs, exports its
+PyTorch graph, and serializes that graph as a `LayerMap`. The raw JSON form is
+the inspectable boundary between source-model execution and Cactus graph
+optimization.
+
+Important files:
+
+- `Converter/convert.py` orchestrates graph export and writes raw `LayerMap`
+  JSON.
+- `Converter/models.py` defines the exported graph representation and model
+  loading paths.
+- `Converter/input_processor.py` creates text, image, and audio example inputs.
+- `Converter/overrides.py` contains narrowly scoped export-time compatibility
+  patches.
+- `Converter/cache_utils.py` interprets and annotates exported cache objects.
+- `Converter/constants.py` holds shared converter configuration.
+
+Converter changes should preserve source-model semantics. Do not add a rewrite
+here merely because it makes lowering easier; semantic graph simplification
+belongs to Pompey.
+
+### Model Profile responsibilities
+
+`ModelProfiles/` is the declarative model-policy layer. Profiles describe:
+
+- Supported modalities and sample-input strategies.
+- Model loading and export patches.
+- Prefill and decode inference modes.
+- Cache style and cache precision contracts.
+- Enabled and disabled fusion groups.
+- Component definitions and execution routes.
+- Prompt, media, persistent-state, and output-alias contracts.
+
+Important files:
+
+- `ModelProfiles/profiles.py` defines optimized profiles and `MODEL_ID_MAP`.
+- `ModelProfiles/models.py` defines profile and contract dataclasses.
+- `ModelProfiles/components.py` defines reusable component declarations.
+- `ModelProfiles/routes.py` defines execution patterns and routes.
+- `ModelProfiles/combinations.py` defines valid component combinations.
+
+`MODEL_ID_MAP` is the boundary between optimized and generic conversion. An
+exact, case-insensitive registered model ID uses its complete profile. Generic
+CLI flags must not override that profile.
+
+An unregistered model uses a generic contract built from `--task`,
+`--modalities`, `--cache-style`, and optional `--fusion-groups`. Never infer an
+optimized architecture from words in a repository name. To optimize a new
+model, create or reuse a profile and explicitly register every supported model
+ID.
+
+When changing Crassus:
+
+- Test every modality and inference mode declared by the profile.
+- Compare exported raw IR before and after the change.
+- Keep generic conversion functional for unregistered model IDs.
+- Confirm export changes do not alter PyTorch reference outputs unexpectedly.
+
+## Pompey: Fusions and IR
+
+Pompey turns the verbose exported graph into a smaller semantic graph that maps
+cleanly to Cactus operations.
+
+### IR responsibilities
+
+`IR/` owns graph structure, matching, normalization, simplification, dead-code
+removal, and repeated fusion passes.
+
+Important files:
+
+- `IR/models.py` defines graph, node, edge, value, and annotation structures.
+- `IR/simplify_ir.py` runs simplification and fusion passes to a stable result.
+- `IR/match.py` performs structural fusion-graph matching.
+- `IR/match_utils.py` provides reusable graph and tensor matching helpers.
+- `IR/extra_matchers/` implements semantic constraints that cannot be expressed
+  by graph shape alone.
+- `IR/special_fusions.py` handles complex structural replacements requiring
+  dedicated logic.
+
+Simplification must run repeatedly because one cleanup or fusion can expose a
+pattern consumed by an earlier pass. Run at least two complete rounds and
+continue until the serialized graph is stable or the configured pass limit is
+reached.
+
+### Fusion responsibilities
+
+`Fusions/` describes reusable source subgraphs and their replacement Cactus
+operations.
+
+Important files:
+
+- `Fusions/fusions.py` is the fusion registry and assigns fusion groups.
+- `Fusions/fusion_defs/` contains operation-family graph definitions.
+- `Fusions/fusion_cache_conv.py` contains cache and convolution definitions.
+- `Fusions/fusion_builders.py` provides helpers for constructing definitions.
+- `Fusions/models.py`, `nodes.py`, and `edges.py` define reusable fusion data
+  and graph fragments.
+
+When adding or modifying a fusion:
+
+- Match graph structure and tensor semantics, not generated node names alone.
+- Validate ranks, dimensions, axes, constants, layouts, cache modes, inference
+  modes, and source-module context.
+- Preserve every output consumed outside the replaced subgraph.
+- Keep generic fusion groups conservative; model-specific groups may be more
+  aggressive when a profile proves their preconditions.
+- Add a focused matcher regression test.
+- Regenerate both prefill and decode graphs for every affected profile.
+- Compare operation censuses before and after simplification. Unexpected `bmm`,
+  `softmax`, cache copies, reshapes, or transposes often indicate a partial
+  match.
+- Verify that another complete simplification pass makes no unintended change.
+
+Pompey decides **what computation a graph means**. It must not decide component
+residency, execution order, or cross-component state lifetime; those belong to
+Caesar.
+
+## Caesar: Generator and Runtime Plan
+
+Caesar converts the simplified semantic graph into native Cactus components and
+defines how those components cooperate at runtime.
+
+### Generator responsibilities
+
+`Generator/` splits IR graphs into components, lowers nodes to Cactus graph
+operations, binds converted weights, and writes component graph artifacts.
+
+Important files:
+
+- `Generator/generate.py` orchestrates component generation and bundle output.
+- `Generator/component_splits/` identifies component boundaries and extracts
+  component graphs.
+- `Generator/lowerings/` maps simplified IR operations to Cactus graph-builder
+  calls.
+- `Generator/models.py` defines generation results, component graphs, manifests,
+  and lowering rules.
+- `Generator/constants.py` and `errors.py` hold shared generator configuration
+  and lowering failures.
+
+Lowering should be mechanical. If a lowering function needs to rediscover a
+large decomposed pattern, add or improve a Pompey fusion instead. Unsupported
+semantic operations should remain visible as unsupported rather than silently
+lowering to an incorrect approximation.
+
+### Runtime Plan responsibilities
+
+`RuntimePlan/` converts generated component metadata and model-profile contracts
+into `runtime_plan.json` and `components/manifest.json`.
+
+`RuntimePlan/models.py` defines and serializes:
+
+- Components, graph paths, logical inputs, and logical outputs.
+- Weight bindings and cache-state node bindings.
+- Execution routes and component roles.
+- Runtime-owned persistent states.
+- Producer-to-consumer tensor aliases and copy fallbacks.
+- State lifetimes and last-consumer release metadata.
+- Cache movement, persistence, and ownership policies.
+
+When changing Caesar:
 
 - Declare component routes and persistent outputs explicitly.
-- Only retain outputs or cache entries that a later component consumes.
-- Keep reusable state in runtime-owned storage; component graphs should bind to it rather than own or copy it.
-- Regenerate the complete bundle and test every modality combination supported by the profile.
+- Retain only outputs and cache entries consumed by a later component.
+- Keep reusable cache and activation state in runtime-owned storage.
+- Prefer compatible `TensorStorage` aliasing over inter-component copies.
+- Always retain a correct copy fallback when storage cannot be aliased.
+- Ensure unloading a component cannot invalidate state required by a later one.
+- Confirm final component manifests reference valid graph nodes and weight files.
+- Test bundle loading, prefill, decode, reset, and every supported modality.
+
+Caesar decides **where computation lives and how it executes**. It should not
+redefine source-model semantics or compensate for missing IR fusions.
+
+## End-to-end contribution workflow
+
+Follow the data through all three members before deciding where a change
+belongs:
+
+```text
+PyTorch model
+    -> Crassus: raw LayerMap + model contract
+    -> Pompey: stable simplified/fused IR
+    -> Caesar: component graphs + runtime plan
+    -> Cactus engine execution
+```
+
+For every meaningful change:
 
-## Validation
+1. Inspect raw and simplified IR.
+2. Confirm the selected optimized or generic model profile.
+3. Compare operation and component censuses.
+4. Regenerate the complete model bundle from scratch.
+5. Load it through `cactus run` and test every supported modality.
+6. Verify output quality against the source model or a known-good bundle.
+7. Measure prefill throughput, decode throughput, peak and active RAM, component
+   load time, copied bytes, and aliased bytes.
+8. Compare performance with `main` using equivalent weights and backend flags.
 
-Run the transpiler and runtime-plan tests, compile all transpiler modules, regenerate all available model bundles, and execute the graph/runtime tests. Performance changes should be benchmarked to determine prefill throughput, decode thorughput, peak RAM usage, and model output quality.
+At minimum, compile all transpiler modules and run the converter, IR/fusion,
+generator/component-split, runtime-plan, graph, and engine tests relevant to the
+change. A successful transpilation is not sufficient: the generated bundle must
+load and execute correctly.
diff --git a/python/cactus/transpiler/Fusions/fusions.py b/python/cactus/transpiler/Fusions/fusions.py
index b71be4a27..b37859ecf 100644
--- a/python/cactus/transpiler/Fusions/fusions.py
+++ b/python/cactus/transpiler/Fusions/fusions.py
@@ -79,6 +79,14 @@
         supported_inference_modes=("decode_with_cache",),
         metadata={"special_matcher": "generic_cached_attention"},
     ),
+    "generic_gqa_attention": _definition(
+        "generic_gqa_attention",
+        "attention_cached",
+        GENERIC_CACHED_ATTENTION_GRAPH,
+        fusion_fields=("generic_gqa_attention",),
+        supported_inference_modes=("prefill_with_cache",),
+        metadata={"special_matcher": "generic_cached_attention"},
+    ),
     "gemma4_bmm_masked_attention": _definition(
         "gemma4_bmm_masked_attention",
         "attention",
diff --git a/python/cactus/transpiler/Generator/component_splits/component_split_boundaries.py b/python/cactus/transpiler/Generator/component_splits/component_split_boundaries.py
index d29c02636..604f3e025 100644
--- a/python/cactus/transpiler/Generator/component_splits/component_split_boundaries.py
+++ b/python/cactus/transpiler/Generator/component_splits/component_split_boundaries.py
@@ -4,6 +4,7 @@
 
 from .component_split_types import Gemma4Boundaries, LfmVlmBoundaries, OutputSpec, WhisperBoundaries
 from .component_split_utils import tensor_last_dim, tensor_rank, tensor_shape
+from .. import constants
 from ...Fusions import models as FModels
 from ...IR import models as IRModels
 
@@ -294,6 +295,26 @@ def find_logits_output(graph: IRModels.Graph) -> str:
             return node.name
     raise ValueError("Gemma4 component split could not find logits output")
 
+def find_logits_projection_input(graph: IRModels.Graph) -> str:
+    """Return the final hidden activation consumed by the vocabulary head."""
+    logits = graph.nodes_map[find_logits_output(graph)]
+    vocabulary_width = tensor_last_dim(logits)
+    queue = [logits]
+    visited: set[str] = set()
+    while queue:
+        node = queue.pop(0)
+        if node.name in visited:
+            continue
+        visited.add(node.name)
+        if (
+            node.target in constants.MATMUL_TARGETS | constants.LINEAR_TARGETS
+            and len(node.parents) >= 2
+            and tensor_last_dim(node) == vocabulary_width
+        ):
+            return node.parents[0].name
+        queue.extend(node.parents)
+    raise ValueError("Causal LM component split could not find the logits projection input")
+
 def graph_output_specs(graph: IRModels.Graph, publish_only_logits: bool = False) -> tuple[OutputSpec, ...]:
     refs = graph_output_refs(graph)
     if not refs:
diff --git a/python/cactus/transpiler/Generator/component_splits/component_split_builder.py b/python/cactus/transpiler/Generator/component_splits/component_split_builder.py
index 0817f6367..432ed1aa5 100644
--- a/python/cactus/transpiler/Generator/component_splits/component_split_builder.py
+++ b/python/cactus/transpiler/Generator/component_splits/component_split_builder.py
@@ -47,6 +47,9 @@ def extract_component_graph(spec: ComponentSplitSpec) -> IRModels.Graph:
             final_output_name = add_permuted_output_node(nodes, final_output_name, output.logical_name, output.permutation, used_names)
         if output.row_limit is not None:
             final_output_name = add_row_limited_output_node(nodes, output_name, output.logical_name, output.row_limit, used_names)
+        if output.tail_rows is not None:
+            final_output_name = add_tail_rows_output_node(
+                nodes, final_output_name, output.logical_name, output.tail_rows, used_names)
         output_logical_names[final_output_name] = output.logical_name
         final_output_names.append(final_output_name)
     nodes.append(create_output_node(spec.name, tuple(final_output_names), nodes))
@@ -129,6 +132,51 @@ def add_row_limited_output_node(
     )
     return name
 
+def add_tail_rows_output_node(
+    nodes: list[IRModels.Node],
+    source_name: str,
+    logical_name: str,
+    row_count: int,
+    used_names: set[str],
+) -> str:
+    source = next((node for node in nodes if node.name == source_name), None)
+    if source is None:
+        raise ValueError(f"Component split cannot add tail output for missing node {source_name}")
+    source_shape = tensor_shape(source)
+    axis = sequence_axis(source_shape)
+    if axis is None or not isinstance(source_shape[axis], int) or source_shape[axis] < row_count:
+        raise ValueError(f"{source_name}: tail output requires a known sequence shape, got {source_shape}")
+    output_shape = list(source_shape)
+    start = int(output_shape[axis]) - row_count
+    output_shape[axis] = row_count
+    name = unique_component_node_name(f"{source_name}_last_{row_count}", used_names)
+    source_meta = source.tensor_output_meta if isinstance(source.tensor_output_meta, dict) else {}
+    nodes.append(
+        IRModels.Node(
+            index=max((node.index for node in nodes), default=-1) + 1,
+            name=name,
+            node_type="call_function",
+            target="cactus.slice",
+            args=[{"node": source_name}],
+            kwargs={},
+            users=(),
+            tensor_output_meta={**source_meta, "shape": output_shape},
+            module_stack=source.module_stack,
+            value_kind=FModels.ValueKind.ACTIVATION,
+            attrs={"axis": axis, "start": start, "length": row_count, "step": 1},
+            ir_metadata={"logical_output": logical_name},
+            cache=None,
+        )
+    )
+    return name
+
+def sequence_axis(shape: list[Any]) -> int | None:
+    if len(shape) >= 3 and shape[0] == 1:
+        return 1
+    if len(shape) >= 2:
+        return 0
+    return None
+
 def retarget_chunk_graph_sequence_length(graph: IRModels.Graph, chunk_tokens: int) -> IRModels.Graph:
     source_lengths = chunk_source_lengths(graph)
     if not source_lengths:
@@ -379,6 +427,13 @@ def create_placeholder_node(
     metadata["logical_input"] = placeholder.logical_name
     if placeholder.name in output_logical_names:
         metadata["logical_output"] = output_logical_names[placeholder.name]
+    tensor_output_meta = template.tensor_output_meta if template is not None else None
+    if placeholder.sequence_tokens is not None and isinstance(tensor_output_meta, dict):
+        shape = list(tensor_output_meta.get("shape") or ())
+        axis = sequence_axis(shape)
+        if axis is not None:
+            shape[axis] = placeholder.sequence_tokens
+            tensor_output_meta = {**tensor_output_meta, "shape": shape}
     return IRModels.Node(
         index=template.index if template is not None else -1,
         name=name,
@@ -387,7 +442,7 @@ def create_placeholder_node(
         args=[],
         kwargs={},
         users=(),
-        tensor_output_meta=template.tensor_output_meta if template is not None else None,
+        tensor_output_meta=tensor_output_meta,
         module_stack=template.module_stack if template is not None else None,
         value_kind=placeholder.value_kind,
         attrs={},
diff --git a/python/cactus/transpiler/Generator/component_splits/component_split_plans.py b/python/cactus/transpiler/Generator/component_splits/component_split_plans.py
index 5a2f8c60b..6f0e2f351 100644
--- a/python/cactus/transpiler/Generator/component_splits/component_split_plans.py
+++ b/python/cactus/transpiler/Generator/component_splits/component_split_plans.py
@@ -177,6 +177,9 @@ def split_causal_lm_components(
     step_position = PlaceholderSpec("position_ids", "position_ids", tensor_node="input_ids", force=True)
     chunk_position = PlaceholderSpec("position_ids", "position_ids", tensor_node="input_ids", force=True)
     chunk_tokens = causal_lm_prefill_chunk_tokens(model_profile)
+    use_generic_chunk_attention = "generic_gqa_attention" in tuple(
+        getattr(model_profile, "fusion_fields", ()) or ())
+    logits_hidden = find_logits_projection_input(prefill) if use_generic_chunk_attention else None
     step_specs = (
         ComponentSplitSpec(
             name="lm_encoder_step",
@@ -196,6 +199,30 @@ def split_causal_lm_components(
             input_aliases={"input_ids": "inputs_embeds"},
         ),
     )
+    cache_only_specs = (
+        ComponentSplitSpec(
+            name="decoder_prefill_cache_chunk",
+            graph=prefill,
+            outputs=(OutputSpec(logits_hidden, "last_hidden_state", tail_rows=1),),
+            side_effects=cache_side_effect_node_names(prefill),
+            placeholders=(chunk_position,),
+            input_aliases={"input_ids": "inputs_embeds"},
+            metadata={"prefill_role": "cache_only"},
+            chunk_tokens=chunk_tokens,
+        ),
+        ComponentSplitSpec(
+            name="decoder_prefill_logits_head",
+            graph=prefill,
+            outputs=(OutputSpec(find_logits_output(prefill), "logits"),),
+            placeholders=(PlaceholderSpec(
+                "last_hidden_state",
+                "last_hidden_state",
+                source_node=logits_hidden,
+                sequence_tokens=1,
+            ),),
+            metadata={"prefill_role": "logits_head"},
+        ),
+    ) if use_generic_chunk_attention else ()
     prefill_specs = (
         ComponentSplitSpec(
             name="lm_encoder_text_prefill_chunk",
@@ -207,6 +234,7 @@ def split_causal_lm_components(
             placeholders=(chunk_position,),
             chunk_tokens=chunk_tokens,
         ),
+        *cache_only_specs,
         ComponentSplitSpec(
             name="decoder_prefill_chunk",
             graph=prefill,
@@ -387,5 +415,9 @@ def cache_side_effect_node_names(graph: IRModels.Graph) -> tuple[str, ...]:
             "cactus.conv_cache_append",
             "cactus.conv_cache_initialize",
             "cactus.recurrent_cache_write",
+            # Native cached attention owns the append when structural cache
+            # concatenation has been fused. It must remain in cache-only
+            # components even when no activation output is published.
+            "cactus.attention_cached",
         }
     )
diff --git a/python/cactus/transpiler/Generator/component_splits/component_split_types.py b/python/cactus/transpiler/Generator/component_splits/component_split_types.py
index 3a00f3016..9b36a1dc9 100644
--- a/python/cactus/transpiler/Generator/component_splits/component_split_types.py
+++ b/python/cactus/transpiler/Generator/component_splits/component_split_types.py
@@ -10,13 +10,10 @@
 DECODE_WITH_CACHE_TASK = "decode_with_cache"
 GEMMA4_TEXT_PREFILL_CHUNK_TOKENS = 32
 GEMMA4_MEDIA_PREFILL_CHUNK_TOKENS = 128
-# LFM-VLM has convolutional cache state, so a partial fixed-size chunk cannot
-# be tail-padded and rolled back like a pure KV model. Keep a small media graph
-# for typical image prompts, alongside a wide graph for long text prefill.
 LFM_VLM_MEDIA_PREFILL_CHUNK_TOKENS = 16
 LFM_VLM_TEXT_PREFILL_CHUNK_TOKENS = 128
 LFM_MOE_PREFILL_CHUNK_TOKENS = 128
-GENERIC_CAUSAL_PREFILL_CHUNK_TOKENS = 4
+GENERIC_CAUSAL_PREFILL_CHUNK_TOKENS = 64
 
 @dataclass(slots=True, frozen=True)
 class PlaceholderSpec:
@@ -27,12 +24,14 @@ class PlaceholderSpec:
     target: str | None = None
     value_kind: str = FModels.ValueKind.USER_INPUT
     force: bool = False
+    sequence_tokens: int | None = None
 
 @dataclass(slots=True, frozen=True)
 class OutputSpec:
     node: str
     logical_name: str
     row_limit: int | None = None
+    tail_rows: int | None = None
     permutation: tuple[int, ...] | None = None
     publish: bool = True
 
diff --git a/python/cactus/transpiler/Generator/lowerings/lowering_cache.py b/python/cactus/transpiler/Generator/lowerings/lowering_cache.py
index b97dc5bab..5edd2ab1e 100644
--- a/python/cactus/transpiler/Generator/lowerings/lowering_cache.py
+++ b/python/cactus/transpiler/Generator/lowerings/lowering_cache.py
@@ -198,12 +198,14 @@ def lower_attention(context: models.GenerationContext, node: IRModels.Node) -> A
         require_len(node, inputs, 5)
         query, key_new, value_new = cached_attention_inputs_for_layout(context, node, inputs)
         mask = cached_attention_mask(context, node, inputs)
+        key_cache = cached_attention_state_input(context, node, inputs, 3)
+        value_cache = cached_attention_state_input(context, node, inputs, 4)
         output = context.graph.attention_cached(
             query,
             key_new,
             value_new,
-            inputs[3],
-            inputs[4],
+            key_cache,
+            value_cache,
             scale=attention_scale(node),
             position_offset=cached_attention_position_offset(context, node),
             window_size=int(node.attrs.get("window_size", 0)),
@@ -215,6 +217,26 @@ def lower_attention(context: models.GenerationContext, node: IRModels.Node) -> A
         return attention_output_for_layout(context, node, output)
     raise UnsupportedLoweringError(f"{node.name}: unsupported attention target {node.target}")
 
+def cached_attention_state_input(
+    context: models.GenerationContext,
+    node: IRModels.Node,
+    inputs: tuple[Any, ...],
+    parent_index: int,
+) -> Any:
+    """Resolve prefill cache concatenations to their persistent native state.
+
+    A prefill concatenation deliberately lowers to its activation passthrough
+    for ordinary tensor consumers while recording the newly created cache
+    state out of band. A structurally fused ``attention_cached`` consumes the
+    state itself, so recover it from that table instead of forwarding the
+    activation value as if it contained cache metadata.
+    """
+    if parent_index < len(node.parents):
+        state = context.prefill_cache_cat_states.get(node.parents[parent_index].name)
+        if state is not None:
+            return state
+    return inputs[parent_index]
+
 def cached_attention_inputs_for_layout(context: models.GenerationContext, node: IRModels.Node, inputs: tuple[Any, ...]) -> tuple[Any, Any, Any]:
     if node.attrs.get("input_layout") == "bhqd_bhsd_bhsd":
         return (
diff --git a/python/cactus/transpiler/IR/special_fusions.py b/python/cactus/transpiler/IR/special_fusions.py
index dc7964898..a3c61e658 100644
--- a/python/cactus/transpiler/IR/special_fusions.py
+++ b/python/cactus/transpiler/IR/special_fusions.py
@@ -149,19 +149,31 @@ def match_generic_cached_attention(
     input_modalities: tuple[str, ...],
     fusion_fields: tuple[str, ...],
 ) -> models.FusionResult | None:
-    """Fuse exported causal attention by its typed cache/dataflow contract."""
-    if inference_mode != "decode_with_cache":
+    """Fuse exported causal GQA/MQA by its typed cache/dataflow contract."""
+    if inference_mode == "prefill_with_cache":
+        if fusion.target != "cactus.attention_cached":
+            return None
+        return match_generic_prefill_gqa(source, fusion)
+    if inference_mode != "decode_with_cache" or fusion.target != "cactus.attention_cached":
         return None
-    # Anchor on the first rank-4 layout restoration after the value BMM.  This
-    # avoids absorbing projection/output-layout operations that follow it.
-    if source.target not in {"cactus.view", "aten.view.default", "aten.reshape.default"}:
+    # Anchor on an output-layout wrapper after the value BMM.  Generic exports
+    # may restore [B,H,T,D], [B,T,H*D], or flattened [T,H*D] before the output
+    # projection; stopping at wrappers prevents absorbing that projection.
+    if source.target not in ATTENTION_WRAPPER_TARGETS:
         return None
-    if len(source.parents) != 1 or source.parents[0].target != "aten.bmm.default":
+    value_bmm, _ = value_bmm_from_unary_path(source, max_depth=10)
+    if value_bmm is None:
         return None
-    value_bmm = source.parents[0]
-    if len(value_bmm.parents) < 2 or not has_softmax_ancestor(value_bmm.parents[0]):
+    if len(value_bmm.parents) < 2:
         return None
-    qk_bmm = first_bmm_ancestor(value_bmm.parents[0], value_bmm)
+    softmax = first_target_ancestor(
+        value_bmm.parents[0],
+        {"cactus.softmax", "aten.softmax.int", "aten._softmax.default"},
+        max_depth=32,
+    )
+    if softmax is None or not softmax.parents:
+        return None
+    qk_bmm = first_bmm_ancestor(softmax.parents[0], value_bmm)
     if qk_bmm is None or len(qk_bmm.parents) < 2:
         return None
     key_match: models.CacheConcatMatch | None = None
@@ -219,20 +231,126 @@ def match_generic_cached_attention(
             "position_offset": HISTORY_POSITION_OFFSET,
             "window_size": int(key_cache.window_size or 0),
             "input_layout": "bhqd_bhsd_bhsd",
-            "output_layout": "bhqd",
+            "output_layout": "bthd_flat" if node_rank(source) in {2, 3} else "bhqd",
             "v_head_dim": last_int_dim(value_match.new_value),
+            "additive_mask": False,
             "dropped_mask_builder": True,
+            "preserved_prefill_mask": False,
             "dropped_gqa_repeat": True,
             "cache_contract": "typed_kv_concat",
         },
     )
 
-def has_softmax_ancestor(node: models.Node, max_depth: int = 32) -> bool:
-    return first_target_ancestor(
-        node,
+def match_generic_prefill_gqa(
+    source: models.Node,
+    fusion: FModels.FusionDefinition,
+) -> models.FusionResult | None:
+    """Fuse initial-cache prefill GQA into native cache-backed attention."""
+    if source.target not in ATTENTION_WRAPPER_TARGETS:
+        return None
+    value_bmm, _ = value_bmm_from_unary_path(source, max_depth=10)
+    if value_bmm is None or len(value_bmm.parents) < 2:
+        return None
+    softmax = first_target_ancestor(
+        value_bmm.parents[0],
         {"cactus.softmax", "aten.softmax.int", "aten._softmax.default"},
-        max_depth=max_depth,
-    ) is not None
+        max_depth=32,
+    )
+    if softmax is None or not softmax.parents:
+        return None
+    qk_bmm, mask = generic_prefill_qk_and_mask(softmax, value_bmm)
+    if qk_bmm is None or len(qk_bmm.parents) < 2:
+        return None
+    key_cat: models.Node | None = None
+    query_path: models.Node | None = None
+    for parent_index, parent in enumerate(qk_bmm.parents[:2]):
+        candidate = prefill_cache_cat_ancestor(parent, max_depth=40)
+        if candidate is None:
+            continue
+        key_cat = candidate
+        query_path = qk_bmm.parents[1 - parent_index]
+        break
+    value_cat = prefill_cache_cat_ancestor(value_bmm.parents[1], max_depth=40)
+    if key_cat is None or value_cat is None or query_path is None:
+        return None
+    key_new = prefill_cache_new_value(key_cat)
+    value_new = prefill_cache_new_value(value_cat)
+    query = generic_unscaled_bqhd_attention_input(query_path, max_depth=24)
+    if key_new is None or value_new is None or query is None:
+        return None
+    if not generic_cached_attention_shapes_match(query, key_new, value_new):
+        return None
+    # The cache concatenations lower to graph-independent KV state plus the
+    # native append. Passing the unappended K/V rows separately lets the
+    # cached-attention kernel use FP16 for the current chunk while reading the
+    # quantized/retained history directly from that state.
+    external_inputs = (query, key_new, value_new, key_cat, value_cat)
+    matched_nodes = attention_internal_nodes(source, external_inputs)
+    matched_names = {node.name for node in matched_nodes}
+    if qk_bmm.name not in matched_names or value_bmm.name not in matched_names:
+        return None
+    return models.FusionResult.from_match(
+        fusion=fusion,
+        source=source,
+        matched_nodes=matched_nodes,
+        external_inputs=external_inputs,
+        attrs={
+            "scale": decomposed_attention_scale(qk_bmm),
+            "is_causal": True,
+            "position_offset": HISTORY_POSITION_OFFSET,
+            "window_size": int(key_cat.cache.window_size or 0) if key_cat.cache is not None else 0,
+            "input_layout": "bhqd_bhsd_bhsd",
+            "output_layout": "bthd_flat" if node_rank(source) in {2, 3} else "bhqd",
+            "v_head_dim": last_int_dim(value_new),
+            # The exported mask has the captured T×T shape and cannot express
+            # later chunk history. Native cached causal attention derives the
+            # growing prefix from cache length. Runtime padding is right-sided,
+            # so padded queries cannot affect earlier real-token outputs and
+            # their appended cache rows are rolled back after the chunk.
+            "additive_mask": False,
+            "dropped_mask_builder": True,
+            "preserved_prefill_mask": False,
+            "dropped_gqa_repeat": True,
+            "cache_contract": "native_prefill_kv",
+        },
+    )
+
+def prefill_cache_new_value(cache_cat: models.Node) -> models.Node | None:
+    nonempty = [
+        parent for parent in cache_cat.parents[:2]
+        if element_count(models.tensor_shape(parent)) != 0
+    ]
+    return nonempty[0] if len(nonempty) == 1 else None
+
+def generic_prefill_qk_and_mask(
+    softmax: models.Node,
+    value_bmm: models.Node,
+) -> tuple[models.Node | None, models.Node | None]:
+    """Return QK and an optional safe, native-compatible prefill mask.
+
+    Decode can discard the exported causal mask because the cache position
+    defines the only visible prefix.  Prefill may contain padded rows, so only
+    fuse a direct QK path or an additive QK + rank-3/4 mask and preserve that
+    mask as the sixth native attention input.
+    """
+    logits = softmax.parents[0]
+    if logits.target not in {"cactus.add", "aten.add.Tensor"}:
+        return first_bmm_ancestor(logits, value_bmm), None
+    if len(logits.parents) != 2:
+        return None, None
+    qk: models.Node | None = None
+    mask: models.Node | None = None
+    for parent in logits.parents:
+        candidate = first_bmm_ancestor(parent, value_bmm)
+        if candidate is not None:
+            if qk is not None:
+                return None, None
+            qk = candidate
+        else:
+            if mask is not None or node_rank(parent) not in {3, 4}:
+                return None, None
+            mask = parent
+    return (qk, mask) if qk is not None else (None, None)
 
 def generic_cached_attention_shapes_match(
     query: models.Node,
@@ -250,7 +368,11 @@ def generic_cached_attention_shapes_match(
         return False
     if query_shape[2] != key_shape[2]:
         return False
-    return all(isinstance(dim, int) and dim > 0 for dim in (*query_shape, *key_shape))
+    if not all(isinstance(dim, int) and dim > 0 for dim in (*query_shape, *key_shape, value_shape[-1])):
+        return False
+    query_heads = int(query_shape[1])
+    kv_heads = int(key_shape[1])
+    return query_heads >= kv_heads and query_heads % kv_heads == 0
 
 def lfm_attention_matched_nodes(
     bindings: dict[str, models.Node],
diff --git a/python/cactus/transpiler/ModelProfiles/profiles.py b/python/cactus/transpiler/ModelProfiles/profiles.py
index 3e047f4f3..691b2a653 100644
--- a/python/cactus/transpiler/ModelProfiles/profiles.py
+++ b/python/cactus/transpiler/ModelProfiles/profiles.py
@@ -39,7 +39,7 @@ def decoder_kv_state(producer: str = "decoder_prefill_chunk") -> StateContract:
 TEXT_RUNTIME_CONTRACT = RuntimeContract(
     plan_name="generic_text",
     states=(
-        decoder_kv_state(),
+        decoder_kv_state("decoder_prefill_cache_chunk,decoder_prefill_chunk"),
     ),
     aliases=(
         AliasContract(
@@ -49,6 +49,20 @@ def decoder_kv_state(producer: str = "decoder_prefill_chunk") -> StateContract:
             target_input="inputs_embeds",
             required=False,
         ),
+        AliasContract(
+            source_component="text_embedding",
+            source_output="inputs_embeds",
+            target_component="decoder_prefill_cache_chunk",
+            target_input="inputs_embeds",
+            required=False,
+        ),
+        AliasContract(
+            source_component="decoder_prefill_cache_chunk",
+            source_output="last_hidden_state",
+            target_component="decoder_prefill_logits_head",
+            target_input="last_hidden_state",
+            required=False,
+        ),
     ),
 )
 
@@ -631,10 +645,24 @@ def generic_profile_for_contract(contract: GenericTranspileContract) -> ModelPro
         )
     default_groups = tuple(base.fusion_fields)
     fusion_groups = tuple(dict.fromkeys(contract.fusion_groups or default_groups))
-    if cache_style == GENERIC_CACHE_DYNAMIC_KV and "generic_cached_attention" not in fusion_groups:
-        fusion_groups = (*fusion_groups, "generic_cached_attention")
+    generic_cache_fusions = ("generic_cached_attention", "generic_gqa_attention")
+    if cache_style == GENERIC_CACHE_DYNAMIC_KV:
+        base = replace(
+            base,
+            cache_policy=tuple(
+                policy for policy in base.cache_policy if policy != "scalar_prefill"
+            ),
+            cache_contract=replace(
+                base.cache_contract,
+                fp16_kv_cache_components=tuple(dict.fromkeys((
+                    *base.cache_contract.fp16_kv_cache_components,
+                    "decoder_prefill_cache_chunk",
+                ))),
+            ),
+        )
+        fusion_groups = tuple(dict.fromkeys((*fusion_groups, *generic_cache_fusions)))
     elif cache_style != GENERIC_CACHE_DYNAMIC_KV:
-        fusion_groups = tuple(group for group in fusion_groups if group != "generic_cached_attention")
+        fusion_groups = tuple(group for group in fusion_groups if group not in generic_cache_fusions)
     return replace(
         base,
         fusion_fields=fusion_groups,
diff --git a/python/tests/test_generator_component_split.py b/python/tests/test_generator_component_split.py
index c06306bdc..b32147c94 100644
--- a/python/tests/test_generator_component_split.py
+++ b/python/tests/test_generator_component_split.py
@@ -7,6 +7,87 @@
 
 
 class TestGeneratorComponentSplit(unittest.TestCase):
+    def test_cache_only_component_keeps_cached_attention_and_prunes_logits(self) -> None:
+        def node(index, name, target, parents=(), shape=(1, 4, 8), kind=FModels.ValueKind.ACTIVATION):
+            return IRModels.Node(
+                index=index, name=name,
+                node_type="placeholder" if not parents else "call_function",
+                target=target, args=[{"node": parent} for parent in parents], kwargs={}, users=(),
+                tensor_output_meta={"shape": list(shape), "dtype": "torch.float16"},
+                module_stack=None, value_kind=kind,
+            )
+
+        nodes = (
+            node(0, "query", "query"),
+            node(1, "key", "key"),
+            node(2, "value", "value"),
+            node(3, "key_cache", "key_cache"),
+            node(4, "value_cache", "value_cache"),
+            node(5, "attention", "cactus.attention_cached", ("query", "key", "value", "key_cache", "value_cache")),
+            node(6, "lm_head_weight", "lm_head_weight", shape=(32, 8), kind=FModels.ValueKind.PARAMETER),
+            node(7, "logits", "cactus.linear", ("attention", "lm_head_weight"), shape=(1, 4, 32)),
+            IRModels.Node(
+                index=8, name="output", node_type="output", target="output",
+                args=[[{"node": "logits"}]], kwargs={}, users=(), tensor_output_meta=None,
+                module_stack=None, value_kind=FModels.ValueKind.OUTPUT,
+            ),
+        )
+        graph = IRModels.rebuild_graph(nodes, empty_graph())
+        spec = component_split.ComponentSplitSpec(
+            name="decoder_prefill_cache_chunk",
+            graph=graph,
+            outputs=(),
+            side_effects=("attention",),
+        )
+
+        component = component_split.extract_component_graph(spec)
+
+        self.assertIn("attention", component.nodes_map)
+        self.assertNotIn("logits", component.nodes_map)
+        self.assertNotIn("lm_head_weight", component.nodes_map)
+        self.assertEqual(component.outputs[0].args, [[]])
+
+    def test_cache_component_feeds_one_token_logits_head(self) -> None:
+        def node(index, name, target, parents=(), shape=(1, 4, 8), kind=FModels.ValueKind.ACTIVATION):
+            return IRModels.Node(
+                index=index, name=name,
+                node_type="placeholder" if not parents else "call_function",
+                target=target, args=[{"node": parent} for parent in parents], kwargs={}, users=(),
+                tensor_output_meta={"shape": list(shape), "dtype": "torch.float16"},
+                module_stack=None, value_kind=kind,
+            )
+
+        nodes = (
+            node(0, "hidden", "hidden"),
+            node(1, "lm_head_weight", "lm_head_weight", shape=(32, 8), kind=FModels.ValueKind.PARAMETER),
+            node(2, "logits", "cactus.linear", ("hidden", "lm_head_weight"), shape=(1, 4, 32)),
+            IRModels.Node(
+                index=3, name="output", node_type="output", target="output",
+                args=[[{"node": "logits"}]], kwargs={}, users=(), tensor_output_meta=None,
+                module_stack=None, value_kind=FModels.ValueKind.OUTPUT,
+            ),
+        )
+        graph = IRModels.rebuild_graph(nodes, empty_graph())
+        cache = component_split.extract_component_graph(component_split.ComponentSplitSpec(
+            name="decoder_prefill_cache_chunk",
+            graph=graph,
+            outputs=(component_split.OutputSpec("hidden", "last_hidden_state", tail_rows=1),),
+        ))
+        head = component_split.extract_component_graph(component_split.ComponentSplitSpec(
+            name="decoder_prefill_logits_head",
+            graph=graph,
+            outputs=(component_split.OutputSpec("logits", "logits"),),
+            placeholders=(component_split.PlaceholderSpec(
+                "last_hidden_state", "last_hidden_state", source_node="hidden", sequence_tokens=1),),
+        ))
+
+        cache_output = cache.nodes_map[cache.outputs[0].args[0][0]["node"]]
+        self.assertEqual(component_split.tensor_shape(cache_output), [1, 1, 8])
+        self.assertNotIn("lm_head_weight", cache.nodes_map)
+        self.assertEqual(component_split.tensor_shape(head.nodes_map["last_hidden_state"]), [1, 1, 8])
+        self.assertIn("lm_head_weight", head.nodes_map)
+        self.assertNotIn("hidden", head.nodes_map)
+
     def test_chunk_retarget_preserves_expand_broadcast_factor(self) -> None:
         input_ids = IRModels.Node(
             index=0,
diff --git a/python/tests/test_transpiler_reporting.py b/python/tests/test_transpiler_reporting.py
index a699d1e9a..640d62777 100644
--- a/python/tests/test_transpiler_reporting.py
+++ b/python/tests/test_transpiler_reporting.py
@@ -14,6 +14,7 @@
 from cactus.transpiler.IR import simplify_ir, special_fusions
 from cactus.transpiler.Generator import lowering_utils
 from cactus.transpiler.Generator import lowering_basic_ops
+from cactus.transpiler.Generator.lowerings import lowering_cache
 from cactus.transpiler.ModelProfiles.profiles import GENERIC_TEXT_PROFILE
 from cactus.cli import transpiler as cli_transpiler
 from cactus.cli import convert as cli_convert
@@ -22,6 +23,17 @@
 
 class TestTranspilerReporting(unittest.TestCase):
 
+    def test_prefill_cached_attention_resolves_persistent_cat_state(self):
+        state = object()
+        context = mock.Mock(prefill_cache_cat_states={"key_cat": state})
+        parents = [mock.Mock() for _ in range(4)]
+        for parent, name in zip(parents, ("q", "k", "v", "key_cat"), strict=True):
+            parent.name = name
+        node = mock.Mock(parents=parents)
+        inputs = (object(), object(), object(), object())
+
+        self.assertIs(lowering_cache.cached_attention_state_input(context, node, inputs, 3), state)
+
     def test_nonfinite_attention_mask_constants_survive_ir_json_roundtrip(self):
         record = CModels.LayerRecord(
             index=0,
@@ -160,7 +172,7 @@ def test_unknown_model_uses_only_explicit_generic_contract(self):
         self.assertEqual(resolved.profile.load_strategy, "image_text_to_text")
         self.assertEqual(
             resolved.profile.fusion_fields,
-            ("generic", "linear", "attention", "generic_cached_attention"),
+            ("generic", "linear", "attention", "generic_cached_attention", "generic_gqa_attention"),
         )
 
     def test_unknown_model_defaults_to_generic_text_no_cache(self):
@@ -463,6 +475,82 @@ def record(index, name, target, args, shape, kwargs=None):
         )
         self.assertEqual(cached[0].kwargs["scale"], 0.5)
 
+    def test_generic_prefill_gqa_uses_cache_derived_causal_mask(self):
+        def record(index, name, target, args, shape, kwargs=None):
+            return CModels.LayerRecord(
+                index=index, name=name,
+                node_type="placeholder" if not args else "call_function",
+                target=target, args=args, kwargs=kwargs or {}, users=[],
+                tensor_output_meta={"shape": shape, "dtype": "torch.float16"},
+                module_stack=None,
+            )
+
+        q = [1, 4, 4, 8]
+        kv = [1, 2, 4, 8]
+        layer_map = CModels.LayerMap(
+            model_name="unknown-prefill-gqa", task="prefill_with_cache",
+            graph_signature="", range_constants="", nodes=[
+                record(0, "query", "query", [], q),
+                record(1, "past_key_values_0", "past_key_values_0", [], [0]),
+                record(2, "past_key_values_1", "past_key_values_1", [], [0]),
+                record(3, "key_new", "key_new", [], kv),
+                record(4, "value_new", "value_new", [], kv),
+                record(5, "key_cat", "cactus.cat", [{"node": "past_key_values_0"}, {"node": "key_new"}], kv),
+                record(6, "value_cat", "cactus.cat", [{"node": "past_key_values_1"}, {"node": "value_new"}], kv),
+                record(7, "key_expand", "cactus.expand", [{"node": "key_cat"}], [1, 4, 4, 8]),
+                record(8, "qk", "aten.bmm.default", [{"node": "query"}, {"node": "key_expand"}], [4, 4, 4]),
+                record(9, "mask", "mask", [], [1, 1, 4, 4]),
+                record(10, "masked_qk", "cactus.add", [{"node": "qk"}, {"node": "mask"}], [1, 4, 4, 4]),
+                record(11, "softmax", "cactus.softmax", [{"node": "masked_qk"}], [1, 4, 4, 4], {"axis": -1}),
+                record(12, "value_expand", "cactus.expand", [{"node": "value_cat"}], [1, 4, 4, 8]),
+                record(13, "value_bmm", "aten.bmm.default", [{"node": "softmax"}, {"node": "value_expand"}], [4, 4, 8]),
+                record(14, "attention_out", "cactus.view", [{"node": "value_bmm"}], q),
+                CModels.LayerRecord(
+                    index=15, name="output", node_type="output", target="output",
+                    args=[[{"node": "attention_out"}, {"node": "key_cat"}, {"node": "value_cat"}]],
+                    kwargs={}, users=[], tensor_output_meta=None, module_stack=None,
+                ),
+            ],
+        )
+
+        simplified = simplify_ir.simplify(
+            layer_map,
+            fusion_fields=("generic", "attention", "cache", "generic_gqa_attention"),
+        )
+        cached = [node for node in simplified.nodes if node.target == "cactus.attention_cached"]
+
+        self.assertEqual(len(cached), 1)
+        self.assertEqual(
+            [item["node"] for item in cached[0].args[:5]],
+            ["query", "key_new", "value_new", "key_cat", "value_cat"],
+        )
+        self.assertEqual(len(cached[0].args), 5)
+        self.assertFalse(cached[0].kwargs["additive_mask"])
+        self.assertTrue(cached[0].kwargs["dropped_mask_builder"])
+        self.assertFalse(cached[0].kwargs["preserved_prefill_mask"])
+
+    def test_generic_dynamic_kv_enables_chunk_prefill(self):
+        resolved = cli_transpiler.resolve_transpile_config(
+            "example/generic-causal-lm",
+            input_modalities="text",
+            generic_task="causal-lm",
+            cache_style="dynamic-kv",
+        )
+
+        self.assertNotIn("scalar_prefill", resolved.profile.cache_policy)
+        self.assertIn("generic_gqa_attention", resolved.profile.fusion_fields)
+        self.assertIn(
+            "decoder_prefill_cache_chunk",
+            resolved.profile.cache_contract.fp16_kv_cache_components,
+        )
+
+    def test_generic_gqa_rejects_nonintegral_query_groups(self):
+        query = mock.Mock(tensor_output_meta={"shape": [1, 6, 4, 8]})
+        key = mock.Mock(tensor_output_meta={"shape": [1, 4, 4, 8]})
+        value = mock.Mock(tensor_output_meta={"shape": [1, 4, 4, 8]})
+
+        self.assertFalse(special_fusions.generic_cached_attention_shapes_match(query, key, value))
+
     def test_noop_cleanup_exposes_fusion_in_same_simplify_call(self):
         tensor_meta = {"shape": [1, 8], "dtype": "torch.float16"}
         layer_map = CModels.LayerMap(