ferrum_interfaces/

model_executor.rs

//! Model execution interface with clear prefill/decode separation
//!
//! This module provides the ModelExecutor trait that replaces the "fat" Model
//! interface, focusing purely on tensor operations without tokenization or sampling.

use crate::{KvCacheHandle, RecurrentStateHandle, RecurrentStateSpec, TensorRef};
use async_trait::async_trait;
use ferrum_types::{ExecutorAdmissionLimits, FerrumError, ModelInfo, RequestId, Result, TokenId};
use serde::{Deserialize, Serialize};
use std::{
    collections::{hash_map::DefaultHasher, HashMap, HashSet},
    future::Future,
    hash::{Hash, Hasher},
    num::NonZeroU64,
    ops::Range,
    pin::Pin,
    sync::Arc,
};

mod prefix_capture;
mod prefix_restore;
pub use prefix_capture::{
    PrefixCaptureBoundary, PrefixCaptureLease, PrefixCapturePlan, PrefixCaptureRequest,
    PrefixCaptureStatus,
};
pub use prefix_restore::{
    PlanRuntimePrefixRestoreDeferral, PlanRuntimePrefixRestoreInput,
    PlanRuntimePrefixRestoreOutcome, PlanRuntimePrefixRestoreOutput, PrefixRestoreDecision,
    PrefixRestoreObservation, PrefixRestoreSource,
};

/// One model-owned KV slot reservation request.
///
/// `cache_id` is the executor/model cache key attached to a sequence. `target_len`
/// is the sequence length that must be writable before the next forward runs.
/// `admission_target_len`, when present, is a larger known-context bound used
/// only for admission fit checks. Paged models must not allocate future blocks
/// for it; it mirrors vLLM's chunked-prefill full-context fit gate.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct KvSlotRequest {
    pub cache_id: String,
    pub target_len: usize,
    pub admission_target_len: Option<usize>,
}

/// Per-cache outcome from a KV slot reservation attempt.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct KvSlotAllocation {
    pub cache_id: String,
    pub blocks_before: usize,
    pub blocks_after: usize,
    pub new_blocks: usize,
}

/// Model-owned paged-KV reservation evidence.
///
/// Executors that own a vLLM-style physical KV block pool return this after
/// reserving all requested slots. Executors without model-owned paged KV return
/// `None` from `reserve_kv_slots`.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct KvSlotReservation {
    pub block_size: usize,
    pub total_blocks: usize,
    pub free_blocks_before: usize,
    pub free_blocks_after: usize,
    pub allocations: Vec<KvSlotAllocation>,
}

/// Point-in-time model-owned paged-KV capacity snapshot.
///
/// This is intentionally smaller than [`KvSlotReservation`]: it lets the
/// engine observe whether physical block capacity has actually changed after a
/// release, without allocating speculative slots or depending on model-family
/// names.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct KvSlotCapacitySnapshot {
    pub block_size: usize,
    pub total_blocks: usize,
    pub free_blocks: usize,
}

/// Token-validity mask for model-side greedy argmax.
///
/// `valid_token_mask[id] != 0` means token `id` may be selected. Tokens at or
/// above `valid_token_mask.len()` are invalid. The fingerprint lets model
/// backends cache an uploaded device mask without comparing the full vector on
/// every decode step.
#[derive(Clone)]
pub struct TokenSelectionMask {
    pub fingerprint: u64,
    pub valid_token_mask: Arc<[i8]>,
}

impl TokenSelectionMask {
    pub fn new(valid_token_mask: Vec<i8>) -> Self {
        let fingerprint = Self::fingerprint(&valid_token_mask);
        Self {
            fingerprint,
            valid_token_mask: Arc::from(valid_token_mask),
        }
    }

    fn fingerprint(valid_token_mask: &[i8]) -> u64 {
        let mut hasher = DefaultHasher::new();
        valid_token_mask.hash(&mut hasher);
        hasher.finish()
    }

    /// Change a small set of token-validity slots and refresh the cache key
    /// once. `Arc::make_mut` keeps the common unshared request mask in place
    /// and preserves safety when an in-flight backend policy still owns a
    /// clone.
    pub fn set_tokens_validity(&mut self, token_ids: &[u32], valid: bool) -> bool {
        let value = i8::from(valid);
        let slots = Arc::make_mut(&mut self.valid_token_mask);
        let mut changed = false;
        for &token_id in token_ids {
            if let Some(slot) = slots.get_mut(token_id as usize) {
                if *slot != value {
                    *slot = value;
                    changed = true;
                }
            }
        }
        if changed {
            self.fingerprint = Self::fingerprint(slots);
        }
        changed
    }

    pub fn len(&self) -> usize {
        self.valid_token_mask.len()
    }

    pub fn is_empty(&self) -> bool {
        self.valid_token_mask.is_empty()
    }
}

impl std::fmt::Debug for TokenSelectionMask {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        let valid_count = self.valid_token_mask.iter().filter(|&&v| v != 0).count();
        f.debug_struct("TokenSelectionMask")
            .field("fingerprint", &self.fingerprint)
            .field("len", &self.valid_token_mask.len())
            .field("valid_count", &valid_count)
            .finish()
    }
}

#[cfg(test)]
mod token_selection_mask_tests {
    use super::TokenSelectionMask;

    #[test]
    fn response_completion_mask_is_copy_on_write_and_restores_fingerprint() {
        let mut mask = TokenSelectionMask::new(vec![1, 1, 1]);
        let original = mask.clone();
        let original_fingerprint = mask.fingerprint;

        assert!(mask.set_tokens_validity(&[1], false));
        assert_eq!(mask.valid_token_mask.as_ref(), &[1, 0, 1]);
        assert_eq!(original.valid_token_mask.as_ref(), &[1, 1, 1]);
        assert_ne!(mask.fingerprint, original_fingerprint);

        let masked_fingerprint = mask.fingerprint;
        assert!(!mask.set_tokens_validity(&[1], false));
        assert_eq!(mask.fingerprint, masked_fingerprint);

        assert!(mask.set_tokens_validity(&[1], true));
        assert_eq!(mask.fingerprint, original_fingerprint);
    }
}

#[derive(Clone, Debug)]
pub enum LogitsReturnPolicy {
    FullLogits,
    GreedyArgmax {
        token_mask: Option<TokenSelectionMask>,
        repetition_penalty: Option<GreedyRepetitionPenalty>,
    },
}

impl Default for LogitsReturnPolicy {
    fn default() -> Self {
        Self::FullLogits
    }
}

impl LogitsReturnPolicy {
    pub fn requires_full_logits(&self) -> bool {
        matches!(self, Self::FullLogits)
    }
}

/// Typed product output returned by a plan-runtime execution wave.
///
/// A selected token is not a one-element logits vector. Keeping the variants
/// distinct prevents the engine from inferring product semantics from tensor
/// shape and lets full-logits requests retain every host-side sampler,
/// grammar, and structured-output processor.
#[derive(Debug, Clone, PartialEq)]
pub enum ExecutorSamplingOutput {
    FullLogits(Vec<f32>),
    GreedyToken(TokenId),
}

impl ExecutorSamplingOutput {
    pub fn full_logits(logits: Vec<f32>) -> Result<Self> {
        if logits.is_empty() {
            return Err(FerrumError::backend(
                "plan-runtime sampling output requires non-empty logits",
            ));
        }
        Ok(Self::FullLogits(logits))
    }

    pub const fn greedy_token(token: TokenId) -> Self {
        Self::GreedyToken(token)
    }

    /// Validate that a device-selected token was explicitly authorized.
    ///
    /// Returning full logits for a greedy policy remains legal: heterogeneous
    /// batches may deliberately fall back to host sampling. The inverse would
    /// skip required product processors and therefore fails closed.
    pub fn validate_for_policy(
        &self,
        policy: &LogitsReturnPolicy,
        vocabulary_size: usize,
    ) -> Result<()> {
        match self {
            Self::FullLogits(logits) if logits.len() != vocabulary_size => {
                return Err(FerrumError::backend(format!(
                    "plan runtime returned {} logits for vocabulary {vocabulary_size}",
                    logits.len()
                )));
            }
            Self::GreedyToken(_) if policy.requires_full_logits() => {
                return Err(FerrumError::backend(
                    "plan runtime returned a greedy token for a full-logits request",
                ));
            }
            Self::GreedyToken(token)
                if usize::try_from(token.get())
                    .ok()
                    .is_none_or(|token| token >= vocabulary_size) =>
            {
                return Err(FerrumError::backend(format!(
                    "plan runtime returned token {} outside vocabulary {vocabulary_size}",
                    token.get()
                )));
            }
            _ => {}
        }
        Ok(())
    }

    pub fn into_full_logits(self) -> Result<Vec<f32>> {
        match self {
            Self::FullLogits(logits) => Ok(logits),
            Self::GreedyToken(_) => Err(FerrumError::backend(
                "plan-runtime prefill unexpectedly returned a selected token",
            )),
        }
    }
}

#[cfg(test)]
mod executor_sampling_output_tests {
    use super::{ExecutorSamplingOutput, LogitsReturnPolicy};
    use ferrum_types::TokenId;

    #[test]
    fn full_logits_require_exact_vocabulary_width() {
        let output = ExecutorSamplingOutput::full_logits(vec![0.0; 4]).unwrap();
        assert!(output
            .validate_for_policy(&LogitsReturnPolicy::FullLogits, 4)
            .is_ok());
        assert!(output
            .validate_for_policy(&LogitsReturnPolicy::FullLogits, 5)
            .is_err());
    }

    #[test]
    fn greedy_token_requires_greedy_policy_and_in_vocabulary_token() {
        let allowed = LogitsReturnPolicy::GreedyArgmax {
            token_mask: None,
            repetition_penalty: None,
        };
        let output = ExecutorSamplingOutput::greedy_token(TokenId::new(3));
        assert!(output.validate_for_policy(&allowed, 4).is_ok());
        assert!(output.validate_for_policy(&allowed, 3).is_err());
        assert!(output
            .validate_for_policy(&LogitsReturnPolicy::FullLogits, 4)
            .is_err());
    }

    #[test]
    fn full_logits_are_a_legal_greedy_batch_fallback() {
        let policy = LogitsReturnPolicy::GreedyArgmax {
            token_mask: None,
            repetition_penalty: None,
        };
        let output = ExecutorSamplingOutput::full_logits(vec![0.0; 4]).unwrap();
        assert!(output.validate_for_policy(&policy, 4).is_ok());
    }
}

/// Sparse repetition-penalty metadata for model-side greedy argmax.
///
/// The token list is request-local and de-duplicated. Applying the penalty
/// before GPU argmax avoids downloading full `[batch, vocab]` logits for the
/// common greedy chat path while preserving repeat avoidance.
#[derive(Clone, Debug)]
pub struct GreedyRepetitionPenalty {
    penalty: f32,
    token_ids: Arc<[u32]>,
}

impl GreedyRepetitionPenalty {
    pub fn new(penalty: f32, mut token_ids: Vec<u32>) -> Self {
        let mut seen = HashSet::with_capacity(token_ids.len());
        token_ids.retain(|token| seen.insert(*token));
        Self {
            penalty,
            token_ids: Arc::from(token_ids),
        }
    }

    pub const fn penalty(&self) -> f32 {
        self.penalty
    }

    pub fn token_ids(&self) -> &[u32] {
        &self.token_ids
    }

    pub fn is_empty(&self) -> bool {
        self.token_ids.is_empty() || self.penalty == 1.0
    }
}

#[cfg(test)]
mod greedy_repetition_penalty_tests {
    use super::GreedyRepetitionPenalty;

    #[test]
    fn constructor_preserves_first_seen_order_and_removes_duplicates() {
        let repetition = GreedyRepetitionPenalty::new(1.1, vec![7, 3, 7, 9, 3]);
        assert_eq!(repetition.penalty(), 1.1);
        assert_eq!(repetition.token_ids(), [7, 3, 9]);
    }
}

/// Input for prefill phase (processing the initial prompt)
#[derive(Debug, Clone)]
pub struct PrefillInput {
    /// Stable product request identity for plan-runtime resources.
    pub request_id: Option<RequestId>,
    /// Maximum sequence extent this request may reach, including the prompt.
    /// Executors use this for fit validation without allocating future pages.
    pub maximum_sequence_tokens: Option<usize>,
    /// Exact scheduler-owned prompt chunk for this invocation.
    ///
    /// The input tensor still contains the full prompt so token identity and
    /// global offsets remain stable. Plan runtimes execute only this range.
    pub chunk: Option<PrefillChunk>,
    /// Input token IDs [batch_size, sequence_length]
    pub input_ids: TensorRef,
    /// Attention mask [batch_size, sequence_length] (optional)
    pub attention_mask: Option<TensorRef>,
    /// Position IDs [batch_size, sequence_length] (optional, for RoPE)
    pub position_ids: Option<TensorRef>,
    /// Pre-allocated KV cache handle (optional, for paged attention)
    pub kv_cache: Option<Arc<dyn KvCacheHandle>>,
    /// Pre-allocated recurrent-state handle (optional, for state-space layers)
    pub recurrent_state: Option<Arc<dyn RecurrentStateHandle>>,
    /// Request metadata that can affect model execution.
    pub metadata: HashMap<String, serde_json::Value>,
}

impl PrefillInput {
    /// Create new prefill input
    pub fn new(input_ids: TensorRef) -> Self {
        Self {
            request_id: None,
            maximum_sequence_tokens: None,
            chunk: None,
            input_ids,
            attention_mask: None,
            position_ids: None,
            kv_cache: None,
            recurrent_state: None,
            metadata: HashMap::new(),
        }
    }

    /// Attach the typed request boundary consumed by plan runtimes.
    pub fn with_request_context(
        mut self,
        request_id: RequestId,
        maximum_sequence_tokens: usize,
    ) -> Self {
        self.request_id = Some(request_id);
        self.maximum_sequence_tokens = Some(maximum_sequence_tokens);
        self
    }

    /// Attach the exact scheduler-published prompt chunk.
    pub fn with_chunk(mut self, chunk: PrefillChunk) -> Self {
        self.chunk = Some(chunk);
        self
    }

    /// Create prefill input with a pre-allocated KV cache handle.
    pub fn with_kv_cache(mut self, kv_cache: Arc<dyn KvCacheHandle>) -> Self {
        self.kv_cache = Some(kv_cache);
        self
    }

    /// Create prefill input with a pre-allocated recurrent-state handle.
    pub fn with_recurrent_state(mut self, recurrent_state: Arc<dyn RecurrentStateHandle>) -> Self {
        self.recurrent_state = Some(recurrent_state);
        self
    }

    /// Attach request metadata.
    pub fn with_metadata(mut self, metadata: HashMap<String, serde_json::Value>) -> Self {
        self.metadata = metadata;
        self
    }

    /// Add attention mask
    pub fn with_attention_mask(mut self, mask: TensorRef) -> Self {
        self.attention_mask = Some(mask);
        self
    }

    /// Add position IDs
    pub fn with_position_ids(mut self, positions: TensorRef) -> Self {
        self.position_ids = Some(positions);
        self
    }

    /// Get batch size
    pub fn batch_size(&self) -> usize {
        self.input_ids.shape()[0]
    }

    /// Get sequence length
    pub fn sequence_length(&self) -> usize {
        if self.input_ids.shape().len() >= 2 {
            self.input_ids.shape()[1]
        } else {
            1
        }
    }
}

/// Exact, validated prompt progress assigned to one prefill invocation.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
pub struct PrefillChunk {
    tokens_processed: usize,
    tokens_to_process: usize,
    total_prompt_tokens: usize,
}

#[cfg(test)]
mod prefill_chunk_tests {
    use super::PrefillChunk;

    #[test]
    fn validates_exact_progress_and_finality() {
        let first = PrefillChunk::new(0, 3, 8).unwrap();
        assert_eq!(first.range(), 0..3);
        assert_eq!(first.end(), 3);
        assert!(!first.is_final());

        let final_chunk = PrefillChunk::new(3, 5, 8).unwrap();
        assert_eq!(final_chunk.range(), 3..8);
        assert!(final_chunk.is_final());
    }

    #[test]
    fn rejects_empty_out_of_bounds_and_overflowing_progress() {
        assert!(PrefillChunk::new(0, 0, 8).is_err());
        assert!(PrefillChunk::new(0, 1, 0).is_err());
        assert!(PrefillChunk::new(7, 2, 8).is_err());
        assert!(PrefillChunk::new(usize::MAX, 1, usize::MAX).is_err());
    }
}

impl PrefillChunk {
    pub fn new(
        tokens_processed: usize,
        tokens_to_process: usize,
        total_prompt_tokens: usize,
    ) -> Result<Self> {
        let end = tokens_processed
            .checked_add(tokens_to_process)
            .ok_or_else(|| {
                ferrum_types::FerrumError::request_validation("prefill chunk overflows")
            })?;
        if tokens_to_process == 0 || total_prompt_tokens == 0 || end > total_prompt_tokens {
            return Err(ferrum_types::FerrumError::request_validation(
                "prefill chunk must be non-empty and within the full prompt",
            ));
        }
        Ok(Self {
            tokens_processed,
            tokens_to_process,
            total_prompt_tokens,
        })
    }

    pub const fn tokens_processed(self) -> usize {
        self.tokens_processed
    }

    pub const fn tokens_to_process(self) -> usize {
        self.tokens_to_process
    }

    pub const fn total_prompt_tokens(self) -> usize {
        self.total_prompt_tokens
    }

    pub fn range(self) -> Range<usize> {
        self.tokens_processed..self.tokens_processed + self.tokens_to_process
    }

    pub const fn end(self) -> usize {
        self.tokens_processed + self.tokens_to_process
    }

    pub const fn is_final(self) -> bool {
        self.end() == self.total_prompt_tokens
    }
}

/// Output from prefill phase
#[derive(Debug, Clone)]
pub struct PrefillOutput {
    /// Logits for all positions [batch_size, sequence_length, vocab_size]
    pub logits: TensorRef,
    /// KV cache handle populated with prompt states
    pub kv_cache: Arc<dyn KvCacheHandle>,
    /// Recurrent-state handle populated with prompt state, when used.
    pub recurrent_state: Option<Arc<dyn RecurrentStateHandle>>,
    /// Hidden states at each layer (optional, for analysis)
    pub hidden_states: Option<Vec<TensorRef>>,
    /// Attention weights (optional, for analysis)
    pub attention_weights: Option<Vec<TensorRef>>,
}

impl PrefillOutput {
    /// Create new prefill output
    pub fn new(logits: TensorRef, kv_cache: Arc<dyn KvCacheHandle>) -> Self {
        Self {
            logits,
            kv_cache,
            recurrent_state: None,
            hidden_states: None,
            attention_weights: None,
        }
    }

    /// Attach updated recurrent state to the prefill output.
    pub fn with_recurrent_state(mut self, recurrent_state: Arc<dyn RecurrentStateHandle>) -> Self {
        self.recurrent_state = Some(recurrent_state);
        self
    }

    /// Get logits for last position (for next token generation)
    pub fn last_token_logits(&self) -> Result<TensorRef> {
        let shape = self.logits.shape();
        if shape.len() != 3 {
            return Err(ferrum_types::FerrumError::backend(
                "Expected 3D logits tensor [batch, seq, vocab]",
            ));
        }

        let seq_len = shape[1];
        if seq_len == 0 {
            return Err(ferrum_types::FerrumError::backend("Empty sequence"));
        }

        // Extract last position: [batch, seq-1:seq, vocab] -> [batch, vocab]
        self.logits
            .view(&[0, seq_len - 1, 0], &[shape[0], seq_len, shape[2]])
    }
}

/// Input for decode phase (generating one token at a time)
#[derive(Debug, Clone)]
pub struct DecodeInput {
    /// Stable product request identity for plan-runtime resources.
    pub request_id: Option<RequestId>,
    /// Input token ID for current step [batch_size, 1]
    pub input_ids: TensorRef,
    /// Existing KV cache from previous steps
    pub kv_cache: Arc<dyn KvCacheHandle>,
    /// Existing recurrent state from previous steps, when used.
    pub recurrent_state: Option<Arc<dyn RecurrentStateHandle>>,
    /// Position IDs for current step [batch_size, 1] (optional)
    pub position_ids: Option<TensorRef>,
    /// Request metadata that can affect model execution.
    pub metadata: HashMap<String, serde_json::Value>,
    /// How the model may return final-position logits for this request.
    pub logits_policy: LogitsReturnPolicy,
}

impl DecodeInput {
    /// Create new decode input
    pub fn new(input_ids: TensorRef, kv_cache: Arc<dyn KvCacheHandle>) -> Self {
        Self {
            request_id: None,
            input_ids,
            kv_cache,
            recurrent_state: None,
            position_ids: None,
            metadata: HashMap::new(),
            logits_policy: LogitsReturnPolicy::FullLogits,
        }
    }

    /// Attach the product request identity to this decode step.
    pub fn with_request_id(mut self, request_id: RequestId) -> Self {
        self.request_id = Some(request_id);
        self
    }

    /// Add position IDs
    pub fn with_position_ids(mut self, positions: TensorRef) -> Self {
        self.position_ids = Some(positions);
        self
    }

    /// Attach recurrent state for state-space or hybrid layers.
    pub fn with_recurrent_state(mut self, recurrent_state: Arc<dyn RecurrentStateHandle>) -> Self {
        self.recurrent_state = Some(recurrent_state);
        self
    }

    /// Attach request metadata.
    pub fn with_metadata(mut self, metadata: HashMap<String, serde_json::Value>) -> Self {
        self.metadata = metadata;
        self
    }

    pub fn with_logits_policy(mut self, policy: LogitsReturnPolicy) -> Self {
        self.logits_policy = policy;
        self
    }

    /// Get batch size
    pub fn batch_size(&self) -> usize {
        self.input_ids.shape()[0]
    }
}

/// One sequence's contribution to a unified mixed-batch forward.
///
/// A unified batch lets a single model forward pass process a mix of
/// per-sequence work units: a prefill chunk (q_tokens.len() ≥ 1, possibly
/// continuing from `pos_offset > 0` for chunked prefill) and a decode step
/// (q_tokens.len() == 1, `pos_offset` = current cache length) coexist in
/// the same call. The model layer concatenates all `q_tokens` into one
/// [M_total, hidden] tensor and runs all GEMMs / norms once; only the
/// attention kernel sees per-item segmentation.
///
/// This is the abstraction that enables vLLM-style chunked prefill where
/// decode tokens for already-running sequences are produced in the same
/// iter as a prefill chunk for a newly-arriving sequence.
#[derive(Clone)]
pub struct UnifiedBatchItem {
    /// Identifier matching the sequence's KV cache (model-side keying).
    pub seq_id: String,
    /// Tokens to process this iter. For decode this is exactly 1 token;
    /// for prefill (chunked or whole) this is the chunk's tokens.
    pub q_tokens: Vec<u32>,
    /// KV cache handle for this sequence.
    pub kv_cache: Arc<dyn KvCacheHandle>,
    /// Recurrent-state handle for this sequence, when used.
    pub recurrent_state: Option<Arc<dyn RecurrentStateHandle>>,
    /// Starting absolute position for the FIRST token in `q_tokens`.
    /// 0 for a fresh prefill, `kv_len` for a decode step or a continuing
    /// chunked-prefill slice.
    pub pos_offset: usize,
    /// True iff this item completes the request's prefill (or is a decode
    /// item) — i.e. logits at the last token of `q_tokens` should be
    /// returned for sampling. Intermediate prefill chunks set this false
    /// to skip the lm_head + sampling path.
    pub is_final_chunk: bool,
    /// Request metadata that can affect model execution.
    pub metadata: HashMap<String, serde_json::Value>,
    /// How the model may return final-position logits for this item.
    pub logits_policy: LogitsReturnPolicy,
}

impl std::fmt::Debug for UnifiedBatchItem {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("UnifiedBatchItem")
            .field("seq_id", &self.seq_id)
            .field("q_len", &self.q_tokens.len())
            .field("has_recurrent_state", &self.recurrent_state.is_some())
            .field("pos_offset", &self.pos_offset)
            .field("is_final_chunk", &self.is_final_chunk)
            .finish()
    }
}

/// A mixed-batch forward request: any combination of in-progress prefill
/// chunks and decode steps. See [`UnifiedBatchItem`] for the per-item
/// semantics. The producer (engine) groups all sequences active in this
/// iter into a single batch; the consumer (model) runs one forward and
/// returns per-item logits (only for items with `is_final_chunk = true`,
/// in the order they appear in `items`).
#[derive(Debug, Clone, Default)]
pub struct UnifiedBatch {
    pub items: Vec<UnifiedBatchItem>,
}

impl UnifiedBatch {
    pub fn new() -> Self {
        Self::default()
    }

    /// Total query tokens across all items — corresponds to the M dim of
    /// the model's per-layer GEMMs in the unified forward.
    pub fn total_q_tokens(&self) -> usize {
        self.items.iter().map(|it| it.q_tokens.len()).sum()
    }

    /// Number of items that will produce a logits vector (decode items
    /// always; prefill items only on their final chunk).
    pub fn num_sampled_items(&self) -> usize {
        self.items.iter().filter(|it| it.is_final_chunk).count()
    }
}

/// Tensor-free decode input for an executor with plan-runtime resource
/// authority.
///
/// The runtime already owns the request's physical cache and accepts exactly
/// one token per decode frontier, so materializing a host tensor here adds no
/// information. Legacy and modality executors continue to use [`DecodeInput`].
#[derive(Debug, Clone)]
pub struct PlanRuntimeDecodeInput {
    pub request_id: RequestId,
    pub input_token: TokenId,
    pub kv_cache: Arc<dyn KvCacheHandle>,
    pub logits_policy: LogitsReturnPolicy,
}

impl PlanRuntimeDecodeInput {
    pub fn new(
        request_id: RequestId,
        input_token: TokenId,
        kv_cache: Arc<dyn KvCacheHandle>,
    ) -> Self {
        Self {
            request_id,
            input_token,
            kv_cache,
            logits_policy: LogitsReturnPolicy::FullLogits,
        }
    }

    pub fn with_logits_policy(mut self, logits_policy: LogitsReturnPolicy) -> Self {
        self.logits_policy = logits_policy;
        self
    }
}

/// Tensor-free prefill input for an executor with plan-runtime resource
/// authority.
///
/// The complete prompt remains immutable across chunk retries so scheduler
/// offsets and executor sequence identity cannot drift. Physical KV and
/// recurrent resources remain opaque executor-owned state.
#[derive(Debug, Clone)]
pub struct PlanRuntimePrefillInput {
    pub request_id: RequestId,
    pub input_tokens: Arc<[TokenId]>,
    pub maximum_sequence_tokens: usize,
    pub chunk: PrefillChunk,
}

impl PlanRuntimePrefillInput {
    pub fn new(
        request_id: RequestId,
        input_tokens: impl Into<Arc<[TokenId]>>,
        maximum_sequence_tokens: usize,
        chunk: PrefillChunk,
    ) -> Result<Self> {
        let input_tokens = input_tokens.into();
        if input_tokens.is_empty() {
            return Err(FerrumError::request_validation(
                "plan-runtime prefill requires at least one input token",
            ));
        }
        if chunk.total_prompt_tokens() != input_tokens.len() {
            return Err(FerrumError::request_validation(format!(
                "plan-runtime prefill chunk declares {} prompt tokens for input length {}",
                chunk.total_prompt_tokens(),
                input_tokens.len()
            )));
        }
        if maximum_sequence_tokens < input_tokens.len() {
            return Err(FerrumError::request_validation(format!(
                "plan-runtime sequence ceiling {maximum_sequence_tokens} does not cover prompt length {}",
                input_tokens.len()
            )));
        }
        Ok(Self {
            request_id,
            input_tokens,
            maximum_sequence_tokens,
            chunk,
        })
    }
}

/// Output from decode phase
#[derive(Debug, Clone)]
pub struct DecodeOutput {
    /// Logits for next token [batch_size, vocab_size]
    pub logits: TensorRef,
    /// Updated KV cache with new token state
    pub kv_cache: Arc<dyn KvCacheHandle>,
    /// Updated recurrent state, when used.
    pub recurrent_state: Option<Arc<dyn RecurrentStateHandle>>,
    /// Hidden state for current token (optional)
    pub hidden_state: Option<TensorRef>,
    /// Attention weights for current token (optional)
    pub attention_weights: Option<Vec<TensorRef>>,
}

impl DecodeOutput {
    /// Create new decode output
    pub fn new(logits: TensorRef, kv_cache: Arc<dyn KvCacheHandle>) -> Self {
        Self {
            logits,
            kv_cache,
            recurrent_state: None,
            hidden_state: None,
            attention_weights: None,
        }
    }

    /// Attach updated recurrent state to the decode output.
    pub fn with_recurrent_state(mut self, recurrent_state: Arc<dyn RecurrentStateHandle>) -> Self {
        self.recurrent_state = Some(recurrent_state);
        self
    }
}

/// Tensor-free decode output from a plan-runtime executor.
#[derive(Debug, Clone)]
pub struct PlanRuntimeDecodeOutput {
    pub sampling_output: ExecutorSamplingOutput,
    pub kv_cache: Arc<dyn KvCacheHandle>,
}

impl PlanRuntimeDecodeOutput {
    pub fn new(sampling_output: ExecutorSamplingOutput, kv_cache: Arc<dyn KvCacheHandle>) -> Self {
        Self {
            sampling_output,
            kv_cache,
        }
    }
}

/// Product state emitted by one completed plan-runtime prefill chunk.
#[derive(Debug)]
pub enum PlanRuntimePrefillProduct {
    /// An intermediate chunk updates executor-owned state but is not sampleable.
    Intermediate,
    /// A final chunk returns the complete vocabulary row to the engine.
    FinalLogits(Vec<f32>),
}

/// Exact executor-owned authority advanced by one prefill completion.
#[derive(Debug)]
pub struct PlanRuntimePrefillAuthority {
    request_id: RequestId,
    committed_tokens: usize,
    kv_cache: Arc<dyn KvCacheHandle>,
}

impl PlanRuntimePrefillAuthority {
    pub fn request_id(&self) -> &RequestId {
        &self.request_id
    }

    pub const fn committed_tokens(&self) -> usize {
        self.committed_tokens
    }

    pub fn kv_cache(&self) -> &Arc<dyn KvCacheHandle> {
        &self.kv_cache
    }

    pub fn into_cache(self) -> Arc<dyn KvCacheHandle> {
        self.kv_cache
    }
}

/// Tensor-free prefill output bound to one request and committed KV extent.
#[derive(Debug)]
pub struct PlanRuntimePrefillOutput {
    authority: PlanRuntimePrefillAuthority,
    product: PlanRuntimePrefillProduct,
}

impl PlanRuntimePrefillOutput {
    pub fn intermediate(
        request_id: RequestId,
        committed_tokens: usize,
        kv_cache: Arc<dyn KvCacheHandle>,
    ) -> Self {
        Self {
            authority: PlanRuntimePrefillAuthority {
                request_id,
                committed_tokens,
                kv_cache,
            },
            product: PlanRuntimePrefillProduct::Intermediate,
        }
    }

    pub fn final_logits(
        request_id: RequestId,
        committed_tokens: usize,
        logits: Vec<f32>,
        kv_cache: Arc<dyn KvCacheHandle>,
    ) -> Result<Self> {
        if logits.is_empty() {
            return Err(FerrumError::backend(
                "plan-runtime final prefill returned empty logits",
            ));
        }
        Ok(Self {
            authority: PlanRuntimePrefillAuthority {
                request_id,
                committed_tokens,
                kv_cache,
            },
            product: PlanRuntimePrefillProduct::FinalLogits(logits),
        })
    }

    pub fn request_id(&self) -> &RequestId {
        self.authority.request_id()
    }

    pub const fn committed_tokens(&self) -> usize {
        self.authority.committed_tokens()
    }

    pub fn product(&self) -> &PlanRuntimePrefillProduct {
        &self.product
    }

    pub fn kv_cache(&self) -> &Arc<dyn KvCacheHandle> {
        self.authority.kv_cache()
    }

    pub fn validate_for_completion(
        &self,
        expected_request_id: &RequestId,
        completed_chunk: PrefillChunk,
        vocabulary_size: usize,
    ) -> Result<()> {
        if self.request_id() != expected_request_id {
            return Err(FerrumError::backend(format!(
                "plan runtime returned prefill output for request {}, expected {expected_request_id}",
                self.request_id()
            )));
        }
        if self.committed_tokens() != completed_chunk.end() {
            return Err(FerrumError::backend(format!(
                "plan runtime returned prefill extent {}, expected {}",
                self.committed_tokens(),
                completed_chunk.end()
            )));
        }
        if self.kv_cache().num_tokens() != self.committed_tokens() {
            return Err(FerrumError::backend(format!(
                "plan runtime prefill cache `{}` reports {} tokens for committed extent {}",
                self.kv_cache().cache_id(),
                self.kv_cache().num_tokens(),
                self.committed_tokens()
            )));
        }
        match (&self.product, completed_chunk.is_final()) {
            (PlanRuntimePrefillProduct::Intermediate, false) => Ok(()),
            (PlanRuntimePrefillProduct::FinalLogits(logits), true)
                if logits.len() == vocabulary_size =>
            {
                Ok(())
            }
            (PlanRuntimePrefillProduct::FinalLogits(logits), true) => {
                Err(FerrumError::backend(format!(
                    "plan runtime returned {} final prefill logits for vocabulary {vocabulary_size}",
                    logits.len()
                )))
            }
            (PlanRuntimePrefillProduct::Intermediate, true) => Err(FerrumError::backend(
                "plan runtime returned an intermediate product for a final prefill chunk",
            )),
            (PlanRuntimePrefillProduct::FinalLogits(_), false) => Err(FerrumError::backend(
                "plan runtime returned final logits for an intermediate prefill chunk",
            )),
        }
    }

    pub fn into_parts(self) -> (PlanRuntimePrefillAuthority, PlanRuntimePrefillProduct) {
        (self.authority, self.product)
    }
}

/// Product-authoritative evidence for a successfully completed sequence.
///
/// Physical cache release is not completion evidence: cancellation, failure,
/// recompute, and successful generation all release the same cache authority.
/// The engine constructs this receipt only after it has finalized user-visible
/// token usage, allowing plan runtimes to reconcile terminal events without
/// inferring output counts from execution frames.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct ExecutorSequenceCompletion {
    request_id: RequestId,
    cache_id: String,
    input_tokens: u64,
    output_tokens: u64,
}

impl ExecutorSequenceCompletion {
    pub fn new(
        request_id: RequestId,
        cache_id: String,
        input_tokens: usize,
        output_tokens: usize,
    ) -> Result<Self> {
        if cache_id.is_empty() {
            return Err(FerrumError::request_validation(
                "executor sequence completion requires a cache identity",
            ));
        }
        let input_tokens = u64::try_from(input_tokens).map_err(|_| {
            FerrumError::request_validation("executor completion input token count exceeds u64")
        })?;
        let output_tokens = u64::try_from(output_tokens).map_err(|_| {
            FerrumError::request_validation("executor completion output token count exceeds u64")
        })?;
        Ok(Self {
            request_id,
            cache_id,
            input_tokens,
            output_tokens,
        })
    }

    pub fn request_id(&self) -> &RequestId {
        &self.request_id
    }

    pub fn cache_id(&self) -> &str {
        &self.cache_id
    }

    pub const fn input_tokens(&self) -> u64 {
        self.input_tokens
    }

    pub const fn output_tokens(&self) -> u64 {
        self.output_tokens
    }
}

pub use ferrum_types::ExecutionResourceAuthority;

/// Request-scoped authority selected for a capacity-pressure preemption.
///
/// The cache identity prevents the engine from releasing a newer sequence
/// incarnation after a stale scheduler decision. Implementations must retire
/// retained prefill and active decode authority through the same operation.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct ExecutorExecutionCapacityPreemption {
    request_id: RequestId,
    cache_id: String,
}

impl ExecutorExecutionCapacityPreemption {
    pub fn new(request_id: RequestId, cache_id: String) -> Result<Self> {
        if cache_id.is_empty() {
            return Err(FerrumError::request_validation(
                "execution-capacity preemption requires a cache identity",
            ));
        }
        Ok(Self {
            request_id,
            cache_id,
        })
    }

    pub fn request_id(&self) -> &RequestId {
        &self.request_id
    }

    pub fn cache_id(&self) -> &str {
        &self.cache_id
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum ExecutorExecutionCapacityPreemptionAuthority {
    RetainedPrefill,
    ActiveSequence,
}

/// Proof that the executor retired one exact request authority to a terminal
/// state. Source-generation advancement remains independently verified by the
/// engine before the scheduler may resume another frontier.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct ExecutorExecutionCapacityPreemptionReceipt {
    request_id: RequestId,
    cache_id: String,
    authority: ExecutorExecutionCapacityPreemptionAuthority,
}

impl ExecutorExecutionCapacityPreemptionReceipt {
    pub fn new(
        request_id: RequestId,
        cache_id: String,
        authority: ExecutorExecutionCapacityPreemptionAuthority,
    ) -> Self {
        Self {
            request_id,
            cache_id,
            authority,
        }
    }

    pub fn request_id(&self) -> &RequestId {
        &self.request_id
    }

    pub fn cache_id(&self) -> &str {
        &self.cache_id
    }

    pub const fn authority(&self) -> ExecutorExecutionCapacityPreemptionAuthority {
        self.authority
    }
}

/// Point-in-time memory evidence emitted by the shared plan runtime.
///
/// Static model allocations are separated from dynamic request resources so
/// product telemetry never reports model weights as KV or recurrent-state
/// usage. Process-wide claims are included because another live plan can
/// consume capacity visible to this runtime. Dynamic free bytes remain
/// reusable by this plan; quarantined and other claimed bytes do not.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct PlanRuntimeResourceSnapshot {
    device_capacity_bytes: u64,
    usable_capacity_bytes: u64,
    process_claimed_bytes: u64,
    plan_claimed_bytes: u64,
    static_bytes: u64,
    dynamic_resident_bytes: u64,
    dynamic_free_bytes: u64,
    pending_growth_bytes: u64,
    quarantined_bytes: u64,
}

impl PlanRuntimeResourceSnapshot {
    #[allow(clippy::too_many_arguments)]
    pub fn new(
        device_capacity_bytes: u64,
        usable_capacity_bytes: u64,
        process_claimed_bytes: u64,
        plan_claimed_bytes: u64,
        static_bytes: u64,
        dynamic_resident_bytes: u64,
        dynamic_free_bytes: u64,
        pending_growth_bytes: u64,
        quarantined_bytes: u64,
    ) -> Result<Self> {
        let snapshot = Self {
            device_capacity_bytes,
            usable_capacity_bytes,
            process_claimed_bytes,
            plan_claimed_bytes,
            static_bytes,
            dynamic_resident_bytes,
            dynamic_free_bytes,
            pending_growth_bytes,
            quarantined_bytes,
        };
        snapshot.validate()?;
        Ok(snapshot)
    }

    /// Revalidates deserialized evidence before it crosses a trusted failure or
    /// profile boundary.
    pub fn validate(&self) -> Result<()> {
        if self.usable_capacity_bytes > self.device_capacity_bytes {
            return Err(ferrum_types::FerrumError::internal(format!(
                "plan runtime usable capacity {} exceeds device capacity {}",
                self.usable_capacity_bytes, self.device_capacity_bytes
            )));
        }
        if self.process_claimed_bytes > self.usable_capacity_bytes {
            return Err(ferrum_types::FerrumError::internal(format!(
                "plan runtime process claims {} exceed usable capacity {}",
                self.process_claimed_bytes, self.usable_capacity_bytes
            )));
        }
        if self.plan_claimed_bytes > self.process_claimed_bytes {
            return Err(ferrum_types::FerrumError::internal(format!(
                "plan runtime plan claims {} exceed process claims {}",
                self.plan_claimed_bytes, self.process_claimed_bytes
            )));
        }
        if self.dynamic_free_bytes > self.dynamic_resident_bytes {
            return Err(ferrum_types::FerrumError::internal(format!(
                "plan runtime dynamic free bytes {} exceed resident bytes {}",
                self.dynamic_free_bytes, self.dynamic_resident_bytes
            )));
        }
        let minimum_plan_claim = self
            .static_bytes
            .checked_add(self.dynamic_resident_bytes)
            .and_then(|bytes| bytes.checked_add(self.quarantined_bytes))
            .ok_or_else(|| {
                ferrum_types::FerrumError::internal(
                    "plan runtime static, resident, and quarantined bytes overflow u64",
                )
            })?;
        if minimum_plan_claim > self.plan_claimed_bytes {
            return Err(ferrum_types::FerrumError::internal(format!(
                "plan runtime accounted plan bytes {minimum_plan_claim} exceed plan claims {}",
                self.plan_claimed_bytes
            )));
        }
        Ok(())
    }

    pub const fn device_capacity_bytes(&self) -> u64 {
        self.device_capacity_bytes
    }

    pub const fn usable_capacity_bytes(&self) -> u64 {
        self.usable_capacity_bytes
    }

    pub const fn process_claimed_bytes(&self) -> u64 {
        self.process_claimed_bytes
    }

    pub const fn plan_claimed_bytes(&self) -> u64 {
        self.plan_claimed_bytes
    }

    pub const fn static_bytes(&self) -> u64 {
        self.static_bytes
    }

    pub const fn dynamic_resident_bytes(&self) -> u64 {
        self.dynamic_resident_bytes
    }

    pub const fn dynamic_free_bytes(&self) -> u64 {
        self.dynamic_free_bytes
    }

    pub const fn dynamic_used_bytes(&self) -> u64 {
        self.dynamic_resident_bytes - self.dynamic_free_bytes
    }

    pub const fn pending_growth_bytes(&self) -> u64 {
        self.pending_growth_bytes
    }

    pub const fn quarantined_bytes(&self) -> u64 {
        self.quarantined_bytes
    }

    /// Capacity immediately reusable by this plan without reclaiming another
    /// plan: process-wide unclaimed bytes plus free extents already resident
    /// in this plan's dynamic pools.
    pub fn available_bytes(&self) -> Result<u64> {
        self.usable_capacity_bytes
            .checked_sub(self.process_claimed_bytes)
            .and_then(|bytes| bytes.checked_add(self.dynamic_free_bytes))
            .ok_or_else(|| {
                ferrum_types::FerrumError::internal(
                    "plan runtime available capacity calculation overflowed",
                )
            })
    }

    pub fn used_bytes(&self) -> Result<u64> {
        self.available_bytes().and_then(|available| {
            self.usable_capacity_bytes
                .checked_sub(available)
                .ok_or_else(|| {
                    ferrum_types::FerrumError::internal(
                        "plan runtime available bytes exceed usable capacity",
                    )
                })
        })
    }
}

#[cfg(test)]
mod plan_runtime_resource_snapshot_tests {
    use super::PlanRuntimeResourceSnapshot;

    #[test]
    fn separates_static_and_dynamic_usage() {
        let snapshot =
            PlanRuntimeResourceSnapshot::new(1_000, 900, 710, 710, 400, 300, 200, 20, 10).unwrap();

        assert_eq!(snapshot.available_bytes().unwrap(), 390);
        assert_eq!(snapshot.used_bytes().unwrap(), 510);
        assert_eq!(snapshot.dynamic_resident_bytes(), 300);
        assert_eq!(snapshot.dynamic_used_bytes(), 100);
        assert_eq!(snapshot.dynamic_free_bytes(), 200);
        assert_eq!(snapshot.pending_growth_bytes(), 20);
        assert_eq!(snapshot.quarantined_bytes(), 10);
    }

    #[test]
    fn rejects_incoherent_capacity_evidence() {
        assert!(PlanRuntimeResourceSnapshot::new(1_000, 1_001, 0, 0, 0, 0, 0, 0, 0).is_err());
        assert!(PlanRuntimeResourceSnapshot::new(1_000, 900, 901, 0, 0, 0, 0, 0, 0).is_err());
        assert!(PlanRuntimeResourceSnapshot::new(1_000, 900, 500, 501, 0, 0, 0, 0, 0).is_err());
        assert!(PlanRuntimeResourceSnapshot::new(1_000, 900, 100, 100, 0, 100, 101, 0, 0).is_err());
        assert!(PlanRuntimeResourceSnapshot::new(1_000, 900, 500, 500, 400, 100, 0, 0, 1).is_err());
    }
}

/// Typed origin for an executor-owned request lifecycle.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ExecutorRequestOrigin {
    Product,
    Startup,
    Diagnostic,
}

impl ExecutorRequestOrigin {
    pub const fn namespace(self) -> &'static str {
        match self {
            Self::Product => "product",
            Self::Startup => "startup",
            Self::Diagnostic => "diagnostic",
        }
    }

    pub fn from_namespaced_request_identity(identity: &str) -> Option<Self> {
        let suffix = identity.strip_prefix("request.")?;
        let (namespace, request_id) = suffix.split_once('.')?;
        if request_id.is_empty() {
            return None;
        }
        match namespace {
            "product" => Some(Self::Product),
            "startup" => Some(Self::Startup),
            "diagnostic" => Some(Self::Diagnostic),
            _ => None,
        }
    }
}

/// Borrowed, already-tokenized input used to probe plan-runtime prefill
/// admission before the request can enter a device submission batch.
///
/// This carries semantic token identity rather than an aggregate token count:
/// vNext derives the exact resource work shape and its fingerprint from this
/// boundary. The request remains owned by the scheduler while the executor
/// retains any admitted authority internally until [`ModelExecutor::prefill`]
/// consumes it or cancellation releases it.
#[derive(Debug, Clone, Copy)]
pub struct ExecutorPrefillAdmission<'a> {
    pub request_id: &'a RequestId,
    pub input_tokens: &'a [TokenId],
    pub maximum_sequence_tokens: usize,
    /// Prompt tokens reported in the final product usage.
    pub product_prompt_tokens: usize,
    /// Already-committed output tokens replayed as part of a recompute input.
    pub replayed_output_tokens: usize,
    pub request_origin: ExecutorRequestOrigin,
}

impl<'a> ExecutorPrefillAdmission<'a> {
    pub const fn for_startup(
        request_id: &'a RequestId,
        input_tokens: &'a [TokenId],
        maximum_sequence_tokens: usize,
    ) -> Self {
        Self {
            request_id,
            input_tokens,
            maximum_sequence_tokens,
            product_prompt_tokens: input_tokens.len(),
            replayed_output_tokens: 0,
            request_origin: ExecutorRequestOrigin::Startup,
        }
    }

    pub const fn for_diagnostic(
        request_id: &'a RequestId,
        input_tokens: &'a [TokenId],
        maximum_sequence_tokens: usize,
    ) -> Self {
        Self {
            request_id,
            input_tokens,
            maximum_sequence_tokens,
            product_prompt_tokens: input_tokens.len(),
            replayed_output_tokens: 0,
            request_origin: ExecutorRequestOrigin::Diagnostic,
        }
    }

    /// Construct admission for a product request whose execution context may
    /// include output tokens replayed after preemption.
    pub fn for_product_request(
        request_id: &'a RequestId,
        input_tokens: &'a [TokenId],
        maximum_sequence_tokens: usize,
        product_prompt_tokens: usize,
        replayed_output_tokens: usize,
    ) -> Result<Self> {
        let admission = Self {
            request_id,
            input_tokens,
            maximum_sequence_tokens,
            product_prompt_tokens,
            replayed_output_tokens,
            request_origin: ExecutorRequestOrigin::Product,
        };
        admission.validate()?;
        Ok(admission)
    }

    pub fn validate(&self) -> Result<()> {
        if self.input_tokens.is_empty() {
            return Err(FerrumError::request_validation(
                "executor prefill admission requires at least one execution-context token",
            ));
        }
        if self.product_prompt_tokens == 0 {
            return Err(FerrumError::request_validation(
                "executor prefill admission requires at least one product prompt token",
            ));
        }
        let execution_context_tokens = self
            .product_prompt_tokens
            .checked_add(self.replayed_output_tokens)
            .ok_or_else(|| {
                FerrumError::request_validation(
                    "executor prefill product token accounting exceeds usize",
                )
            })?;
        if execution_context_tokens != self.input_tokens.len() {
            return Err(FerrumError::request_validation(format!(
                "executor prefill execution context has {} tokens but product accounting declares {} prompt + {} replayed output",
                self.input_tokens.len(),
                self.product_prompt_tokens,
                self.replayed_output_tokens
            )));
        }
        if self.maximum_sequence_tokens < execution_context_tokens {
            return Err(FerrumError::request_validation(format!(
                "executor prefill sequence ceiling {} does not cover execution context {execution_context_tokens}",
                self.maximum_sequence_tokens
            )));
        }
        Ok(())
    }
}

#[cfg(test)]
mod executor_prefill_admission_tests {
    use super::{ExecutorPrefillAdmission, ExecutorRequestOrigin};
    use ferrum_types::{RequestId, TokenId};

    #[test]
    fn product_accounting_distinguishes_replayed_output_from_prompt() {
        let request_id = RequestId::new();
        let tokens = [1, 2, 3, 4, 5]
            .into_iter()
            .map(TokenId::new)
            .collect::<Vec<_>>();

        let admission =
            ExecutorPrefillAdmission::for_product_request(&request_id, &tokens, 8, 3, 2)
                .expect("recompute accounting must be accepted");

        assert_eq!(admission.product_prompt_tokens, 3);
        assert_eq!(admission.replayed_output_tokens, 2);
        assert_eq!(admission.request_origin, ExecutorRequestOrigin::Product);
        assert_eq!(
            ExecutorPrefillAdmission::for_startup(&request_id, &tokens, 8).request_origin,
            ExecutorRequestOrigin::Startup
        );
        assert_eq!(
            ExecutorPrefillAdmission::for_diagnostic(&request_id, &tokens, 8).request_origin,
            ExecutorRequestOrigin::Diagnostic
        );
        assert_eq!(ExecutorRequestOrigin::Product.namespace(), "product");
        assert_eq!(ExecutorRequestOrigin::Startup.namespace(), "startup");
        assert_eq!(ExecutorRequestOrigin::Diagnostic.namespace(), "diagnostic");
        assert_eq!(
            ExecutorRequestOrigin::from_namespaced_request_identity("request.product.123"),
            Some(ExecutorRequestOrigin::Product)
        );
        assert_eq!(
            ExecutorRequestOrigin::from_namespaced_request_identity("request.startup.123"),
            Some(ExecutorRequestOrigin::Startup)
        );
        assert_eq!(
            ExecutorRequestOrigin::from_namespaced_request_identity("request.diagnostic.123"),
            Some(ExecutorRequestOrigin::Diagnostic)
        );
        assert_eq!(
            ExecutorRequestOrigin::from_namespaced_request_identity("request.product."),
            None
        );
        assert_eq!(
            ExecutorRequestOrigin::from_namespaced_request_identity("request/external"),
            None
        );
    }

    #[test]
    fn product_accounting_rejects_context_drift_and_short_ceiling() {
        let request_id = RequestId::new();
        let tokens = [1, 2, 3].into_iter().map(TokenId::new).collect::<Vec<_>>();

        assert!(
            ExecutorPrefillAdmission::for_product_request(&request_id, &tokens, 3, 2, 0).is_err()
        );
        assert!(
            ExecutorPrefillAdmission::for_product_request(&request_id, &tokens, 2, 2, 1).is_err()
        );
    }
}

/// Scheduler-visible proof that an executor retained request and sequence
/// authority for future scheduler-owned prefill chunks.
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct ExecutorPrefillAdmissionReceipt {
    pub request_id: RequestId,
}

/// Stable scheduler-facing projection of one plan-runtime capacity domain.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
pub struct ExecutorAdmissionEpochs {
    pub coordinator_id: NonZeroU64,
    pub release_epoch: u64,
    pub capacity_epoch: u64,
}

impl ExecutorAdmissionEpochs {
    pub const fn new(coordinator_id: NonZeroU64, release_epoch: u64, capacity_epoch: u64) -> Self {
        Self {
            coordinator_id,
            release_epoch,
            capacity_epoch,
        }
    }

    pub fn from_capacity(epochs: crate::vnext::CapacityEpochs) -> Self {
        Self::new(
            NonZeroU64::new(epochs.coordinator_id().get())
                .expect("core-issued admission coordinator ids are non-zero"),
            epochs.release_epoch(),
            epochs.capacity_epoch(),
        )
    }
}

type ExecutorCapacityWaitFuture =
    Pin<Box<dyn Future<Output = Result<ExecutorAdmissionEpochs>> + Send + 'static>>;

/// Type-erased, single-use registration for one plan-runtime capacity wait.
///
/// Registration is created synchronously so the executor can subscribe before
/// the engine releases its iteration lock. Awaiting it never grants resources;
/// it only returns fresh epochs that permit another authoritative admission
/// probe.
#[must_use = "capacity wait registrations must be awaited or explicitly dropped"]
pub struct ExecutorCapacityWaitRegistration {
    future: ExecutorCapacityWaitFuture,
}

impl ExecutorCapacityWaitRegistration {
    pub fn new<F>(future: F) -> Self
    where
        F: Future<Output = Result<ExecutorAdmissionEpochs>> + Send + 'static,
    {
        Self {
            future: Box::pin(future),
        }
    }

    pub async fn wait_for_change(self) -> Result<ExecutorAdmissionEpochs> {
        self.future.await
    }
}

/// Pre-submit runtime stage that could not acquire its exact dynamic capacity.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum ExecutorExecutionCapacityStage {
    SequenceExtension,
    StepAdmission,
    SubmissionWave,
}

/// One exact physical pool mutation completed while an executor was trying to
/// make an unsubmitted frontier runnable.
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct ExecutorExecutionMaintenanceMutation {
    pool_id: crate::vnext::DynamicBackingPoolId,
    domain_id: crate::vnext::CapacityDomainId,
    chunk: crate::vnext::BackingChunkIdentity,
    chunk_bytes: u64,
    published_capacity_bytes: u64,
    capacity_epoch: u64,
}

impl ExecutorExecutionMaintenanceMutation {
    pub fn pool_id(&self) -> &crate::vnext::DynamicBackingPoolId {
        &self.pool_id
    }

    pub const fn domain_id(&self) -> crate::vnext::CapacityDomainId {
        self.domain_id
    }

    pub fn chunk(&self) -> &crate::vnext::BackingChunkIdentity {
        &self.chunk
    }

    pub const fn chunk_bytes(&self) -> u64 {
        self.chunk_bytes
    }

    pub const fn published_capacity_bytes(&self) -> u64 {
        self.published_capacity_bytes
    }

    pub const fn capacity_epoch(&self) -> u64 {
        self.capacity_epoch
    }
}

/// Typed proof that a bounded executor call committed relevant backing growth
/// before yielding the same logical frontier back to the scheduler.
///
/// This is deliberately stronger than observing a global capacity epoch. Every
/// mutation is reconstructed from a real growth receipt and bound back to the
/// pool's exact capacity domain. The scheduler may grant one fairness-bounded
/// retry only when this proof is present.
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct ExecutorExecutionMaintenanceProgress {
    attempts: u32,
    coordinator_id: NonZeroU64,
    mutations: Vec<ExecutorExecutionMaintenanceMutation>,
    latest_capacity_epoch: u64,
}

impl ExecutorExecutionMaintenanceProgress {
    pub fn from_growth_receipts(
        attempts: u32,
        observed: ExecutorAdmissionEpochs,
        receipts: &[crate::vnext::DynamicPoolGrowthBatchReceipt],
        pools: &[crate::vnext::DynamicPoolStatus],
    ) -> Result<Self> {
        if attempts == 0 || receipts.is_empty() || receipts.len() > attempts as usize {
            return Err(FerrumError::internal(
                "execution maintenance retry requires bounded, non-empty growth receipts",
            ));
        }

        let mut mutations = Vec::new();
        let mut previous_capacity_epoch = None;
        for receipt in receipts {
            if receipt.coordinator_id().get() != observed.coordinator_id.get() {
                return Err(FerrumError::internal(
                    "execution maintenance receipt belongs to another capacity coordinator",
                ));
            }
            if receipt.growths().is_empty()
                || previous_capacity_epoch
                    .is_some_and(|previous| receipt.capacity_epoch() <= previous)
            {
                return Err(FerrumError::internal(
                    "execution maintenance receipts contain no new ordered capacity mutation",
                ));
            }
            previous_capacity_epoch = Some(receipt.capacity_epoch());

            for growth in receipt.growths() {
                let pool = pools
                    .iter()
                    .find(|pool| pool.pool_id() == growth.pool_id())
                    .ok_or_else(|| {
                        FerrumError::internal(
                            "execution maintenance receipt references an unknown dynamic pool",
                        )
                    })?;
                if growth.chunk().pool_id() != growth.pool_id()
                    || growth.chunk_bytes() == 0
                    || growth.published_capacity_bytes() == 0
                    || growth.capacity_epoch() != receipt.capacity_epoch()
                {
                    return Err(FerrumError::internal(
                        "execution maintenance receipt contains an invalid pool mutation",
                    ));
                }
                if mutations
                    .iter()
                    .any(|mutation: &ExecutorExecutionMaintenanceMutation| {
                        mutation.pool_id() == growth.pool_id() && mutation.chunk() == growth.chunk()
                    })
                {
                    return Err(FerrumError::internal(
                        "execution maintenance receipts repeat one physical pool mutation",
                    ));
                }
                mutations.push(ExecutorExecutionMaintenanceMutation {
                    pool_id: growth.pool_id().clone(),
                    domain_id: pool.domain_id(),
                    chunk: growth.chunk().clone(),
                    chunk_bytes: growth.chunk_bytes(),
                    published_capacity_bytes: growth.published_capacity_bytes(),
                    capacity_epoch: growth.capacity_epoch(),
                });
            }
        }

        let latest_capacity_epoch = previous_capacity_epoch.expect("receipts are non-empty");
        if latest_capacity_epoch > observed.capacity_epoch {
            return Err(FerrumError::internal(
                "execution maintenance receipt is newer than the exported capacity observation",
            ));
        }
        mutations.sort_by(|left, right| {
            (
                left.capacity_epoch,
                left.pool_id.as_str(),
                left.chunk.ordinal(),
                left.chunk.generation(),
            )
                .cmp(&(
                    right.capacity_epoch,
                    right.pool_id.as_str(),
                    right.chunk.ordinal(),
                    right.chunk.generation(),
                ))
        });
        Ok(Self {
            attempts,
            coordinator_id: observed.coordinator_id,
            mutations,
            latest_capacity_epoch,
        })
    }

    pub const fn attempts(&self) -> u32 {
        self.attempts
    }

    pub const fn coordinator_id(&self) -> NonZeroU64 {
        self.coordinator_id
    }

    pub fn mutations(&self) -> &[ExecutorExecutionMaintenanceMutation] {
        &self.mutations
    }

    pub const fn latest_capacity_epoch(&self) -> u64 {
        self.latest_capacity_epoch
    }
}

/// Scheduler-consumable proof binding physical maintenance to the exact
/// logical frontiers whose unsubmitted execution attempt observed it.
///
/// Keeping the request scope inside the proof prevents a caller from attaching
/// one sequence's growth receipt to an unrelated decode cohort.
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct ExecutorExecutionMaintenanceRetry {
    affected_request_ids: Vec<RequestId>,
    progress: ExecutorExecutionMaintenanceProgress,
}

impl ExecutorExecutionMaintenanceRetry {
    fn new(
        affected_request_ids: Vec<RequestId>,
        progress: ExecutorExecutionMaintenanceProgress,
    ) -> Result<Self> {
        let unique = affected_request_ids.iter().collect::<HashSet<_>>();
        if affected_request_ids.is_empty() || unique.len() != affected_request_ids.len() {
            return Err(FerrumError::internal(
                "execution maintenance retry requires unique affected requests",
            ));
        }
        if progress.mutations().is_empty() {
            return Err(FerrumError::internal(
                "execution maintenance retry requires physical mutations",
            ));
        }
        Ok(Self {
            affected_request_ids,
            progress,
        })
    }

    pub fn affected_request_ids(&self) -> &[RequestId] {
        &self.affected_request_ids
    }

    pub const fn progress(&self) -> &ExecutorExecutionMaintenanceProgress {
        &self.progress
    }
}

/// Scheduler-visible proof that an execution attempt was not submitted and
/// must not be retried until one of its exact capacity sources changes.
///
/// This value owns no resource authority. The executor retains the committed
/// request/sequence authority and has already retired any unsubmitted step or
/// submission-wave authority before returning it.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum ExecutorExecutionCapacityEvidenceOwner {
    Logical,
    Backing,
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
enum ExecutorExecutionCapacityEvidenceKind {
    Logical {
        shortfalls: Vec<crate::vnext::CapacityShortfall>,
        #[serde(skip_serializing_if = "Option::is_none")]
        pressure: Option<crate::vnext::DynamicBackingPressure>,
    },
    BackingDeferred {
        blockers: Vec<crate::vnext::DynamicBackingBlocker>,
        #[serde(skip_serializing_if = "Option::is_none")]
        pressure: Option<crate::vnext::DynamicBackingPressure>,
    },
    BackingPressure {
        pressure: crate::vnext::DynamicBackingPressure,
    },
}

/// Exactly one typed owner for an execution-capacity deferral.
///
/// Its fields are private so callers cannot construct an empty or ambiguous
/// logical/backing evidence combination.
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct ExecutorExecutionCapacityEvidence {
    owner: ExecutorExecutionCapacityEvidenceOwner,
    #[serde(flatten)]
    kind: ExecutorExecutionCapacityEvidenceKind,
    #[serde(skip_serializing_if = "Option::is_none")]
    maintenance_boundary: Option<crate::vnext::DynamicPoolMaintenanceBoundaryReceipt>,
}

impl ExecutorExecutionCapacityEvidence {
    fn logical(shortfalls: Vec<crate::vnext::CapacityShortfall>) -> Result<Self> {
        Self::logical_with_pressure(shortfalls, None)
    }

    fn logical_with_pressure(
        shortfalls: Vec<crate::vnext::CapacityShortfall>,
        pressure: Option<crate::vnext::DynamicBackingPressure>,
    ) -> Result<Self> {
        if shortfalls.is_empty() {
            return Err(FerrumError::internal(
                "logical execution deferral requires at least one shortfall",
            ));
        }
        Ok(Self {
            owner: ExecutorExecutionCapacityEvidenceOwner::Logical,
            kind: ExecutorExecutionCapacityEvidenceKind::Logical {
                shortfalls,
                pressure,
            },
            maintenance_boundary: None,
        })
    }

    fn backing_deferred(blockers: Vec<crate::vnext::DynamicBackingBlocker>) -> Result<Self> {
        Self::backing_deferred_with_pressure(blockers, None)
    }

    fn backing_deferred_with_pressure(
        blockers: Vec<crate::vnext::DynamicBackingBlocker>,
        pressure: Option<crate::vnext::DynamicBackingPressure>,
    ) -> Result<Self> {
        if blockers.is_empty() {
            return Err(FerrumError::internal(
                "physical execution deferral requires at least one backing blocker",
            ));
        }
        Ok(Self {
            owner: ExecutorExecutionCapacityEvidenceOwner::Backing,
            kind: ExecutorExecutionCapacityEvidenceKind::BackingDeferred { blockers, pressure },
            maintenance_boundary: None,
        })
    }

    fn direct_backing_pressure(pressure: crate::vnext::DynamicBackingPressure) -> Self {
        Self {
            owner: ExecutorExecutionCapacityEvidenceOwner::Backing,
            kind: ExecutorExecutionCapacityEvidenceKind::BackingPressure { pressure },
            maintenance_boundary: None,
        }
    }

    pub const fn owner(&self) -> ExecutorExecutionCapacityEvidenceOwner {
        self.owner
    }

    pub fn shortfalls(&self) -> &[crate::vnext::CapacityShortfall] {
        match &self.kind {
            ExecutorExecutionCapacityEvidenceKind::Logical { shortfalls, .. } => shortfalls,
            ExecutorExecutionCapacityEvidenceKind::BackingDeferred { .. }
            | ExecutorExecutionCapacityEvidenceKind::BackingPressure { .. } => &[],
        }
    }

    pub fn backing_blockers(&self) -> &[crate::vnext::DynamicBackingBlocker] {
        match &self.kind {
            ExecutorExecutionCapacityEvidenceKind::BackingDeferred { blockers, .. } => blockers,
            ExecutorExecutionCapacityEvidenceKind::Logical { .. }
            | ExecutorExecutionCapacityEvidenceKind::BackingPressure { .. } => &[],
        }
    }

    pub const fn backing_pressure(&self) -> Option<&crate::vnext::DynamicBackingPressure> {
        match &self.kind {
            ExecutorExecutionCapacityEvidenceKind::Logical { pressure, .. }
            | ExecutorExecutionCapacityEvidenceKind::BackingDeferred { pressure, .. } => {
                pressure.as_ref()
            }
            ExecutorExecutionCapacityEvidenceKind::BackingPressure { pressure } => Some(pressure),
        }
    }

    pub const fn maintenance_boundary(
        &self,
    ) -> Option<&crate::vnext::DynamicPoolMaintenanceBoundaryReceipt> {
        self.maintenance_boundary.as_ref()
    }

    fn with_maintenance_boundary(
        mut self,
        boundary: Option<crate::vnext::DynamicPoolMaintenanceBoundaryReceipt>,
    ) -> Result<Self> {
        match (self.backing_pressure(), boundary.as_ref()) {
            (
                Some(crate::vnext::DynamicBackingPressure::DeviceCapacity(pressure)),
                Some(boundary),
            ) if pressure == boundary.pressure() && !boundary.reclaim_sufficient() => {}
            (Some(crate::vnext::DynamicBackingPressure::PoolResident(_)), None) | (None, None) => {}
            (Some(crate::vnext::DynamicBackingPressure::DeviceCapacity(_)), None) => {
                return Err(FerrumError::internal(
                    "device-capacity execution maintenance lost its boundary receipt",
                ));
            }
            _ => {
                return Err(FerrumError::internal(
                    "execution maintenance boundary differs from its blocked pressure",
                ));
            }
        }
        self.maintenance_boundary = boundary;
        Ok(self)
    }

    fn has_relevant_mutation(&self, mutation: &ExecutorExecutionMaintenanceMutation) -> bool {
        let logical_matches = |shortfalls: &[crate::vnext::CapacityShortfall]| {
            shortfalls.iter().any(|shortfall| {
                shortfall.kind() == crate::vnext::CapacityShortfallKind::BackingGrowthRequired
                    && shortfall.domain() == Some(mutation.domain_id())
            })
        };
        let backing_matches = |blockers: &[crate::vnext::DynamicBackingBlocker]| {
            blockers.iter().any(|blocker| {
                blocker.pool_id() == mutation.pool_id()
                    && blocker.domain_id() == mutation.domain_id()
            })
        };
        match &self.kind {
            ExecutorExecutionCapacityEvidenceKind::Logical { shortfalls, .. } => {
                logical_matches(shortfalls)
            }
            ExecutorExecutionCapacityEvidenceKind::BackingDeferred { blockers, .. } => {
                backing_matches(blockers)
            }
            ExecutorExecutionCapacityEvidenceKind::BackingPressure { .. } => false,
        }
    }
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct ExecutorExecutionCapacityDeferral {
    observed: ExecutorAdmissionEpochs,
    wait_condition: crate::vnext::CapacityWaitCondition,
    stage: ExecutorExecutionCapacityStage,
    evidence: ExecutorExecutionCapacityEvidence,
    #[serde(skip_serializing_if = "Option::is_none")]
    maintenance_retry: Option<ExecutorExecutionMaintenanceRetry>,
}

impl ExecutorExecutionCapacityDeferral {
    fn with_evidence(
        observed: ExecutorAdmissionEpochs,
        wait_condition: crate::vnext::CapacityWaitCondition,
        stage: ExecutorExecutionCapacityStage,
        evidence: ExecutorExecutionCapacityEvidence,
    ) -> Result<Self> {
        if wait_condition.coordinator_id().get() != observed.coordinator_id.get() {
            return Err(ferrum_types::FerrumError::request_validation(
                "executor execution deferral belongs to a different capacity coordinator",
            ));
        }
        Ok(Self {
            observed,
            wait_condition,
            stage,
            evidence,
            maintenance_retry: None,
        })
    }

    /// Construct a physical deferral for an executor that observes device or
    /// pool pressure directly rather than through Ferrum's backing allocator.
    pub fn from_backing_pressure(
        observed: ExecutorAdmissionEpochs,
        wait_condition: crate::vnext::CapacityWaitCondition,
        pressure: crate::vnext::DynamicBackingPressure,
        stage: ExecutorExecutionCapacityStage,
    ) -> Result<Self> {
        Self::with_evidence(
            observed,
            wait_condition,
            stage,
            ExecutorExecutionCapacityEvidence::direct_backing_pressure(pressure),
        )
    }

    pub fn from_admission(
        deferred: &crate::vnext::AdmissionDeferred,
        stage: ExecutorExecutionCapacityStage,
    ) -> Result<Self> {
        if deferred.action() != crate::vnext::DeferredAction::WaitForRelease {
            return Err(ferrum_types::FerrumError::internal(
                "execution capacity deferral must be reduced to WaitForRelease before export",
            ));
        }
        let evidence = ExecutorExecutionCapacityEvidence::logical(deferred.blockers().to_vec())?;
        Self::with_evidence(
            ExecutorAdmissionEpochs::from_capacity(deferred.epochs()),
            deferred.wait_condition().clone(),
            stage,
            evidence,
        )
    }

    /// Export an unresolved logical backing-growth decision after the
    /// executor's bounded in-call maintenance attempts are exhausted.
    ///
    /// This remains a pre-submit scheduling deferral. The bound controls only
    /// how much allocator maintenance one executor call may perform; it must
    /// not turn temporary capacity pressure into a terminal request failure.
    pub fn from_pending_maintenance(
        deferred: &crate::vnext::AdmissionDeferred,
        stage: ExecutorExecutionCapacityStage,
    ) -> Result<Self> {
        if deferred.action() != crate::vnext::DeferredAction::AwaitBackingGrowth {
            return Err(ferrum_types::FerrumError::internal(
                "pending execution maintenance must await backing growth",
            ));
        }
        let evidence = ExecutorExecutionCapacityEvidence::logical(deferred.blockers().to_vec())?;
        Self::with_evidence(
            ExecutorAdmissionEpochs::from_capacity(deferred.epochs()),
            deferred.wait_condition().clone(),
            stage,
            evidence,
        )
    }

    /// Export unresolved physical backing pressure without discarding its
    /// exact pool/domain evidence.
    pub fn from_backing(
        deferred: &crate::vnext::DynamicBackingDeferred,
        stage: ExecutorExecutionCapacityStage,
    ) -> Result<Self> {
        let evidence =
            ExecutorExecutionCapacityEvidence::backing_deferred(deferred.blockers().to_vec())?;
        Self::with_evidence(
            ExecutorAdmissionEpochs::from_capacity(deferred.epochs()),
            deferred.wait_condition().clone(),
            stage,
            evidence,
        )
    }

    /// Attach a scheduler retry only when this call produced at least one real
    /// physical mutation relevant to the final blocker. An empty or unrelated
    /// receipt set is an ordinary typed wait, not an internal error.
    pub fn with_relevant_maintenance_retry(
        mut self,
        attempts: u32,
        receipts: &[crate::vnext::DynamicPoolGrowthBatchReceipt],
        pools: &[crate::vnext::DynamicPoolStatus],
        affected_request_ids: Vec<RequestId>,
    ) -> Result<Self> {
        if receipts.is_empty() {
            return Ok(self);
        }
        let progress = ExecutorExecutionMaintenanceProgress::from_growth_receipts(
            attempts,
            self.observed,
            receipts,
            pools,
        )?;
        if progress.coordinator_id() != self.observed.coordinator_id
            || progress.latest_capacity_epoch() > self.observed.capacity_epoch
            || progress.mutations().is_empty()
        {
            return Err(FerrumError::internal(
                "execution maintenance progress does not match the exported deferral",
            ));
        }
        let relevant_mutation = progress
            .mutations()
            .iter()
            .any(|mutation| self.evidence.has_relevant_mutation(mutation));
        if !relevant_mutation {
            return Ok(self);
        }
        let retry = ExecutorExecutionMaintenanceRetry::new(affected_request_ids, progress)?;
        if self.stage == ExecutorExecutionCapacityStage::SequenceExtension
            && retry.affected_request_ids().len() != 1
        {
            return Err(FerrumError::internal(
                "sequence-extension maintenance retry must affect exactly one request",
            ));
        }
        self.maintenance_retry = Some(retry);
        Ok(self)
    }

    pub fn from_admission_maintenance(
        source: &crate::vnext::AdmissionDeferred,
        observed: ExecutorAdmissionEpochs,
        wait_condition: crate::vnext::CapacityWaitCondition,
        pressure: crate::vnext::DynamicBackingPressure,
        maintenance_boundary: Option<crate::vnext::DynamicPoolMaintenanceBoundaryReceipt>,
        stage: ExecutorExecutionCapacityStage,
    ) -> Result<Self> {
        if source.action() != crate::vnext::DeferredAction::AwaitBackingGrowth {
            return Err(ferrum_types::FerrumError::internal(
                "execution maintenance source must await backing growth",
            ));
        }
        let evidence = ExecutorExecutionCapacityEvidence::logical_with_pressure(
            source.blockers().to_vec(),
            Some(pressure),
        )?
        .with_maintenance_boundary(maintenance_boundary)?;
        if evidence.maintenance_boundary().is_some_and(|boundary| {
            boundary.coordinator_id().get() != observed.coordinator_id.get()
        }) {
            return Err(FerrumError::internal(
                "execution maintenance boundary belongs to another coordinator",
            ));
        }
        Self::with_evidence(observed, wait_condition, stage, evidence)
    }

    pub fn from_backing_maintenance(
        source: &crate::vnext::DynamicBackingDeferred,
        observed: ExecutorAdmissionEpochs,
        wait_condition: crate::vnext::CapacityWaitCondition,
        pressure: crate::vnext::DynamicBackingPressure,
        maintenance_boundary: Option<crate::vnext::DynamicPoolMaintenanceBoundaryReceipt>,
        stage: ExecutorExecutionCapacityStage,
    ) -> Result<Self> {
        let evidence = ExecutorExecutionCapacityEvidence::backing_deferred_with_pressure(
            source.blockers().to_vec(),
            Some(pressure),
        )?
        .with_maintenance_boundary(maintenance_boundary)?;
        if evidence.maintenance_boundary().is_some_and(|boundary| {
            boundary.coordinator_id().get() != observed.coordinator_id.get()
        }) {
            return Err(FerrumError::internal(
                "execution maintenance boundary belongs to another coordinator",
            ));
        }
        Self::with_evidence(observed, wait_condition, stage, evidence)
    }

    pub const fn observed(&self) -> ExecutorAdmissionEpochs {
        self.observed
    }

    pub fn wait_condition(&self) -> &crate::vnext::CapacityWaitCondition {
        &self.wait_condition
    }

    pub const fn stage(&self) -> ExecutorExecutionCapacityStage {
        self.stage
    }

    pub const fn evidence(&self) -> &ExecutorExecutionCapacityEvidence {
        &self.evidence
    }

    pub fn shortfalls(&self) -> &[crate::vnext::CapacityShortfall] {
        self.evidence.shortfalls()
    }

    pub fn backing_blockers(&self) -> &[crate::vnext::DynamicBackingBlocker] {
        self.evidence.backing_blockers()
    }

    pub const fn backing_pressure(&self) -> Option<&crate::vnext::DynamicBackingPressure> {
        self.evidence.backing_pressure()
    }

    pub const fn maintenance_boundary(
        &self,
    ) -> Option<&crate::vnext::DynamicPoolMaintenanceBoundaryReceipt> {
        self.evidence.maintenance_boundary()
    }

    pub fn maintenance_retry(&self) -> Option<&ExecutorExecutionMaintenanceRetry> {
        self.maintenance_retry.as_ref()
    }

    /// Validate the bound retry scope against the authoritative logical
    /// frontiers in the current engine call.
    pub fn validated_maintenance_retry_scope(
        &self,
        current_request_ids: &[RequestId],
    ) -> Result<Option<&ExecutorExecutionMaintenanceRetry>> {
        let Some(retry) = self.maintenance_retry.as_ref() else {
            return Ok(None);
        };
        let current = current_request_ids.iter().collect::<HashSet<_>>();
        if current_request_ids.is_empty() || current.len() != current_request_ids.len() {
            return Err(FerrumError::internal(
                "execution maintenance retry received an invalid current request cohort",
            ));
        }
        let affected = retry.affected_request_ids().iter().collect::<HashSet<_>>();
        if !affected.is_subset(&current) {
            return Err(FerrumError::internal(
                "execution maintenance retry affects a request outside the current cohort",
            ));
        }
        match self.stage {
            ExecutorExecutionCapacityStage::SequenceExtension => {
                if affected.len() != 1 {
                    return Err(FerrumError::internal(
                        "sequence-extension maintenance retry must affect one current request",
                    ));
                }
            }
            ExecutorExecutionCapacityStage::StepAdmission
            | ExecutorExecutionCapacityStage::SubmissionWave => {
                if affected != current {
                    return Err(FerrumError::internal(
                        "cohort maintenance retry must cover the complete current cohort",
                    ));
                }
            }
        }
        Ok(Some(retry))
    }

    /// Return a strictly smaller, capacity-informed prefill width.
    ///
    /// This is a cold pressure-path hint, not allocator authority. The caller
    /// must probe the returned prefix through normal typed admission before any
    /// provider encode or device submission. A bounded reduction prevents a
    /// large frontier from producing an unbounded sequence of near-identical
    /// probes when the shortfall is small.
    pub fn narrower_prefill_tokens(&self, attempted_tokens: usize) -> Option<usize> {
        if attempted_tokens <= 1 {
            return None;
        }
        let maximum_next = attempted_tokens
            .saturating_sub(attempted_tokens.div_ceil(4))
            .max(1);
        let proportional = self
            .shortfalls()
            .iter()
            .filter_map(|shortfall| {
                let requested = shortfall.requested().get();
                let available = shortfall.available().get();
                (requested > available).then(|| {
                    let scaled = (attempted_tokens as u128).saturating_mul(available as u128)
                        / requested as u128;
                    usize::try_from(scaled)
                        .unwrap_or(usize::MAX)
                        .clamp(1, attempted_tokens - 1)
                })
            })
            .min();
        Some(
            proportional
                .unwrap_or_else(|| attempted_tokens.div_ceil(2))
                .min(maximum_next)
                .max(1),
        )
    }
}

/// Scheduler-visible proof that execution is temporarily blocked by a
/// Request-lifetime state hazard rather than by physical capacity.
///
/// The embedded hazard evidence retains the exact plan-local coordinator and
/// supports subscribe-before-recheck waiter registration. The request cohort
/// is the product identity projection used by the scheduler; it must stay
/// separate from the allocator's internal request-authority ids.
#[derive(Debug, Clone, Serialize)]
pub struct ExecutorRequestStateDeferral {
    stage: ExecutorExecutionCapacityStage,
    request_ids: Vec<RequestId>,
    hazard: crate::vnext::RequestStateHazardDeferral,
}

impl ExecutorRequestStateDeferral {
    pub fn new(
        stage: ExecutorExecutionCapacityStage,
        request_ids: Vec<RequestId>,
        hazard: crate::vnext::RequestStateHazardDeferral,
    ) -> Result<Self> {
        let unique = request_ids.iter().collect::<HashSet<_>>();
        if request_ids.is_empty() || unique.len() != request_ids.len() {
            return Err(FerrumError::internal(
                "request-state execution deferral requires a non-empty unique product cohort",
            ));
        }
        if hazard.blockers().is_empty() {
            return Err(FerrumError::internal(
                "request-state execution deferral requires exact blockers",
            ));
        }
        Ok(Self {
            stage,
            request_ids,
            hazard,
        })
    }

    pub const fn stage(&self) -> ExecutorExecutionCapacityStage {
        self.stage
    }

    pub fn request_ids(&self) -> &[RequestId] {
        &self.request_ids
    }

    pub const fn hazard(&self) -> &crate::vnext::RequestStateHazardDeferral {
        &self.hazard
    }

    pub fn register_waiter(&self) -> Result<crate::vnext::RequestStateHazardWaitRegistration> {
        self.hazard
            .register_waiter()
            .map_err(|error| FerrumError::backend(error.to_string()))
    }
}

/// A pre-submit execution frontier can be blocked by independently evolving
/// sources. Capacity waits participate in scheduler pressure/yield policy;
/// Request-state waits never do and resume only from their exact hazard
/// coordinator.
#[derive(Debug, Clone, Serialize)]
#[serde(tag = "reason", content = "evidence", rename_all = "snake_case")]
pub enum ExecutorExecutionDeferral {
    Capacity(ExecutorExecutionCapacityDeferral),
    RequestState(ExecutorRequestStateDeferral),
}

impl ExecutorExecutionDeferral {
    pub const fn stage(&self) -> ExecutorExecutionCapacityStage {
        match self {
            Self::Capacity(deferral) => deferral.stage(),
            Self::RequestState(deferral) => deferral.stage(),
        }
    }

    pub const fn as_capacity(&self) -> Option<&ExecutorExecutionCapacityDeferral> {
        match self {
            Self::Capacity(deferral) => Some(deferral),
            Self::RequestState(_) => None,
        }
    }

    pub const fn as_request_state(&self) -> Option<&ExecutorRequestStateDeferral> {
        match self {
            Self::Capacity(_) => None,
            Self::RequestState(deferral) => Some(deferral),
        }
    }
}

impl From<ExecutorExecutionCapacityDeferral> for ExecutorExecutionDeferral {
    fn from(deferral: ExecutorExecutionCapacityDeferral) -> Self {
        Self::Capacity(deferral)
    }
}

impl From<ExecutorRequestStateDeferral> for ExecutorExecutionDeferral {
    fn from(deferral: ExecutorRequestStateDeferral) -> Self {
        Self::RequestState(deferral)
    }
}

#[cfg(test)]
mod execution_capacity_deferral_tests {
    use super::{
        ExecutorAdmissionEpochs, ExecutorExecutionCapacityDeferral,
        ExecutorExecutionCapacityEvidenceOwner, ExecutorExecutionCapacityStage,
        ExecutorExecutionMaintenanceProgress, ExecutorExecutionMaintenanceRetry,
    };
    use crate::vnext::{
        CapacityAvailabilityEpoch, CapacityAvailabilitySource, CapacityWaitCondition,
        DeviceCapacityPressure, DeviceCapacityPressureScope, DynamicBackingPressure,
    };
    use ferrum_types::RequestId;
    use std::num::NonZeroU64;

    fn test_progress() -> ExecutorExecutionMaintenanceProgress {
        ExecutorExecutionMaintenanceProgress {
            attempts: 1,
            coordinator_id: NonZeroU64::new(19).unwrap(),
            mutations: Vec::new(),
            latest_capacity_epoch: 5,
        }
    }

    fn test_deferral(stage: ExecutorExecutionCapacityStage) -> ExecutorExecutionCapacityDeferral {
        let observed =
            CapacityAvailabilityEpoch::new(CapacityAvailabilitySource::ActiveSequenceSlots, 7)
                .unwrap();
        let condition = CapacityWaitCondition::from_observation(19, vec![observed]).unwrap();
        ExecutorExecutionCapacityDeferral::from_backing_pressure(
            ExecutorAdmissionEpochs::new(NonZeroU64::new(19).unwrap(), 3, 5),
            condition,
            test_pressure(),
            stage,
        )
        .unwrap()
    }

    fn test_pressure() -> DynamicBackingPressure {
        DeviceCapacityPressure::new(
            DeviceCapacityPressureScope::PlanBudget,
            "device.execution-capacity-test".to_owned(),
            1,
            1,
            1,
            1,
            1,
        )
        .unwrap()
        .into()
    }

    #[test]
    fn prefill_narrowing_is_strict_bounded_and_stops_at_one_token() {
        let observed =
            CapacityAvailabilityEpoch::new(CapacityAvailabilitySource::ActiveSequenceSlots, 7)
                .unwrap();
        let condition = CapacityWaitCondition::from_observation(19, vec![observed]).unwrap();
        let deferred = ExecutorExecutionCapacityDeferral::from_backing_pressure(
            ExecutorAdmissionEpochs::new(NonZeroU64::new(19).unwrap(), 3, 5),
            condition,
            test_pressure(),
            ExecutorExecutionCapacityStage::StepAdmission,
        )
        .unwrap();

        assert_eq!(deferred.narrower_prefill_tokens(342), Some(171));
        assert_eq!(deferred.narrower_prefill_tokens(2), Some(1));
        assert_eq!(deferred.narrower_prefill_tokens(1), None);
    }

    #[test]
    fn backing_pressure_serializes_one_typed_evidence_owner() {
        let deferred = test_deferral(ExecutorExecutionCapacityStage::SequenceExtension);
        let serialized = serde_json::to_value(&deferred).unwrap();

        assert_eq!(
            deferred.evidence().owner(),
            ExecutorExecutionCapacityEvidenceOwner::Backing
        );
        assert!(deferred.shortfalls().is_empty());
        assert!(deferred.backing_blockers().is_empty());
        assert!(deferred.backing_pressure().is_some());
        assert_eq!(serialized["evidence"]["owner"], "backing");
        assert_eq!(serialized["evidence"]["kind"], "backing_pressure");
        assert!(serialized["evidence"]["pressure"].is_object());
    }

    #[test]
    fn empty_maintenance_receipts_remain_an_ordinary_typed_deferral() {
        let request_id = RequestId::new();
        let deferred = test_deferral(ExecutorExecutionCapacityStage::SequenceExtension)
            .with_relevant_maintenance_retry(2, &[], &[], vec![request_id])
            .unwrap();

        assert!(deferred.maintenance_retry().is_none());
    }

    #[test]
    fn maintenance_retry_rejects_empty_duplicate_or_unproven_scope() {
        let request_id = RequestId::new();
        assert!(ExecutorExecutionMaintenanceRetry::new(Vec::new(), test_progress()).is_err());
        assert!(ExecutorExecutionMaintenanceRetry::new(
            vec![request_id.clone(), request_id.clone()],
            test_progress(),
        )
        .is_err());
        assert!(ExecutorExecutionMaintenanceRetry::new(vec![request_id], test_progress()).is_err());
    }

    #[test]
    fn maintenance_retry_scope_is_fail_closed_for_sequence_and_cohort_stages() {
        let first = RequestId::new();
        let second = RequestId::new();
        let retry = |affected_request_ids| ExecutorExecutionMaintenanceRetry {
            affected_request_ids,
            progress: test_progress(),
        };

        let mut sequence = test_deferral(ExecutorExecutionCapacityStage::SequenceExtension);
        sequence.maintenance_retry = Some(retry(vec![second.clone()]));
        assert_eq!(
            sequence
                .validated_maintenance_retry_scope(&[first.clone(), second.clone()])
                .unwrap()
                .unwrap()
                .affected_request_ids(),
            [second.clone()]
        );
        sequence.maintenance_retry = Some(retry(vec![first.clone(), second.clone()]));
        assert!(sequence
            .validated_maintenance_retry_scope(&[first.clone(), second.clone()])
            .is_err());

        let mut cohort = test_deferral(ExecutorExecutionCapacityStage::SubmissionWave);
        cohort.maintenance_retry = Some(retry(vec![second.clone()]));
        assert!(cohort
            .validated_maintenance_retry_scope(&[first.clone(), second.clone()])
            .is_err());
        cohort.maintenance_retry = Some(retry(vec![first.clone(), second.clone()]));
        assert!(cohort
            .validated_maintenance_retry_scope(&[first, second])
            .unwrap()
            .is_some());
    }
}

/// Capacity-aware batch decode result.
///
/// `Deferred` is only legal before provider encode or device submission. All
/// possibly-submitted failures remain ordinary errors and retain their typed
/// fence/recovery authority inside the executor.
pub enum ExecutorBatchDecodeOutcome {
    Completed(Vec<DecodeOutput>),
    Deferred(ExecutorExecutionDeferral),
}

/// Capacity-aware, tensor-free batch decode result for a plan runtime.
pub enum PlanRuntimeBatchDecodeOutcome {
    Completed(Vec<PlanRuntimeDecodeOutput>),
    Deferred(ExecutorExecutionDeferral),
}

/// Capacity-aware result for one tensor-free plan-runtime prefill frontier.
pub struct PlanRuntimePrefillCompletion {
    output: PlanRuntimePrefillOutput,
    planned_chunk: PrefillChunk,
    completed_chunk: PrefillChunk,
    capacity_probe_count: u32,
}

impl PlanRuntimePrefillCompletion {
    pub fn new(
        output: PlanRuntimePrefillOutput,
        planned_chunk: PrefillChunk,
        completed_chunk: PrefillChunk,
        capacity_probe_count: u32,
    ) -> Result<Self> {
        validate_prefill_completion_shape(planned_chunk, completed_chunk, capacity_probe_count)?;
        Ok(Self {
            output,
            planned_chunk,
            completed_chunk,
            capacity_probe_count,
        })
    }

    pub fn exact(output: PlanRuntimePrefillOutput, chunk: PrefillChunk) -> Self {
        Self {
            output,
            planned_chunk: chunk,
            completed_chunk: chunk,
            capacity_probe_count: 0,
        }
    }

    pub const fn planned_chunk(&self) -> PrefillChunk {
        self.planned_chunk
    }

    pub const fn completed_chunk(&self) -> PrefillChunk {
        self.completed_chunk
    }

    pub const fn capacity_probe_count(&self) -> u32 {
        self.capacity_probe_count
    }

    pub fn output(&self) -> &PlanRuntimePrefillOutput {
        &self.output
    }

    pub fn validate_for(
        &self,
        expected_request_id: &RequestId,
        expected_planned_chunk: PrefillChunk,
        vocabulary_size: usize,
    ) -> Result<()> {
        if self.planned_chunk != expected_planned_chunk {
            return Err(FerrumError::backend(format!(
                "plan runtime completed prefill frontier {:?}, expected {:?}",
                self.planned_chunk.range(),
                expected_planned_chunk.range()
            )));
        }
        validate_prefill_completion_shape(
            self.planned_chunk,
            self.completed_chunk,
            self.capacity_probe_count,
        )?;
        self.output.validate_for_completion(
            expected_request_id,
            self.completed_chunk,
            vocabulary_size,
        )
    }

    pub fn into_parts(self) -> (PlanRuntimePrefillOutput, PrefillChunk, PrefillChunk, u32) {
        (
            self.output,
            self.planned_chunk,
            self.completed_chunk,
            self.capacity_probe_count,
        )
    }
}

pub enum PlanRuntimePrefillOutcome {
    Completed(PlanRuntimePrefillCompletion),
    Deferred(ExecutorExecutionDeferral),
}

pub enum PlanRuntimeBatchPrefillOutcome {
    Completed(Vec<PlanRuntimePrefillCompletion>),
    NotSubmitted(ExecutorExecutionDeferral),
    Unsupported,
}

/// Capacity-aware result for one planned prefill frontier.
///
/// `Deferred` is only legal before provider encode or device submission. The
/// executor may complete a strict prefix after typed capacity probes; the
/// scheduler commits only that prefix and learns the narrower execution ceiling.
pub struct ExecutorPrefillCompletion {
    output: PrefillOutput,
    planned_chunk: PrefillChunk,
    completed_chunk: PrefillChunk,
    capacity_probe_count: u32,
}

impl ExecutorPrefillCompletion {
    pub fn new(
        output: PrefillOutput,
        planned_chunk: PrefillChunk,
        completed_chunk: PrefillChunk,
        capacity_probe_count: u32,
    ) -> Result<Self> {
        validate_prefill_completion_shape(planned_chunk, completed_chunk, capacity_probe_count)?;
        Ok(Self {
            output,
            planned_chunk,
            completed_chunk,
            capacity_probe_count,
        })
    }

    pub fn exact(output: PrefillOutput, chunk: PrefillChunk) -> Self {
        Self {
            output,
            planned_chunk: chunk,
            completed_chunk: chunk,
            capacity_probe_count: 0,
        }
    }

    pub const fn planned_chunk(&self) -> PrefillChunk {
        self.planned_chunk
    }

    pub const fn completed_chunk(&self) -> PrefillChunk {
        self.completed_chunk
    }

    pub const fn capacity_probe_count(&self) -> u32 {
        self.capacity_probe_count
    }

    pub fn into_parts(self) -> (PrefillOutput, PrefillChunk, PrefillChunk, u32) {
        (
            self.output,
            self.planned_chunk,
            self.completed_chunk,
            self.capacity_probe_count,
        )
    }
}

fn validate_prefill_completion_shape(
    planned_chunk: PrefillChunk,
    completed_chunk: PrefillChunk,
    capacity_probe_count: u32,
) -> Result<()> {
    if completed_chunk.tokens_processed() != planned_chunk.tokens_processed()
        || completed_chunk.total_prompt_tokens() != planned_chunk.total_prompt_tokens()
        || completed_chunk.tokens_to_process() > planned_chunk.tokens_to_process()
    {
        return Err(ferrum_types::FerrumError::internal(
            "completed prefill chunk is not a non-empty prefix of its planned chunk",
        ));
    }
    if completed_chunk != planned_chunk && capacity_probe_count == 0 {
        return Err(ferrum_types::FerrumError::internal(
            "partial prefill completion requires a failed capacity probe",
        ));
    }
    Ok(())
}

pub enum ExecutorPrefillOutcome {
    Completed(ExecutorPrefillCompletion),
    Deferred(ExecutorExecutionDeferral),
}

/// Transactional result of attempting one physical prefill batch.
///
/// `NotSubmitted` proves that no participant in the batch reached provider
/// encode or device submission. The caller may therefore retry a narrower
/// partition or the existing per-request capacity path without duplicating
/// model work. `Unsupported` keeps the optimization optional for legacy
/// executors while plan-runtime implementations provide the real batch edge.
pub enum ExecutorBatchPrefillOutcome {
    Completed(Vec<ExecutorPrefillCompletion>),
    NotSubmitted(ExecutorExecutionDeferral),
    Unsupported,
}

/// Stage that must advance before a plan-runtime prefill can be admitted.
///
/// This is scheduler evidence, not allocator authority. The executor retains
/// the sealed logical or physical deferral that authorizes maintenance.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum ExecutorPrefillMaintenanceStage {
    LogicalCapacity,
    PhysicalBacking,
}

/// Scheduler-visible reason that a prefill needs plan-runtime backing
/// maintenance. These values are projections only and cannot allocate memory.
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
#[serde(tag = "source", rename_all = "snake_case")]
pub enum ExecutorPrefillMaintenanceBlocker {
    Capacity {
        domain_id: Option<u32>,
        kind: crate::vnext::CapacityShortfallKind,
        requested: u64,
        available: u64,
        current_total: u64,
        maximum_total: u64,
    },
    Backing {
        pool_id: String,
        domain_id: u32,
        lifetime: crate::vnext::DynamicBackingClaimScope,
        reason: crate::vnext::DynamicBackingDeferralReason,
        requested_bytes: u64,
        free_bytes: u64,
        largest_contiguous_bytes: u64,
    },
}

/// Non-authoritative projection of plan-runtime maintenance work.
///
/// The request id is the only handle returned to the engine. Implementations
/// must retain the sealed deferral internally and validate it again when
/// [`ModelExecutor::maintain_prefill_backing`] is called.
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct ExecutorPrefillMaintenanceDeferral {
    request_id: RequestId,
    observed: ExecutorAdmissionEpochs,
    wait_condition: crate::vnext::CapacityWaitCondition,
    stage: ExecutorPrefillMaintenanceStage,
    blockers: Vec<ExecutorPrefillMaintenanceBlocker>,
}

impl ExecutorPrefillMaintenanceDeferral {
    pub fn new(
        request_id: RequestId,
        observed: ExecutorAdmissionEpochs,
        wait_condition: crate::vnext::CapacityWaitCondition,
        stage: ExecutorPrefillMaintenanceStage,
        blockers: Vec<ExecutorPrefillMaintenanceBlocker>,
    ) -> Result<Self> {
        if blockers.is_empty() {
            return Err(ferrum_types::FerrumError::request_validation(
                "executor prefill maintenance deferral requires at least one blocker",
            ));
        }
        if wait_condition.coordinator_id().get() != observed.coordinator_id.get() {
            return Err(ferrum_types::FerrumError::request_validation(
                "executor prefill maintenance wait condition belongs to a different coordinator",
            ));
        }
        Ok(Self {
            request_id,
            observed,
            wait_condition,
            stage,
            blockers,
        })
    }

    pub fn from_admission(
        request_id: &RequestId,
        deferred: &crate::vnext::AdmissionDeferred,
    ) -> Result<Self> {
        if deferred.action() != crate::vnext::DeferredAction::AwaitBackingGrowth {
            return Err(ferrum_types::FerrumError::internal(
                "logical prefill maintenance projection requires AwaitBackingGrowth",
            ));
        }
        let blockers = deferred
            .blockers()
            .iter()
            .map(|blocker| ExecutorPrefillMaintenanceBlocker::Capacity {
                domain_id: blocker.domain().map(|domain| domain.get()),
                kind: blocker.kind(),
                requested: blocker.requested().get(),
                available: blocker.available().get(),
                current_total: blocker.current_total().get(),
                maximum_total: blocker.maximum_total().get(),
            })
            .collect();
        Self::new(
            request_id.clone(),
            ExecutorAdmissionEpochs::from_capacity(deferred.epochs()),
            deferred.wait_condition().clone(),
            ExecutorPrefillMaintenanceStage::LogicalCapacity,
            blockers,
        )
    }

    pub fn from_backing(
        request_id: &RequestId,
        deferred: &crate::vnext::DynamicBackingDeferred,
    ) -> Result<Self> {
        let blockers = deferred
            .blockers()
            .iter()
            .map(|blocker| ExecutorPrefillMaintenanceBlocker::Backing {
                pool_id: blocker.pool_id().as_str().to_string(),
                domain_id: blocker.domain_id().get(),
                lifetime: deferred.scope(),
                reason: blocker.reason(),
                requested_bytes: blocker.requested_bytes(),
                free_bytes: blocker.free_bytes(),
                largest_contiguous_bytes: blocker.largest_contiguous_bytes(),
            })
            .collect();
        Self::new(
            request_id.clone(),
            ExecutorAdmissionEpochs::from_capacity(deferred.epochs()),
            deferred.wait_condition().clone(),
            ExecutorPrefillMaintenanceStage::PhysicalBacking,
            blockers,
        )
    }

    pub fn request_id(&self) -> &RequestId {
        &self.request_id
    }

    pub const fn observed(&self) -> ExecutorAdmissionEpochs {
        self.observed
    }

    pub fn wait_condition(&self) -> &crate::vnext::CapacityWaitCondition {
        &self.wait_condition
    }

    pub const fn stage(&self) -> ExecutorPrefillMaintenanceStage {
        self.stage
    }

    pub fn blockers(&self) -> &[ExecutorPrefillMaintenanceBlocker] {
        &self.blockers
    }
}

/// Result of one bounded plan-runtime backing maintenance attempt.
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
#[serde(tag = "outcome", rename_all = "snake_case")]
pub enum ExecutorPrefillMaintenanceOutcome {
    /// Cancellation won the race before the maintenance task consumed the
    /// retained deferral.
    NoLongerPending,
    /// The physical allocator changed while maintenance was installing its
    /// wait predicate. The scheduler must clear the old backing deferral and
    /// perform one authoritative admission probe, even if publication of the
    /// corresponding capacity epoch is still in flight.
    RetryAdmission { current: ExecutorAdmissionEpochs },
    /// The requested backing is valid but cannot be installed while current
    /// device claims remain live. The scheduler must wait for release evidence
    /// rather than completing the request as an error.
    WaitForRelease {
        current: ExecutorAdmissionEpochs,
        wait_condition: crate::vnext::CapacityWaitCondition,
        pressure: crate::vnext::DynamicBackingPressure,
    },
    /// The executor installed real backing and published the resulting
    /// capacity epoch.
    Maintained {
        current: ExecutorAdmissionEpochs,
        pools_grown: usize,
        allocated_bytes: u64,
        pools_reclaimed: usize,
        chunks_reclaimed: usize,
        reclaimed_bytes: u64,
        /// Exact allocator-issued rebalance receipt. The aggregate counters
        /// above remain for stable metrics and must reconcile with this value.
        rebalance: Option<crate::vnext::DynamicPoolRebalanceReceipt>,
    },
}

/// Typed result of probing plan-runtime prefill capacity.
///
/// `Deferred` and `MaintenanceDeferred` preserve the capacity domains and
/// epochs required by the plan-local dynamic admission queue. They must never
/// be flattened into a generic resource error at the scheduler boundary.
#[derive(Debug, Clone)]
pub enum ExecutorPrefillAdmissionDecision {
    Admitted(ExecutorPrefillAdmissionReceipt),
    Deferred(crate::vnext::AdmissionDeferred),
    MaintenanceDeferred(ExecutorPrefillMaintenanceDeferral),
    PermanentRejected(crate::vnext::AdmissionRejected),
}

/// Core model executor trait focusing on tensor operations
#[async_trait]
pub trait ModelExecutor: Send + Sync {
    /// Plan an optional prompt-tail checkpoint before a prefill chunk is
    /// dispatched. The boundary must lie after this chunk's start and no later
    /// than its end, leaving a legal suffix for logits. None preserves the chunk. Planning
    /// grants no capacity or capture authority and does not report a failed
    /// capacity probe; the caller dispatches the resulting chunk explicitly.
    fn plan_prompt_tail_capture_boundary(&self, _chunk: PrefillChunk) -> Option<PrefixCapturePlan> {
        None
    }

    /// Pure boundary planning for optional sharing. None leaves normal scheduling
    /// unchanged; a returned boundary never grants request or resource authority.
    fn plan_prefix_capture_boundary(
        &self,
        _input: PrefixCaptureBoundary<'_>,
    ) -> Option<PrefixCapturePlan> {
        None
    }

    /// Arm interest only against an already-admitted, exact source incarnation.
    /// No device allocation, capacity reservation, provider encoding, or device
    /// submission is allowed here.
    fn retain_prefix_capture_interest(
        &self,
        _input: PrefixCaptureRequest<'_>,
    ) -> Result<Option<Arc<dyn PrefixCaptureLease>>> {
        Ok(None)
    }

    /// Get model information and metadata
    fn info(&self) -> &ModelInfo;

    /// Selects the single authority for request-lifetime model resources.
    /// Existing executors remain on the transitional legacy-engine path by
    /// default. A runtime that returns `PlanRuntime` must return the shared
    /// runtime's opaque cache handle from prefill/decode and delegate release
    /// of that authority from `release_cache`.
    fn execution_resource_authority(&self) -> ExecutionResourceAuthority {
        ExecutionResourceAuthority::LegacyEngine
    }

    /// Immutable admission limits compiled into this executor's runtime plan.
    /// A PlanRuntime executor must return `Some`; legacy executors may defer to
    /// the engine-owned scheduler and recurrent-state limits.
    fn admission_limits(&self) -> Result<Option<ExecutorAdmissionLimits>> {
        Ok(None)
    }

    /// Returns the immutable product plan that owns planning, provider
    /// selection, and resource authority for a plan-runtime executor.
    /// Legacy executors return `None`; `PlanRuntime` executors must expose the
    /// exact plan used for provisioning and dispatch.
    fn resolved_model_plan(&self) -> Option<&crate::vnext::ResolvedModelPlan> {
        None
    }

    /// Returns the authoritative memory breakdown for a shared plan runtime.
    /// `LegacyEngine` executors return `None`; `PlanRuntime` executors must
    /// return `Some` while they are ready.
    fn plan_runtime_resource_snapshot(&self) -> Result<Option<PlanRuntimeResourceSnapshot>> {
        Ok(None)
    }

    /// Whether this executor's backend can run the unified mixed prefill+decode
    /// forward natively. When false, the engine routes Qwen3-MoE batches through
    /// the legacy split path. Reported by the (backend-aware) executor so the
    /// engine stays backend-agnostic — replaces a `cfg(target_os)` branch that
    /// previously hard-coded "Metal/CPU lack native unified" in the hot path.
    ///
    /// Default false (conservative legacy path); accelerators with a native
    /// unified forward override to true.
    fn supports_native_unified_decode(&self) -> bool {
        false
    }

    /// Per-request KV capacity in tokens when the executor owns a smaller
    /// runtime cache window than the model's declared context length.
    fn kv_capacity(&self) -> Option<usize> {
        None
    }

    /// Device-aware limits selected from the exact compiled memory plan before
    /// static model upload. Legacy executors do not provide this evidence.
    fn startup_memory_plan(&self) -> Option<&ferrum_types::StartupMemoryPlan> {
        None
    }

    /// Installs the product-owned execution event sink before requests start.
    ///
    /// Legacy executors have no typed execution journal and keep the default
    /// no-op. Executors backed by the vNext runtime retain this sink with each
    /// admitted request so node/operation events share the product artifact.
    fn attach_execution_event_sink(&self, _sink: Arc<dyn crate::vnext::ExecutionEventSink>) {}

    /// Current plan-local capacity evidence for scheduler wake suppression.
    /// Legacy-engine executors return `None`; an executor declaring
    /// [`ExecutionResourceAuthority::PlanRuntime`] must return `Some`.
    fn execution_capacity_epochs(&self) -> Result<Option<ExecutorAdmissionEpochs>> {
        Ok(None)
    }

    /// Writes the canonical per-source availability generations into
    /// caller-owned storage and returns the matching global audit epochs.
    /// Executors with typed dynamic admission override this to avoid allocating
    /// on steady scheduler ticks.
    fn write_execution_capacity_snapshot(
        &self,
        availability: &mut Vec<crate::vnext::CapacityAvailabilityEpoch>,
    ) -> Result<Option<ExecutorAdmissionEpochs>> {
        availability.clear();
        self.execution_capacity_epochs()
    }

    /// Synchronously subscribes to every source named by one passive capacity
    /// wait. The returned registration must remain alive until it is awaited or
    /// deliberately cancelled by being dropped.
    ///
    /// Legacy-engine executors return `None`. An executor declaring
    /// [`ExecutionResourceAuthority::PlanRuntime`] must return `Some` for a
    /// wait condition issued by its own admission coordinator.
    fn register_execution_capacity_waiter(
        &self,
        _observed: &crate::vnext::CapacityWaitCondition,
    ) -> Result<Option<ExecutorCapacityWaitRegistration>> {
        Ok(None)
    }

    /// Probe and retain the exact request/sequence authority needed by a
    /// future prefill. No provider encode, kernel launch, or device submit may
    /// occur in this method.
    fn try_admit_prefill(
        &self,
        _input: ExecutorPrefillAdmission<'_>,
    ) -> Result<ExecutorPrefillAdmissionDecision> {
        Err(ferrum_types::FerrumError::unsupported(
            "plan-runtime prefill admission is not implemented",
        ))
    }

    /// Release an admitted but not yet active prefill authority.
    ///
    /// Returns true only when a retained authority was found and released.
    fn cancel_prefill_admission(&self, _request_id: &RequestId) -> bool {
        false
    }

    /// Whether this exact plan can retain and restore independent prefix state.
    /// This must include resolved model/provider support and product policy.
    fn supports_plan_runtime_prefix_restore(&self) -> bool {
        false
    }

    /// Restore a proper prefix into a freshly admitted request before the
    /// scheduler publishes any work for it. A miss submits no device work and
    /// preserves the admitted target. Successful restoration remains gated
    /// until the engine publishes the matching scheduler/executor progress and
    /// acknowledges the returned owner. Cancellation must retain unknown native
    /// writes until the existing completion path proves quiescence.
    async fn try_restore_plan_runtime_prefix(
        &self,
        _input: PlanRuntimePrefixRestoreInput<'_>,
    ) -> Result<PlanRuntimePrefixRestoreOutcome> {
        Ok(PlanRuntimePrefixRestoreOutcome::Unavailable)
    }

    /// Writes the exact availability sources advanced when this request
    /// authority is preempted for recompute.
    ///
    /// `true` proves that `preemption` still identifies a live, quiescently
    /// releasable authority and that `sources` is its complete release
    /// footprint. Callers must treat `false` as not releasable; a generic
    /// "owns some cache" observation is not evidence that another request can
    /// advance the source on which the current frontier is blocked.
    fn write_execution_capacity_release_sources(
        &self,
        _preemption: &ExecutorExecutionCapacityPreemption,
        sources: &mut Vec<crate::vnext::CapacityAvailabilitySource>,
    ) -> Result<bool> {
        sources.clear();
        Ok(false)
    }

    /// Retire one exact request-scoped runtime authority for recompute.
    ///
    /// Success is a terminal release fence: all provider/device work that can
    /// access the authority is quiescent and the request can be admitted as a
    /// new sequence incarnation. Implementations must fail closed on identity
    /// mismatch or an in-flight authority they cannot terminalize.
    async fn preempt_execution_capacity(
        &self,
        _preemption: ExecutorExecutionCapacityPreemption,
    ) -> Result<ExecutorExecutionCapacityPreemptionReceipt> {
        Err(FerrumError::unsupported(
            "request-scoped execution-capacity preemption is not implemented",
        ))
    }

    /// Consume one retained logical/physical backing deferral after the
    /// scheduler waiting lock has been released. Implementations must perform
    /// at most one bounded maintenance attempt and publish capacity epochs only
    /// after real backing is installed.
    fn maintain_prefill_backing(
        &self,
        _request_id: &RequestId,
    ) -> Result<ExecutorPrefillMaintenanceOutcome> {
        Err(ferrum_types::FerrumError::unsupported(
            "plan-runtime prefill backing maintenance is not implemented",
        ))
    }

    /// Reserve model-owned KV slots before a forward is dispatched.
    ///
    /// This is the executor-level admission hook for vLLM-style paged KV. The
    /// engine calls it at the batch boundary so a request that cannot grow its
    /// KV cache is delayed or preempted before kernel launch instead of
    /// panicking inside attention.
    fn reserve_kv_slots(&self, _requests: &[KvSlotRequest]) -> Result<Option<KvSlotReservation>> {
        Ok(None)
    }

    /// Snapshot model-owned paged-KV capacity without allocating slots.
    ///
    /// Executors without model-owned paged KV return `None`.
    fn kv_slot_capacity_snapshot(&self) -> Option<KvSlotCapacitySnapshot> {
        None
    }

    /// Recurrent-state allocation spec for this request, when the model has
    /// state-space or hybrid layers that need per-request recurrent state.
    ///
    /// Attention-only models return `None`. If this returns `Some`, the engine
    /// must allocate a recurrent-state handle before prefill and pass it through
    /// prefill/decode inputs. The default keeps existing executors KV-only.
    fn recurrent_state_spec(
        &self,
        _request_id: &RequestId,
        _input_tokens: &[TokenId],
    ) -> Result<Option<RecurrentStateSpec>> {
        Ok(None)
    }

    /// Execute prefill phase (process initial prompt)
    async fn prefill(&self, input: &PrefillInput) -> Result<PrefillOutput>;

    /// Execute one exact prefill chunk with an explicit pre-submit capacity
    /// deferral edge. Legacy executors inherit full-prefill behavior.
    async fn prefill_with_capacity(&self, input: &PrefillInput) -> Result<ExecutorPrefillOutcome> {
        let output = self.prefill(input).await?;
        let chunk = match input.chunk {
            Some(chunk) => chunk,
            None => PrefillChunk::new(0, input.sequence_length(), input.sequence_length())?,
        };
        Ok(ExecutorPrefillOutcome::Completed(
            ExecutorPrefillCompletion::exact(output, chunk),
        ))
    }

    /// Batch prefill: process multiple prompts' prefill in ONE forward pass.
    ///
    /// Default implementation falls back to per-request `prefill()` (serial,
    /// which is the current behavior the engine sees today). Executors that
    /// support unified mixed-batch forward (e.g. via `model.unified_forward`
    /// over a varlen QKV path) should override this to amortize launch /
    /// kernel-overhead across all `inputs` items in one call.
    ///
    /// Used by the continuous-batching engine to coalesce a cohort of new
    /// prefills (apples M3 c=32 sees 32 simultaneous prefills as one logical
    /// batch; the serial fallback runs each in ~47 ms while a true batched
    /// path runs all 32 in ~100 ms).
    async fn batch_prefill(&self, inputs: &[PrefillInput]) -> Result<Vec<PrefillOutput>> {
        let mut outputs = Vec::with_capacity(inputs.len());
        for input in inputs {
            outputs.push(self.prefill(input).await?);
        }
        Ok(outputs)
    }

    /// Attempt one physical, capacity-aware prefill batch.
    ///
    /// Implementations must either complete every input in original order or
    /// return `NotSubmitted` after restoring every retained prefill authority
    /// to a retryable state. Partial device submission is an ordinary error,
    /// never a `NotSubmitted` result.
    async fn batch_prefill_with_capacity(
        &self,
        _inputs: &[PrefillInput],
    ) -> Result<ExecutorBatchPrefillOutcome> {
        Ok(ExecutorBatchPrefillOutcome::Unsupported)
    }

    /// Tensor-free prefill for executors that declare
    /// [`ExecutionResourceAuthority::PlanRuntime`].
    ///
    /// The default fails closed because adapting through [`PrefillInput`]
    /// would silently restore a host tensor boundary.
    async fn plan_runtime_prefill_with_capacity(
        &self,
        _input: &PlanRuntimePrefillInput,
    ) -> Result<PlanRuntimePrefillOutcome> {
        Err(FerrumError::unsupported(
            "tensor-free plan-runtime prefill is not implemented",
        ))
    }

    /// Attempt one physical tensor-free prefill batch.
    ///
    /// `Unsupported` is an optimization fallback to the typed single-request
    /// method. `NotSubmitted` proves that no participant reached provider
    /// encode or device submission.
    async fn plan_runtime_batch_prefill_with_capacity(
        &self,
        _inputs: &[PlanRuntimePrefillInput],
    ) -> Result<PlanRuntimeBatchPrefillOutcome> {
        Ok(PlanRuntimeBatchPrefillOutcome::Unsupported)
    }

    /// Discard an exact prefill authority after engine-side validation,
    /// sampling, or scheduler commit fails.
    ///
    /// Plan runtimes should override this to remove both retained-prefill and
    /// active state by the exact opaque handle, not by a reusable request id.
    fn discard_plan_runtime_prefill(&self, authority: PlanRuntimePrefillAuthority) -> Result<()> {
        self.release_cache(&authority.kv_cache().cache_id());
        Ok(())
    }

    /// Execute decode phase (generate next token)
    async fn decode(&self, input: &DecodeInput) -> Result<DecodeOutput>;

    /// Batch decode: process multiple sequences in one forward pass.
    ///
    /// A successful result must contain exactly one output per input, in the
    /// original input order, and each output cache must retain the identity of
    /// its corresponding input cache. Implementations must not expose partial
    /// success as a shorter or reordered vector.
    ///
    /// The default implementation falls back to serial per-request `decode()`.
    /// Executors with a typed batch submission path should override this so one
    /// call maps to one resource step and one terminal submission fence.
    async fn batch_decode(&self, inputs: &[DecodeInput]) -> Result<Vec<DecodeOutput>> {
        let mut outputs = Vec::with_capacity(inputs.len());
        for input in inputs {
            outputs.push(self.decode(input).await?);
        }
        Ok(outputs)
    }

    /// Batch decode with an explicit pre-submit capacity deferral edge.
    ///
    /// Legacy executors inherit the successful/error-only behavior. A runtime
    /// with typed resource authority overrides this method so temporary
    /// capacity pressure is never flattened into a stringly resource error.
    async fn batch_decode_with_capacity(
        &self,
        inputs: &[DecodeInput],
    ) -> Result<ExecutorBatchDecodeOutcome> {
        self.batch_decode(inputs)
            .await
            .map(ExecutorBatchDecodeOutcome::Completed)
    }

    /// Tensor-free batch decode for executors that declare
    /// [`ExecutionResourceAuthority::PlanRuntime`].
    ///
    /// Implementations must preserve input ordering and cache identity exactly.
    /// Temporary pressure may return `Deferred` only before device submission.
    /// The default fails closed because adapting through [`DecodeInput`] would
    /// silently restore the host-tensor boundary this contract removes.
    async fn plan_runtime_batch_decode_with_capacity(
        &self,
        _inputs: &[PlanRuntimeDecodeInput],
    ) -> Result<PlanRuntimeBatchDecodeOutcome> {
        Err(FerrumError::unsupported(
            "tensor-free plan-runtime batch decode is not implemented",
        ))
    }

    /// Unified mixed-batch forward: process a [`UnifiedBatch`] containing
    /// any combination of prefill chunks (one or more `q_tokens` per item,
    /// possibly continuing from `pos_offset > 0`) and decode steps
    /// (`q_tokens.len() == 1`, `is_final_chunk = true`) in a single model
    /// forward pass.
    ///
    /// Returns one element per `batch.items[i]`:
    /// - `Some(logits)` for items with `is_final_chunk = true` (the
    ///   request's final-position logits, ready for sampling)
    /// - `None` for intermediate prefill chunks (no lm_head executed —
    ///   model only updates KV state)
    ///
    /// Default implementation returns `Err(unsupported)`. Concrete LLM
    /// executors should override with either:
    /// - A behavioral fallback that dispatches each chunk via existing
    ///   `prefill()` and groups decode items into `batch_decode()` (this
    ///   preserves current behavior; no perf change), OR
    /// - A real unified-forward path that runs all items through one
    ///   `[M_total, hidden]` GEMM chain with a varlen attention kernel
    ///   (this is the chunked-prefill perf unlock).
    async fn unified_decode(&self, _batch: &UnifiedBatch) -> Result<Vec<Option<Vec<f32>>>> {
        Err(ferrum_types::FerrumError::unsupported(
            "unified_decode not implemented for this executor",
        ))
    }

    /// Optional: full forward pass (for non-autoregressive use cases)
    async fn forward(&self, _input: &TensorRef) -> Result<TensorRef> {
        // Default implementation not supported
        Err(ferrum_types::FerrumError::unsupported(
            "Full forward pass not supported by this executor",
        ))
    }

    /// Roll the KV cache for this executor's sequence back to `new_len`.
    /// Used by speculative decoding on partial rejection so the next
    /// iteration sees a KV prefix that matches the accepted token stream.
    /// Default: Ok(()) — executors that don't cache per-sequence state
    /// (stub, mock) are inherently tolerant; real LLM executors override.
    async fn truncate_kv(
        &self,
        _kv_cache: &std::sync::Arc<dyn crate::KvCacheHandle>,
        _new_len: usize,
    ) -> Result<()> {
        Ok(())
    }

    /// Multi-position decode-verify: one forward over `N+1` tokens,
    /// producing one logits row per position. Used by speculative
    /// decoding's target path so we don't pay N+1 sequential forwards.
    ///
    /// Default falls back to N+1 sequential `decode()` calls — correct
    /// but slow; real LLM executors override.
    ///
    /// Returns a `Vec<DecodeOutput>` of length `inputs.len()` with the
    /// final KV handle attached to the last element.
    async fn forward_verify(&self, inputs: &[DecodeInput]) -> Result<Vec<DecodeOutput>> {
        let mut out = Vec::with_capacity(inputs.len());
        for input in inputs {
            out.push(self.decode(input).await?);
        }
        Ok(out)
    }

    /// Get executor capabilities
    fn capabilities(&self) -> ExecutorCapabilities;

    /// Get current executor status
    fn status(&self) -> ExecutorStatus;

    /// Optional model/executor cache metrics.
    ///
    /// Concrete LLM executors use this for model-level paged KV prefix reuse
    /// counters. Default implementations keep non-autoregressive executors
    /// and tests from needing cache-specific plumbing.
    fn cache_metrics_snapshot(&self) -> Option<serde_json::Value> {
        None
    }

    /// Optional compact provider-attribution witness emitted by executors
    /// whose immutable plan can bind quantized source tensors to selected
    /// operation providers without exposing per-tensor logs.
    fn execution_attribution_snapshot(&self) -> Option<serde_json::Value> {
        None
    }

    /// Optional LoRA runtime metrics.
    fn lora_metrics_snapshot(&self) -> Option<serde_json::Value> {
        None
    }

    /// Complete executor-owned startup preparation before either product
    /// entrypoint can accept a request.
    ///
    /// Implementations use this cold-path hook for work that requires the
    /// fully constructed executor but must not be charged to a user's first
    /// request, such as compiling reusable execution shapes. The default is a
    /// no-op so existing executors remain source compatible.
    async fn prepare_startup(&self) -> Result<()> {
        Ok(())
    }

    /// Warm up executor (load model, allocate memory, etc.)
    async fn warmup(&mut self) -> Result<()> {
        // Default no-op implementation
        Ok(())
    }

    /// Shutdown executor gracefully
    async fn shutdown(&mut self) -> Result<()> {
        // Default no-op implementation
        Ok(())
    }

    /// Complete and release one cache authority with product-authoritative
    /// terminal token counts.
    ///
    /// Legacy executors only need physical release and inherit that behavior.
    /// Plan runtimes with terminal journals override this method so completion
    /// cannot be inferred from a generic release operation. Implementations may
    /// await retention of the last completed native state before releasing its
    /// source; sampled output alone is not evidence of an executed frontier.
    async fn complete_cache(&self, completion: ExecutorSequenceCompletion) -> Result<()> {
        self.release_cache(completion.cache_id());
        Ok(())
    }

    /// Release KV cache and state without asserting successful completion.
    ///
    /// Called for cancellation, failure, recompute, and legacy cleanup. The
    /// `cache_id` matches the value embedded in the `KvCacheHandle` returned by
    /// prefill/decode. Successful product completion uses [`Self::complete_cache`].
    fn release_cache(&self, _cache_id: &str) {
        // Default no-op — executors that manage per-sequence KV caches should override.
    }
}

/// Executor capabilities and configuration
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ExecutorCapabilities {
    /// Maximum supported batch size
    pub max_batch_size: usize,
    /// Maximum sequence length
    pub max_sequence_length: usize,
    /// Supported attention mechanisms
    pub attention_mechanisms: Vec<AttentionType>,
    /// Whether executor supports dynamic batching
    pub supports_dynamic_batching: bool,
    /// Whether executor supports continuous batching
    pub supports_continuous_batching: bool,
    /// Whether executor supports speculative decoding
    pub supports_speculative_decoding: bool,
    /// Whether executor supports tensor parallelism
    pub supports_tensor_parallelism: bool,
    /// Whether executor supports pipeline parallelism
    pub supports_pipeline_parallelism: bool,
    /// Supported data types
    pub supported_dtypes: Vec<ferrum_types::DataType>,
    /// Supported devices
    pub supported_devices: Vec<ferrum_types::Device>,
    /// Memory requirements estimation
    pub memory_requirements: MemoryRequirements,
}

/// Attention mechanism types
#[derive(Debug, Clone, Copy, Serialize, Deserialize)]
pub enum AttentionType {
    /// Standard multi-head attention
    MultiHead,
    /// Multi-query attention (MQA)
    MultiQuery,
    /// Grouped-query attention (GQA)
    GroupedQuery,
    /// Flash attention
    Flash,
    /// Paged attention
    Paged,
    /// Sliding window attention
    SlidingWindow,
}

/// Logical state bytes across the complete model, before physical page
/// alignment, pool residency, checkpoint copies or operation workspace.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)]
pub struct TypedSequenceStateMemory {
    /// K/V payload and quantization scale bytes for one token in one sequence.
    pub kv_bytes_per_token: u64,
    /// Other sequence states whose size scales with token count.
    pub other_token_scaled_bytes_per_token: u64,
    /// Fixed sequence state, including recurrent accumulators and windows.
    pub fixed_bytes_per_sequence: u64,
}

/// Memory requirements for model execution. This describes logical model
/// requirements, not measured device allocation or an admission guarantee.
#[derive(Debug, Clone, Deserialize)]
pub struct MemoryRequirements {
    /// Model parameter memory in bytes
    pub parameter_memory: u64,
    /// Minimum activation memory per token
    pub activation_memory_per_token: usize,
    /// Legacy KV cache bytes per token per layer. Ignored and omitted from
    /// serialization when typed_sequence_state supplies complete-model bytes.
    #[serde(default)]
    pub kv_cache_memory_per_token: usize,
    /// Exact logical sequence state from a typed model plan. Old wire values
    /// without this field retain the legacy per-layer interpretation.
    #[serde(default)]
    pub typed_sequence_state: Option<TypedSequenceStateMemory>,
    /// Additional overhead memory
    pub overhead_memory: u64,
}

impl Serialize for MemoryRequirements {
    fn serialize<S: serde::Serializer>(
        &self,
        serializer: S,
    ) -> std::result::Result<S::Ok, S::Error> {
        use serde::ser::SerializeStruct;
        let mut value = serializer.serialize_struct("MemoryRequirements", 4)?;
        value.serialize_field("parameter_memory", &self.parameter_memory)?;
        value.serialize_field(
            "activation_memory_per_token",
            &self.activation_memory_per_token,
        )?;
        match &self.typed_sequence_state {
            Some(state) => value.serialize_field("typed_sequence_state", state)?,
            None => value
                .serialize_field("kv_cache_memory_per_token", &self.kv_cache_memory_per_token)?,
        }
        value.serialize_field("overhead_memory", &self.overhead_memory)?;
        value.end()
    }
}

impl MemoryRequirements {
    /// Calculate a logical memory estimate, returning None on arithmetic
    /// overflow. Typed sequence state is already summed across all layers.
    pub fn checked_calculate_total_memory(
        &self,
        batch_size: usize,
        sequence_length: usize,
        num_layers: usize,
    ) -> Option<u64> {
        let batch = u64::try_from(batch_size).ok()?;
        let tokens = u64::try_from(sequence_length).ok()?;
        let token_count = batch.checked_mul(tokens)?;
        let activation_mem = u64::try_from(self.activation_memory_per_token)
            .ok()?
            .checked_mul(token_count)?;
        let state_mem = match self.typed_sequence_state {
            Some(state) => state
                .kv_bytes_per_token
                .checked_add(state.other_token_scaled_bytes_per_token)?
                .checked_mul(token_count)?
                .checked_add(state.fixed_bytes_per_sequence.checked_mul(batch)?)?,
            None => u64::try_from(self.kv_cache_memory_per_token)
                .ok()?
                .checked_mul(token_count)?
                .checked_mul(u64::try_from(num_layers).ok()?)?,
        };
        self.parameter_memory
            .checked_add(activation_mem)?
            .checked_add(state_mem)?
            .checked_add(self.overhead_memory)
    }

    /// Calculate a logical memory estimate. Overflow saturates to u64::MAX;
    /// callers needing an explicit overflow result can use the checked method.
    pub fn calculate_total_memory(
        &self,
        batch_size: usize,
        sequence_length: usize,
        num_layers: usize,
    ) -> u64 {
        self.checked_calculate_total_memory(batch_size, sequence_length, num_layers)
            .unwrap_or(u64::MAX)
    }
}

#[cfg(test)]
mod memory_requirements_tests {
    use super::{MemoryRequirements, TypedSequenceStateMemory};

    #[test]
    fn typed_sequence_memory_counts_scales_and_fixed_state_once_per_sequence() {
        let memory = MemoryRequirements {
            parameter_memory: 100,
            activation_memory_per_token: 4,
            kv_cache_memory_per_token: 999,
            typed_sequence_state: Some(TypedSequenceStateMemory {
                kv_bytes_per_token: 528,
                other_token_scaled_bytes_per_token: 8,
                fixed_bytes_per_sequence: 64,
            }),
            overhead_memory: 20,
        };
        let expected = 100 + 4 * 2 * 3 + (528 + 8) * 2 * 3 + 64 * 2 + 20;
        for layers in [1, 3, 32] {
            assert_eq!(
                memory.checked_calculate_total_memory(2, 3, layers),
                Some(expected)
            );
        }
        let wire = serde_json::to_value(&memory).unwrap();
        assert!(wire.get("kv_cache_memory_per_token").is_none());
        assert_eq!(wire["typed_sequence_state"]["kv_bytes_per_token"], 528);
        let decoded: MemoryRequirements = serde_json::from_value(wire).unwrap();
        assert_eq!(decoded.calculate_total_memory(2, 3, 32), expected);
    }

    #[test]
    fn legacy_memory_wire_keeps_per_layer_calculation_and_rejects_overflow() {
        let wire = serde_json::json!({
            "parameter_memory": 100,
            "activation_memory_per_token": 4,
            "kv_cache_memory_per_token": 16,
            "overhead_memory": 20,
        });
        let mut memory: MemoryRequirements = serde_json::from_value(wire.clone()).unwrap();
        assert!(memory.typed_sequence_state.is_none());
        assert_eq!(
            memory.calculate_total_memory(2, 3, 5),
            100 + 4 * 2 * 3 + 16 * 2 * 3 * 5 + 20
        );
        assert_eq!(serde_json::to_value(&memory).unwrap(), wire);
        memory.parameter_memory = u64::MAX;
        assert_eq!(memory.checked_calculate_total_memory(1, 1, 1), None);
        assert_eq!(memory.calculate_total_memory(1, 1, 1), u64::MAX);
        memory.parameter_memory = 0;
        memory.typed_sequence_state = Some(TypedSequenceStateMemory {
            kv_bytes_per_token: u64::MAX,
            other_token_scaled_bytes_per_token: 1,
            fixed_bytes_per_sequence: 0,
        });
        assert_eq!(memory.checked_calculate_total_memory(1, 1, 1), None);
    }
}

/// Executor status information
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ExecutorStatus {
    /// Current executor state
    pub state: ExecutorState,
    /// Whether executor is ready to accept requests
    pub is_ready: bool,
    /// Current batch size being processed
    pub current_batch_size: usize,
    /// Number of prefill operations completed
    pub prefill_operations: u64,
    /// Number of decode operations completed
    pub decode_operations: u64,
    /// Average prefill time in milliseconds
    pub avg_prefill_time_ms: f64,
    /// Average decode time in milliseconds
    pub avg_decode_time_ms: f64,
    /// Memory usage statistics
    pub memory_usage: ExecutorMemoryUsage,
    /// Last operation timestamp
    #[serde(skip)]
    pub last_operation: Option<std::time::Instant>,
}

/// Executor state
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub enum ExecutorState {
    /// Executor is initializing
    Initializing,
    /// Executor is ready to accept requests
    Ready,
    /// Executor is processing requests
    Busy,
    /// Executor encountered an error
    Error,
    /// Executor is shutting down
    Shutdown,
}

/// Executor memory usage
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ExecutorMemoryUsage {
    /// Total allocated memory in bytes
    pub allocated_bytes: usize,
    /// Currently used memory in bytes
    pub used_bytes: usize,
    /// Peak memory usage
    pub peak_bytes: usize,
    /// Memory utilization percentage
    pub utilization_percent: f32,
}

/// Batch model executor for processing multiple requests efficiently
#[async_trait]
pub trait BatchModelExecutor: ModelExecutor {
    /// Execute batch prefill for multiple sequences
    async fn batch_prefill(&self, inputs: &[PrefillInput]) -> Result<Vec<PrefillOutput>>;

    /// Execute batch decode for multiple sequences
    async fn batch_decode(&self, inputs: &[DecodeInput]) -> Result<Vec<DecodeOutput>>;

    /// Get optimal batch size for current conditions
    fn optimal_batch_size(&self) -> usize;

    /// Check if batch size is supported
    fn supports_batch_size(&self, batch_size: usize) -> bool;
}

/// Speculative execution support
#[async_trait]
pub trait SpeculativeExecutor: ModelExecutor {
    /// Execute speculative decoding with draft model
    async fn speculative_decode(
        &self,
        input: &DecodeInput,
        draft_tokens: &[ferrum_types::TokenId],
        acceptance_threshold: f32,
    ) -> Result<SpeculativeDecodeOutput>;
}

/// Output from speculative decoding
#[derive(Debug, Clone)]
pub struct SpeculativeDecodeOutput {
    /// Accepted tokens (subset of draft tokens)
    pub accepted_tokens: Vec<ferrum_types::TokenId>,
    /// Logits for the next token after last accepted
    pub next_logits: TensorRef,
    /// Updated KV cache
    pub kv_cache: Arc<dyn KvCacheHandle>,
    /// Number of draft tokens accepted
    pub acceptance_count: usize,
}

/// Model executor factory
#[async_trait]
pub trait ModelExecutorFactory: Send + Sync {
    /// Create executor from model configuration
    async fn create_executor(&self, config: &ExecutorConfig) -> Result<Box<dyn ModelExecutor>>;

    /// Create batch executor
    async fn create_batch_executor(
        &self,
        config: &ExecutorConfig,
    ) -> Result<Box<dyn BatchModelExecutor>>;

    /// Get supported executor types
    fn supported_types(&self) -> Vec<ExecutorType>;

    /// Validate configuration
    fn validate_config(&self, config: &ExecutorConfig) -> Result<()>;
}

/// Executor configuration
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ExecutorConfig {
    /// Model information
    pub model_info: ModelInfo,
    /// Target device
    pub device: ferrum_types::Device,
    /// Data type for computation
    pub dtype: ferrum_types::DataType,
    /// Maximum batch size
    pub max_batch_size: usize,
    /// Maximum sequence length
    pub max_sequence_length: usize,
    /// Attention configuration
    pub attention_config: ExecutorAttentionConfig,
    /// Memory configuration
    pub memory_config: ExecutorMemoryConfig,
    /// Optimization settings
    pub optimization_config: OptimizationConfig,
    /// Additional executor-specific options
    pub executor_options: HashMap<String, serde_json::Value>,
}

/// Runtime attention configuration for model executor
///
/// Note: This is different from ferrum_types::AttentionConfig which describes
/// the model architecture's attention configuration from config.json.
/// This type describes the runtime execution settings.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ExecutorAttentionConfig {
    /// Type of attention to use
    pub attention_type: AttentionType,
    /// Enable flash attention if available
    pub enable_flash_attention: bool,
    /// Enable paged attention
    pub enable_paged_attention: bool,
    /// Block size for paged attention
    pub block_size: Option<usize>,
    /// Sliding window size (if using sliding window attention)
    pub sliding_window_size: Option<usize>,
}

/// Memory configuration for executor
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ExecutorMemoryConfig {
    /// Enable memory pooling
    pub enable_memory_pooling: bool,
    /// Memory pool size in bytes (None for auto)
    pub memory_pool_size: Option<usize>,
    /// Enable KV cache sharing
    pub enable_kv_cache_sharing: bool,
    /// Maximum memory usage percentage
    pub max_memory_usage: f32,
}

/// Optimization configuration
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct OptimizationConfig {
    /// Enable CUDA graphs (if supported)
    pub enable_cuda_graphs: bool,
    /// Enable kernel fusion
    pub enable_kernel_fusion: bool,
    /// Enable mixed precision
    pub enable_mixed_precision: bool,
    /// Optimization level (0-3)
    pub optimization_level: u8,
    /// Custom optimization flags
    pub custom_flags: HashMap<String, bool>,
}

/// Supported executor types
#[derive(Debug, Clone, Copy, Serialize, Deserialize)]
pub enum ExecutorType {
    /// Standard sequential executor
    Sequential,
    /// Batch executor for parallel processing
    Batch,
    /// Continuous batching executor
    ContinuousBatch,
    /// Speculative decoding executor
    Speculative,
    /// Pipeline parallel executor
    PipelineParallel,
    /// Tensor parallel executor
    TensorParallel,
}

/// Executor performance metrics
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ExecutorMetrics {
    /// Total operations executed
    pub total_operations: u64,
    /// Prefill operations
    pub prefill_operations: u64,
    /// Decode operations
    pub decode_operations: u64,
    /// Average prefill latency (ms)
    pub avg_prefill_latency: f64,
    /// Average decode latency (ms)
    pub avg_decode_latency: f64,
    /// P95 prefill latency (ms)
    pub p95_prefill_latency: f64,
    /// P95 decode latency (ms)
    pub p95_decode_latency: f64,
    /// Throughput (tokens per second)
    pub throughput_tps: f64,
    /// Memory efficiency (used/allocated)
    pub memory_efficiency: f32,
    /// Batch utilization
    pub batch_utilization: f32,
}

/// Executor registry for managing multiple executors
pub trait ExecutorRegistry: Send + Sync {
    /// Register executor with name
    fn register(&mut self, name: &str, executor: Box<dyn ModelExecutor>) -> Result<()>;

    /// Get executor by name
    fn get(&self, name: &str) -> Option<&dyn ModelExecutor>;

    /// Remove executor by name
    fn remove(&mut self, name: &str) -> Option<Box<dyn ModelExecutor>>;

    /// List registered executor names
    fn list_names(&self) -> Vec<String>;

    /// Get executor metrics
    fn get_metrics(&self, name: &str) -> Option<ExecutorMetrics>;
}