ferrum_interfaces/vnext/resource/

invocation.rs

use super::{
    begin_participant_flights_dispatch, begin_submission_wave_participant_flights_dispatch,
    finalize_session_frames_with_boundary, fmt, invalid_resource, issue_batch_invocation_id,
    poison_session_frame, prepare_participant_flights, prepare_submission_wave_participant_flights,
    reset_participant_flights_after_definitely_not_submitted,
    reset_submission_wave_participant_flights_after_definitely_not_submitted,
    session_participant_key, ActiveInvocationWaveGuard, ActiveSequenceFrame, AdmissionDeferred,
    AdmissionFitPolicy, AdmissionPressureAction, AdmissionRejected, AdmittedRequestResources,
    AdmittedSequenceResources, AdmittedStepParticipant, AllocationLifetime, Arc, AtomicU64,
    BackingInitializationEncodeError, BackingPrepareDecision, BatchCapacityClaimDecision,
    BatchInvocationId, BatchParticipantAuthority, BatchParticipantTokenSpan, BatchStepId,
    BatchWorkShape, ClaimedBackingTransaction, ClaimedSubmissionWaveBacking,
    DeferredDeviceCleanupDomainId, DeviceCommandBatch, DeviceRuntime, Digest,
    DynamicBackingDeferred, DynamicDeferredMaintenanceOutcome, DynamicResourceDescriptor,
    ExecutionLane, ExecutionLaneId, InvocationRegistry, InvocationResourceAdmissionRequest,
    LaneBackingPrepareDecision, LogicalAdmissionCoordinatorId, LogicalBackingBufferView,
    LogicalBackingSliceAuthority, LogicalBatchCapacityLease, NodeId, Ordering,
    ParticipantFlightCandidate, ParticipantFlightPhase, ParticipantNodeKey, PlanBackingDeferral,
    PlanCapacityWaitRegistration, PreparedBackingInitializations, PreparedParticipantFlightHold,
    PreparedSubmissionWaveParticipantFlightHold, RequestStateHazardAcquireDecision,
    RequestStateHazardDeferral, RequestStateHazardParticipant, RequestStateHazardPermit,
    RequestStateHazardPoison, RequestStateHazardSplitRequired,
    RequestStateHazardTerminalDisposition, ResourceId, SequenceAuthorityId,
    SequenceBackingSnapshot, SequenceRecoveryRegistry, SequenceSessionEpoch,
    SequenceSessionFingerprint, Serialize, Sha256, StaticProvisioningLease,
    StepFinalizationFailure, StepFrameFinalization, StepParticipantFrameAssignment,
    StepParticipantRetirement, StepParticipantRetirementDisposition, StepResourceLease,
    StepRetirementReceipt, TokenSpanWork, TrustedPlanRuntimeEvidence, VNextError,
    SEQUENCE_DISPATCH_POISONED_BIT,
};
use crate::vnext::ReusableExecutionBucketSpec;
use std::time::{Duration, Instant};

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum StepResourceAdmissionProfilePhase {
    AuthorityAndPolicyValidate,
    DemandEvaluate,
    BackingClaim,
    LogicalCapacityClaim,
    TransactionValidateAndFingerprint,
    FrameCaptureAndLease,
}

#[inline(always)]
pub(super) fn step_admission_profile_start<const PROFILE: bool>() -> Option<Instant> {
    PROFILE.then(Instant::now)
}

#[inline(always)]
pub(super) fn record_step_admission_profile<const PROFILE: bool, F>(
    observer: &mut F,
    phase: StepResourceAdmissionProfilePhase,
    started: Option<Instant>,
) where
    F: FnMut(StepResourceAdmissionProfilePhase, Duration),
{
    if PROFILE {
        observer(
            phase,
            started
                .expect("profiled step admission phase owns a start instant")
                .elapsed(),
        );
    }
}

#[derive(Debug)]
pub enum ExecutionStreamCreationError<E> {
    Contract(VNextError),
    Runtime(E),
}

/// A stream created and owned by one exact admitted runtime instance. Its
/// runtime and raw stream are private so execution can only proceed through an
/// active sequence permit.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(super) enum BoundExecutionStreamState {
    Ready,
    InUse,
    Poisoned,
}

#[must_use = "an execution stream must be activated through its resource lease"]
pub struct BoundExecutionStream<R>
where
    R: DeviceRuntime,
{
    // The raw stream drops or transfers into the recovery registry before this
    // owning sequence hold can release logical capacity or backing extents.
    pub(super) runtime: Arc<R>,
    pub(super) coordinator_id: LogicalAdmissionCoordinatorId,
    pub(super) sequence_authority: SequenceAuthorityId,
    pub(super) stream: Option<R::Stream>,
    pub(super) state: BoundExecutionStreamState,
    pub(super) sequence_recovery: Arc<SequenceRecoveryRegistry<R>>,
    pub(super) sequence_dispatch_gate: Arc<AtomicU64>,
    pub(super) abandoned_sequence: Option<(u32, u64)>,
    pub(super) resources: Arc<AdmittedSequenceResources<R>>,
}

impl<R> BoundExecutionStream<R>
where
    R: DeviceRuntime,
{
    pub(super) fn stream(&self) -> &R::Stream {
        self.stream
            .as_ref()
            .expect("bound execution stream retains its raw stream")
    }

    pub(super) fn stream_mut(&mut self) -> &mut R::Stream {
        self.stream
            .as_mut()
            .expect("bound execution stream retains its raw stream")
    }
}

impl<R> Drop for BoundExecutionStream<R>
where
    R: DeviceRuntime,
{
    fn drop(&mut self) {
        let Some(key) = self.abandoned_sequence.take() else {
            return;
        };
        self.sequence_dispatch_gate
            .fetch_or(SEQUENCE_DISPATCH_POISONED_BIT, Ordering::AcqRel);
        if let Some(stream) = self.stream.take() {
            self.sequence_recovery.attach_stream(key, stream);
        }
    }
}

impl<R> StepResourceLease<R>
where
    R: DeviceRuntime,
{
    pub(super) fn new(
        participants: Vec<AdmittedStepParticipant<R>>,
        execution_lane: Arc<ExecutionLane<R>>,
        reusable_execution_bucket: Option<ReusableExecutionBucketSpec>,
        batch_step_id: BatchStepId,
        claimed_backing: ClaimedBackingTransaction,
    ) -> Result<Self, VNextError> {
        if participants.is_empty() {
            return Err(invalid_resource(
                "step resources require a non-empty participant set",
            ));
        }
        let coordinator = participants[0]
            .session
            .resources()
            .request
            .plan
            .logical_admission();
        if claimed_backing.work_shape().participants().len() != participants.len()
            || claimed_backing
                .work_shape()
                .participants()
                .iter()
                .zip(&participants)
                .any(|(authority, participant)| {
                    authority.sequence_authority() != participant.session.sequence_authority()
                        || authority.request_authority() != participant.session.request_authority()
                })
        {
            return Err(invalid_resource(
                "step work shape differs from its exact batch participants",
            ));
        }
        if let Some(capacity) = claimed_backing.logical_capacity() {
            let parents_match = capacity
                .parents()
                .iter()
                .map(|parent| (parent.sequence(), parent.request()))
                .eq(participants.iter().map(|participant| {
                    (
                        participant.session.sequence_authority(),
                        participant.session.request_authority(),
                    )
                }));
            if !coordinator.owns_batch_capacity_claim(capacity) || !parents_match {
                return Err(invalid_resource(
                    "step capacity authority differs from its exact batch participants",
                ));
            }
        }
        Ok(Self {
            claimed_backing,
            participants,
            invocation_registry: Arc::new(InvocationRegistry::default()),
            execution_lane,
            reusable_execution_bucket,
            batch_step_id,
            completed_boundary: super::Mutex::new(super::StepCompletedBoundarySlot::default()),
            finalized: false,
        })
    }

    pub const fn batch_step_id(&self) -> BatchStepId {
        self.batch_step_id
    }

    pub fn execution_lane(&self) -> &Arc<ExecutionLane<R>> {
        &self.execution_lane
    }

    pub fn reusable_execution_bucket(&self) -> Option<&ReusableExecutionBucketSpec> {
        self.reusable_execution_bucket.as_ref()
    }

    pub fn try_retire_normal(
        self: Arc<Self>,
    ) -> Result<StepRetirementReceipt, StepFinalizationFailure<R>> {
        Self::try_finalize(self, StepFrameFinalization::Commit)
    }

    pub fn try_abort(self: Arc<Self>) -> Result<StepRetirementReceipt, StepFinalizationFailure<R>> {
        Self::try_finalize(self, StepFrameFinalization::Abort)
    }

    /// Releases a capacity-deferred step before any invocation was prepared or
    /// submitted. Unlike `try_abort`, this leaves every participant session open
    /// so the scheduler can retry after the exact capacity source advances.
    pub fn try_rollback_unsubmitted(
        self: Arc<Self>,
    ) -> Result<StepRetirementReceipt, StepFinalizationFailure<R>> {
        Self::try_finalize(self, StepFrameFinalization::RollbackUnsubmitted)
    }

    fn try_finalize(
        step: Arc<Self>,
        finalization: StepFrameFinalization,
    ) -> Result<StepRetirementReceipt, StepFinalizationFailure<R>> {
        let mut step = match Arc::try_unwrap(step) {
            Ok(step) => step,
            Err(step) => {
                return Err(StepFinalizationFailure {
                    step,
                    error: invalid_resource(
                        "step cannot finalize while an invocation or scheduler clone retains it",
                    ),
                });
            }
        };
        if finalization == StepFrameFinalization::RollbackUnsubmitted {
            if let Err(error) = step.invocation_registry.ensure_pristine_for_step_rollback() {
                return Err(StepFinalizationFailure {
                    step: Arc::new(step),
                    error,
                });
            }
        }
        let dispositions = match step.finalize_participants(finalization) {
            Ok(dispositions) => dispositions,
            Err(error) => {
                return Err(StepFinalizationFailure {
                    step: Arc::new(step),
                    error,
                });
            }
        };
        let participants = step
            .participants
            .iter()
            .zip(dispositions)
            .map(|(participant, disposition)| StepParticipantRetirement {
                assignment: StepParticipantFrameAssignment::new(
                    participant.session.sequence_authority(),
                    participant.session.request_authority(),
                    participant.frame.frame_id,
                ),
                disposition,
            })
            .collect();
        Ok(StepRetirementReceipt {
            batch_step_id: step.batch_step_id,
            participants,
        })
    }

    fn finalize_participants(
        &mut self,
        finalization: StepFrameFinalization,
    ) -> Result<Vec<StepParticipantRetirementDisposition>, VNextError> {
        if self.finalized {
            return Err(invalid_resource("step resources are already finalized"));
        }
        let mut holds = self
            .participants
            .iter_mut()
            .map(|participant| &mut participant.frame)
            .collect::<Vec<_>>();
        let completed = self
            .completed_boundary
            .get_mut()
            .map_err(|_| invalid_resource("step completed-boundary mutex is poisoned"))?;
        let dispositions =
            finalize_session_frames_with_boundary(&mut holds, finalization, completed.proof())?;
        completed.consume();
        self.finalized = true;
        Ok(dispositions)
    }

    pub fn participant_count(&self) -> u32 {
        u32::try_from(self.participants.len())
            .expect("step participant count was validated before admission")
    }

    pub fn coordinator_id(&self) -> LogicalAdmissionCoordinatorId {
        self.participants[0].session.resources().coordinator_id()
    }

    pub fn participants(
        &self,
    ) -> impl ExactSizeIterator<Item = &Arc<AdmittedSequenceResources<R>>> {
        self.participants
            .iter()
            .map(|participant| participant.session.resources())
    }

    pub(crate) fn participant_backing_snapshot(
        &self,
        participant: BatchParticipantAuthority,
    ) -> Result<&Arc<SequenceBackingSnapshot<R>>, VNextError> {
        let index = self
            .participants
            .binary_search_by_key(&participant.canonical_key(), |candidate| {
                session_participant_key(&candidate.session)
            })
            .map_err(|_| invalid_resource("step does not own that backing participant"))?;
        self.participants
            .get(index)
            .map(|participant| &participant.backing_snapshot)
            .ok_or_else(|| invalid_resource("step backing participant mapping is inconsistent"))
    }

    pub(crate) fn participant_backing_view(
        &self,
        authority: BatchParticipantAuthority,
        resource_id: &ResourceId,
    ) -> Result<LogicalBackingBufferView<'_, R::Buffer>, VNextError> {
        let index = self
            .participants
            .binary_search_by_key(&authority.canonical_key(), |candidate| {
                session_participant_key(&candidate.session)
            })
            .map_err(|_| invalid_resource("step does not own that backing participant"))?;
        let participant = self
            .participants
            .get(index)
            .ok_or_else(|| invalid_resource("step backing participant mapping is inconsistent"))?;
        let authorities = participant.backing_snapshot.backing_slices_for(resource_id);
        if !authorities.is_empty() {
            return participant
                .session
                .resources()
                .request
                .plan
                .dynamic_pools()
                .view_many(authorities);
        }
        participant
            .session
            .resources()
            .request
            .backing_view(resource_id)
    }

    pub fn participant_frames(
        &self,
    ) -> impl ExactSizeIterator<Item = StepParticipantFrameAssignment> + '_ {
        self.participants.iter().map(|participant| {
            StepParticipantFrameAssignment::new(
                participant.session.sequence_authority(),
                participant.session.request_authority(),
                participant.frame.frame_id,
            )
        })
    }

    pub fn backing_slices(&self) -> &[LogicalBackingSliceAuthority] {
        self.claimed_backing.backing_slices()
    }

    pub fn logical_capacity(&self) -> Option<&LogicalBatchCapacityLease> {
        self.claimed_backing.logical_capacity()
    }

    pub fn work_shape(&self) -> &BatchWorkShape {
        self.claimed_backing.work_shape()
    }

    pub fn bind_invocation_work_shape(
        &self,
        mut participant_tokens: Vec<(BatchParticipantAuthority, TokenSpanWork)>,
    ) -> Result<BatchWorkShape, VNextError> {
        participant_tokens.sort_by_key(|(participant, _)| participant.canonical_key());
        if participant_tokens.is_empty()
            || participant_tokens
                .windows(2)
                .any(|pair| pair[0].0.canonical_key() == pair[1].0.canonical_key())
            || participant_tokens.iter().any(|(authority, _)| {
                self.participants
                    .binary_search_by_key(&authority.canonical_key(), |participant| {
                        session_participant_key(&participant.session)
                    })
                    .is_err()
            })
        {
            return Err(invalid_resource(
                "invocation token work must bind a unique non-empty step participant subset",
            ));
        }
        BatchWorkShape::new(
            participant_tokens
                .into_iter()
                .map(|(participant, token_span)| {
                    BatchParticipantTokenSpan::new(participant, token_span)
                })
                .collect(),
        )
    }

    pub fn bind_all_invocation_work_shape(
        &self,
        token_spans: Vec<TokenSpanWork>,
    ) -> Result<BatchWorkShape, VNextError> {
        if token_spans.len() != self.participants.len() {
            return Err(invalid_resource(
                "invocation token work count differs from all step participants",
            ));
        }
        self.bind_invocation_work_shape(
            self.participants
                .iter()
                .zip(token_spans)
                .map(|(participant, token_span)| {
                    (
                        BatchParticipantAuthority::new(
                            participant.session.sequence_authority(),
                            participant.session.request_authority(),
                        ),
                        token_span,
                    )
                })
                .collect(),
        )
    }

    /// Reuses the exact immutable work authority admitted for this step when a
    /// submission wave binds the same participants and token spans.
    pub fn shared_all_invocation_work_shape(
        &self,
        token_spans: &[TokenSpanWork],
    ) -> Result<Arc<BatchWorkShape>, VNextError> {
        let work_shape = self.claimed_backing.work_shape();
        if token_spans.len() != self.participants.len()
            || work_shape
                .participant_work()
                .iter()
                .zip(token_spans)
                .any(|(bound, requested)| bound.token_span() != requested)
        {
            return Err(invalid_resource(
                "submission wave token work differs from its admitted step work authority",
            ));
        }
        Ok(Arc::clone(self.claimed_backing.work_shape_arc()))
    }

    pub fn claimed_backing(&self) -> &ClaimedBackingTransaction {
        &self.claimed_backing
    }

    pub fn static_provisioning(&self) -> Option<&StaticProvisioningLease<R>> {
        self.participants[0]
            .session
            .resources()
            .static_provisioning()
    }

    pub fn plan_evidence(&self) -> TrustedPlanRuntimeEvidence {
        self.participants[0].session.resources().plan_evidence()
    }

    pub(crate) fn backing_view(
        &self,
        resource_id: &ResourceId,
    ) -> Result<LogicalBackingBufferView<'_, R::Buffer>, VNextError> {
        if let Some(authority) = self
            .claimed_backing
            .backing_slices()
            .iter()
            .find(|authority| authority.resource_id() == resource_id)
        {
            return self.participants[0]
                .session
                .resources()
                .request
                .plan
                .dynamic_pools()
                .view(authority);
        }
        Err(invalid_resource(format!(
            "resource `{resource_id}` is not step-shared backing"
        )))
    }

    pub(crate) fn dynamic_descriptor(
        &self,
        resource_id: &ResourceId,
    ) -> Result<&DynamicResourceDescriptor, VNextError> {
        let pools = self.participants[0]
            .session
            .resources()
            .request
            .plan
            .dynamic_pools();
        let mut matches = pools.domains.iter().filter_map(|domain| {
            domain
                .descriptors
                .binary_search_by(|descriptor| descriptor.base_resource_id().cmp(resource_id))
                .ok()
                .map(|index| &domain.descriptors[index])
        });
        let descriptor = matches.next().ok_or_else(|| {
            invalid_resource(format!(
                "resource `{resource_id}` has no dynamic plan descriptor"
            ))
        })?;
        if matches.next().is_some() {
            return Err(invalid_resource(format!(
                "resource `{resource_id}` has duplicate dynamic plan descriptors"
            )));
        }
        Ok(descriptor)
    }

    pub(crate) fn participant_backing_views(
        &self,
        resource_id: &ResourceId,
    ) -> Result<Vec<(SequenceAuthorityId, LogicalBackingBufferView<'_, R::Buffer>)>, VNextError>
    {
        self.participants
            .iter()
            .map(|participant| {
                let authority = BatchParticipantAuthority::new(
                    participant.session.sequence_authority(),
                    participant.session.request_authority(),
                );
                Ok((
                    participant.session.sequence_authority(),
                    self.participant_backing_view(authority, resource_id)?,
                ))
            })
            .collect()
    }

    pub fn try_admit_invocation(
        self: &Arc<Self>,
        request: InvocationResourceAdmissionRequest,
    ) -> Result<InvocationResourceAdmissionDecision<R>, VNextError> {
        let _lifecycle = self.participants[0]
            .session
            .resources()
            .request
            .plan
            .resources
            .read_lifecycle("admit a step invocation")?;
        if self.claimed_backing.has_shared_physical_claims() {
            return Err(invalid_resource(
                "shared Step activation backing requires an ordered single-fence submission wave",
            ));
        }
        let deferred_node_id = request.node_id().clone();
        let deferred_work_fingerprint = request.work_shape().fingerprint().to_owned();
        let prepared = match self.prepare_invocation_scope(request)? {
            PreparedInvocationScopeDecision::Prepared(prepared) => prepared,
            PreparedInvocationScopeDecision::Deferred(deferred) => {
                return Ok(InvocationResourceAdmissionDecision::Deferred(deferred));
            }
            PreparedInvocationScopeDecision::BackingDeferred(deferred) => {
                return Ok(InvocationResourceAdmissionDecision::BackingDeferred(
                    InvocationAdmissionBackingDeferral::new(
                        Arc::clone(self),
                        deferred,
                        deferred_node_id,
                        deferred_work_fingerprint,
                    )?,
                ));
            }
            PreparedInvocationScopeDecision::PermanentRejected(rejected) => {
                return Ok(InvocationResourceAdmissionDecision::PermanentRejected(
                    rejected,
                ));
            }
            PreparedInvocationScopeDecision::RequestStateDeferred(deferred) => {
                return Ok(InvocationResourceAdmissionDecision::RequestStateDeferred(
                    deferred,
                ));
            }
            PreparedInvocationScopeDecision::RequestStateSplitRequired(split) => {
                return Ok(InvocationResourceAdmissionDecision::RequestStateSplitRequired(split));
            }
            PreparedInvocationScopeDecision::RequestStatePoisoned(poison) => {
                return Ok(InvocationResourceAdmissionDecision::RequestStatePoisoned(
                    poison,
                ));
            }
        };
        let PreparedInvocationScope {
            participants,
            participant_frames,
            node_id,
            claimed_backing,
            flight_candidates,
            request_state_hazards,
        } = prepared;
        let batch_invocation_id = issue_batch_invocation_id()?;
        let prepared_participant_flights =
            prepare_participant_flights(&flight_candidates, &node_id)?;
        let topology_keys = participant_frames
            .iter()
            .map(|assignment| {
                ParticipantNodeKey::new(
                    assignment.participant(),
                    assignment.frame_id(),
                    node_id.clone(),
                )
            })
            .collect();
        let active_wave = self.invocation_registry.enter(
            topology_keys,
            batch_invocation_id,
            claimed_backing.work_shape().fingerprint(),
        )?;
        Ok(InvocationResourceAdmissionDecision::Admitted(
            InvocationResourceLease::new(
                Arc::clone(self),
                participants,
                participant_frames,
                node_id,
                batch_invocation_id,
                claimed_backing,
                prepared_participant_flights,
                active_wave,
                request_state_hazards,
            )?,
        ))
    }

    pub fn try_prepare_submission_wave(
        self: &Arc<Self>,
        requests: Vec<InvocationResourceAdmissionRequest>,
    ) -> Result<StepSubmissionWaveAdmissionDecision<R>, VNextError> {
        let plan = &self.participants[0].session.resources().request.plan;
        let plan_nodes = plan.nodes();
        if requests.is_empty()
            || requests.len() != plan_nodes.len()
            || requests
                .iter()
                .zip(plan_nodes)
                .any(|(request, node)| request.node_id() != node.id())
        {
            return Err(invalid_resource(
                "submission wave must cover every plan node exactly once in immutable plan order",
            ));
        }
        let fit_policy = requests[0].fit_policy();
        let pressure_action = requests[0].pressure_action();
        if requests.iter().any(|request| {
            request.fit_policy() != fit_policy || request.pressure_action() != pressure_action
        }) {
            return Err(invalid_resource(
                "submission wave nodes require one admission fit and pressure policy",
            ));
        }
        let work_shape = Arc::clone(&requests[0].work_shape);
        if requests
            .iter()
            .any(|request| request.work_shape.as_ref() != work_shape.as_ref())
        {
            return Err(invalid_resource(
                "submission wave nodes must share one canonical work authority",
            ));
        }
        self.prepare_full_plan_submission_wave(work_shape, fit_policy, pressure_action)
    }

    /// Prepares a canonical immutable-plan node subset that can only be
    /// submitted through the exact determinism restore/readback path.
    pub fn try_prepare_determinism_submission_wave(
        self: &Arc<Self>,
        requests: Vec<InvocationResourceAdmissionRequest>,
    ) -> Result<StepSubmissionWaveAdmissionDecision<R>, VNextError> {
        let plan = &self.participants[0].session.resources().request.plan;
        let plan_nodes = plan.nodes();
        if requests.is_empty() {
            return Err(invalid_resource(
                "determinism submission wave requires a non-empty plan node scope",
            ));
        }
        let node_indices = requests
            .iter()
            .map(|request| {
                plan_nodes
                    .iter()
                    .position(|node| node.id() == request.node_id())
                    .ok_or_else(|| {
                        invalid_resource(
                            "determinism submission wave references an unknown plan node",
                        )
                    })
            })
            .collect::<Result<Vec<_>, _>>()?;
        if node_indices.windows(2).any(|pair| pair[0] >= pair[1]) {
            return Err(invalid_resource(
                "determinism submission wave nodes must be unique and in immutable plan order",
            ));
        }
        let fit_policy = requests[0].fit_policy();
        let pressure_action = requests[0].pressure_action();
        if requests.iter().any(|request| {
            request.fit_policy() != fit_policy || request.pressure_action() != pressure_action
        }) {
            return Err(invalid_resource(
                "determinism submission wave nodes require one admission fit and pressure policy",
            ));
        }
        let work_shape = Arc::clone(&requests[0].work_shape);
        if requests
            .iter()
            .any(|request| request.work_shape.as_ref() != work_shape.as_ref())
        {
            return Err(invalid_resource(
                "determinism submission wave nodes must share one canonical work authority",
            ));
        }
        self.prepare_submission_wave(
            work_shape,
            fit_policy,
            pressure_action,
            &node_indices,
            SubmissionWavePurpose::DeterminismProbe,
        )
    }

    /// Prepares the only production wave topology: every immutable-plan node
    /// in order over one shared participant/work authority. Callers do not
    /// rebuild a per-node request vector for topology already committed by the
    /// plan hash.
    pub fn try_prepare_full_plan_submission_wave(
        self: &Arc<Self>,
        work_shape: Arc<BatchWorkShape>,
        fit_policy: AdmissionFitPolicy,
        pressure_action: AdmissionPressureAction,
    ) -> Result<StepSubmissionWaveAdmissionDecision<R>, VNextError> {
        self.prepare_full_plan_submission_wave(work_shape, fit_policy, pressure_action)
    }

    fn prepare_full_plan_submission_wave(
        self: &Arc<Self>,
        work_shape: Arc<BatchWorkShape>,
        fit_policy: AdmissionFitPolicy,
        pressure_action: AdmissionPressureAction,
    ) -> Result<StepSubmissionWaveAdmissionDecision<R>, VNextError> {
        let node_count = self.participants[0]
            .session
            .resources()
            .request
            .plan
            .nodes()
            .len();
        let node_indices = (0..node_count).collect::<Vec<_>>();
        self.prepare_submission_wave(
            work_shape,
            fit_policy,
            pressure_action,
            &node_indices,
            SubmissionWavePurpose::FullPlan,
        )
    }

    fn prepare_submission_wave(
        self: &Arc<Self>,
        work_shape: Arc<BatchWorkShape>,
        fit_policy: AdmissionFitPolicy,
        pressure_action: AdmissionPressureAction,
        node_indices: &[usize],
        purpose: SubmissionWavePurpose,
    ) -> Result<StepSubmissionWaveAdmissionDecision<R>, VNextError> {
        let _lifecycle = self.participants[0]
            .session
            .resources()
            .request
            .plan
            .resources
            .read_lifecycle("prepare a step submission wave")?;
        let plan = &self.participants[0].session.resources().request.plan;
        let plan_nodes = plan.nodes();
        if plan_nodes.is_empty()
            || node_indices.is_empty()
            || node_indices.windows(2).any(|pair| pair[0] >= pair[1])
            || node_indices
                .last()
                .is_some_and(|node_index| *node_index >= plan_nodes.len())
        {
            return Err(invalid_resource(
                "submission wave requires a non-empty canonical immutable-plan node scope",
            ));
        }
        if work_shape.participants().len() != self.participants.len()
            || work_shape
                .participants()
                .iter()
                .zip(&self.participants)
                .any(|(authority, participant)| {
                    authority.canonical_key() != session_participant_key(&participant.session)
                })
        {
            return Err(invalid_resource(
                "submission wave must bind every step participant exactly once",
            ));
        }
        let participant_authority =
            Arc::new(self.prepare_participant_authority(Arc::clone(&work_shape), fit_policy)?);
        let prepared_nodes = node_indices
            .iter()
            .copied()
            .map(|plan_node_index| {
                PreparedStepSubmissionNode::new(plan_node_index, Arc::clone(&participant_authority))
            })
            .collect::<Vec<_>>();
        let request_participants = participant_authority
            .participants
            .iter()
            .map(|participant| {
                let request = Arc::clone(participant.request_resources());
                RequestStateHazardParticipant::new(
                    Arc::clone(&request.plan.dynamic_pools().request_state_hazards),
                    request.request_authority(),
                    request,
                )
            })
            .collect::<Vec<_>>();
        let request_state_hazards = match plan
            .dynamic_pools()
            .request_state_hazards
            .try_acquire(&request_participants, node_indices)?
        {
            RequestStateHazardAcquireDecision::Acquired(permit) => permit,
            RequestStateHazardAcquireDecision::Deferred(deferred) => {
                return Ok(StepSubmissionWaveAdmissionDecision::RequestStateDeferred(
                    deferred,
                ));
            }
            RequestStateHazardAcquireDecision::SplitRequired(split) => {
                return Ok(StepSubmissionWaveAdmissionDecision::RequestStateSplitRequired(split));
            }
            RequestStateHazardAcquireDecision::Poisoned(poison) => {
                return Ok(StepSubmissionWaveAdmissionDecision::RequestStatePoisoned(
                    poison,
                ));
            }
        };
        let immediate_shape = work_shape.immediate_shape();
        let fit_shape = match fit_policy {
            AdmissionFitPolicy::ImmediateOnly => immediate_shape,
            AdmissionFitPolicy::FullInputMustFit => work_shape.fit_shape(),
        };
        let (demand, requested_slices) = plan.submission_wave_demand(
            immediate_shape,
            fit_shape,
            self.reusable_execution_bucket.as_ref(),
            fit_policy,
            pressure_action,
        )?;
        let prepared_backing = match plan
            .prepare_lane_stable_backing_slices(&self.execution_lane, requested_slices)?
        {
            LaneBackingPrepareDecision::Prepared(prepared) => prepared,
            LaneBackingPrepareDecision::Deferred(deferred) => {
                let deferred_node_work_fingerprints = prepared_nodes
                    .iter()
                    .map(|node| {
                        (
                            node.node_id().clone(),
                            node.work_shape().fingerprint().to_owned(),
                        )
                    })
                    .collect();
                return Ok(StepSubmissionWaveAdmissionDecision::BackingDeferred(
                    StepSubmissionWaveBackingDeferral::new(
                        Arc::clone(self),
                        deferred,
                        deferred_node_work_fingerprints,
                    )?,
                ));
            }
        };
        let logical_capacity = if demand.immediate_claim().is_empty() {
            None
        } else {
            let parents = self
                .participants
                .iter()
                .map(|participant| participant.session.resources().logical_lease())
                .collect::<Vec<_>>();
            match plan
                .logical_admission()
                .try_claim_for_sequences(&parents, &demand)?
            {
                BatchCapacityClaimDecision::Claimed(capacity) => {
                    if !plan
                        .logical_admission()
                        .owns_batch_capacity_claim(&capacity)
                    {
                        return Err(invalid_resource(
                            "submission wave admission returned foreign capacity authority",
                        ));
                    }
                    Some(capacity)
                }
                BatchCapacityClaimDecision::Deferred(deferred) => {
                    return Ok(StepSubmissionWaveAdmissionDecision::Deferred(deferred));
                }
                BatchCapacityClaimDecision::PermanentRejected(rejected) => {
                    return Ok(StepSubmissionWaveAdmissionDecision::PermanentRejected(
                        rejected,
                    ));
                }
            }
        };
        let committed_backing = prepared_backing.commit();
        let has_program_binding_nodes = prepared_nodes.iter().any(|node| {
            plan_nodes[node.plan_node_index()]
                .binding_resource()
                .is_some()
        });
        let program_binding_layout = match (
            self.reusable_execution_bucket.as_ref(),
            has_program_binding_nodes,
        ) {
            (Some(bucket), true) => Some(
                plan.dynamic_pools()
                    .program_binding_layout(bucket.bucket_id())
                    .cloned()
                    .ok_or_else(|| {
                        invalid_resource(
                            "reusable submission wave has no compiled program binding layout",
                        )
                    })?,
            ),
            _ => None,
        };
        let claimed_backing = ClaimedSubmissionWaveBacking::new(
            plan.plan_hash().clone(),
            prepared_nodes.len(),
            plan_nodes.len(),
            Arc::clone(&work_shape),
            demand,
            logical_capacity,
            committed_backing,
            program_binding_layout,
        )?;
        let wave_fingerprint =
            submission_wave_fingerprint(self, &prepared_nodes, &claimed_backing, purpose)?;
        let batch_invocation_id = issue_batch_invocation_id()?;
        let prepared_participant_flights =
            prepare_submission_wave_participant_flights(&participant_authority.flight_candidates)?;
        let covered_participant_nodes = prepared_nodes
            .len()
            .checked_mul(participant_authority.participants.len())
            .ok_or_else(|| invalid_resource("submission wave topology size exceeds usize"))?;
        let active_wave = self.invocation_registry.enter_submission_wave(
            covered_participant_nodes,
            batch_invocation_id,
            &wave_fingerprint,
        )?;
        Ok(StepSubmissionWaveAdmissionDecision::Prepared(
            PreparedStepSubmissionWave {
                claimed_backing,
                initializations: None,
                request_state_hazards,
                nodes: prepared_nodes,
                prepared_participant_flights,
                active_wave,
                step: Arc::clone(self),
                execution_lane_id: self.execution_lane.id(),
                batch_invocation_id,
                fingerprint: wave_fingerprint,
                purpose,
            },
        ))
    }

    fn prepare_invocation_scope(
        &self,
        request: InvocationResourceAdmissionRequest,
    ) -> Result<PreparedInvocationScopeDecision<R>, VNextError> {
        let PreparedInvocationMetadata {
            participants,
            participant_frames,
            participant_session_identities: _,
            flight_candidates,
            node_id,
            work_shape,
            fit_policy,
            pressure_action,
        } = self.prepare_invocation_metadata(request)?;
        let plan = &participants[0].request.plan;
        let node_index = plan
            .nodes()
            .iter()
            .position(|node| node.id() == &node_id)
            .ok_or_else(|| invalid_resource("invocation references an unknown plan node"))?;
        let request_participants = participants
            .iter()
            .map(|participant| {
                let request = Arc::clone(participant.request_resources());
                RequestStateHazardParticipant::new(
                    Arc::clone(&request.plan.dynamic_pools().request_state_hazards),
                    request.request_authority(),
                    request,
                )
            })
            .collect::<Vec<_>>();
        let request_state_hazards = match plan
            .dynamic_pools()
            .request_state_hazards
            .try_acquire(&request_participants, &[node_index])?
        {
            RequestStateHazardAcquireDecision::Acquired(permit) => permit,
            RequestStateHazardAcquireDecision::Deferred(deferred) => {
                return Ok(PreparedInvocationScopeDecision::RequestStateDeferred(
                    deferred,
                ));
            }
            RequestStateHazardAcquireDecision::SplitRequired(split) => {
                return Ok(PreparedInvocationScopeDecision::RequestStateSplitRequired(
                    split,
                ));
            }
            RequestStateHazardAcquireDecision::Poisoned(poison) => {
                return Ok(PreparedInvocationScopeDecision::RequestStatePoisoned(
                    poison,
                ));
            }
        };
        let immediate_shape = work_shape.immediate_shape();
        let fit_shape = match fit_policy {
            AdmissionFitPolicy::ImmediateOnly => immediate_shape,
            AdmissionFitPolicy::FullInputMustFit => work_shape.fit_shape(),
        };
        let (demand, requested_slices) = plan.scoped_demand(
            AllocationLifetime::Invocation,
            Some(&node_id),
            immediate_shape,
            fit_shape,
            self.reusable_execution_bucket.as_ref(),
            fit_policy,
            pressure_action,
        )?;
        let prepared = match plan.prepare_backing_slices(requested_slices)? {
            BackingPrepareDecision::Prepared(prepared) => prepared,
            BackingPrepareDecision::Deferred(deferred) => {
                return Ok(PreparedInvocationScopeDecision::BackingDeferred(deferred));
            }
        };
        let logical_capacity = if demand.immediate_claim().is_empty() {
            None
        } else {
            let parents = participants
                .iter()
                .map(|participant| participant.logical_lease())
                .collect::<Vec<_>>();
            match plan
                .logical_admission()
                .try_claim_for_sequences(&parents, &demand)?
            {
                BatchCapacityClaimDecision::Claimed(capacity) => {
                    let parents_match = capacity
                        .parents()
                        .iter()
                        .map(|parent| (parent.sequence(), parent.request()))
                        .eq(participants.iter().map(|participant| {
                            (
                                participant.sequence_authority(),
                                participant.request_authority(),
                            )
                        }));
                    if !plan
                        .logical_admission()
                        .owns_batch_capacity_claim(&capacity)
                        || !parents_match
                    {
                        return Err(invalid_resource(
                            "invocation admission returned capacity for another participant set",
                        ));
                    }
                    Some(capacity)
                }
                BatchCapacityClaimDecision::Deferred(deferred) => {
                    return Ok(PreparedInvocationScopeDecision::Deferred(deferred));
                }
                BatchCapacityClaimDecision::PermanentRejected(rejected) => {
                    return Ok(PreparedInvocationScopeDecision::PermanentRejected(rejected));
                }
            }
        };
        let backing_slices = prepared.commit();
        let claimed_backing = ClaimedBackingTransaction::new(
            work_shape,
            demand,
            logical_capacity,
            backing_slices,
            None,
        )?;
        Ok(PreparedInvocationScopeDecision::Prepared(
            PreparedInvocationScope {
                participants,
                participant_frames,
                node_id,
                claimed_backing,
                flight_candidates,
                request_state_hazards,
            },
        ))
    }

    fn prepare_invocation_metadata(
        &self,
        request: InvocationResourceAdmissionRequest,
    ) -> Result<PreparedInvocationMetadata<R>, VNextError> {
        let InvocationResourceAdmissionRequest {
            node_id,
            work_shape,
            fit_policy,
            pressure_action,
        } = request;
        let PreparedParticipantAuthority {
            plan_evidence: _,
            participants,
            participant_frames,
            participant_session_identities,
            flight_candidates,
            work_shape,
        } = self.prepare_participant_authority(work_shape, fit_policy)?;
        Ok(PreparedInvocationMetadata {
            participants,
            participant_frames,
            participant_session_identities,
            flight_candidates,
            node_id,
            work_shape,
            fit_policy,
            pressure_action,
        })
    }

    fn prepare_participant_authority(
        &self,
        work_shape: Arc<BatchWorkShape>,
        fit_policy: AdmissionFitPolicy,
    ) -> Result<PreparedParticipantAuthority<R>, VNextError> {
        let immediate_shape = work_shape.immediate_shape();
        let fit_shape = match fit_policy {
            AdmissionFitPolicy::ImmediateOnly => immediate_shape,
            AdmissionFitPolicy::FullInputMustFit => work_shape.fit_shape(),
        };
        let participant_sessions = work_shape
            .participants()
            .iter()
            .map(|authority| {
                self.participants
                    .binary_search_by_key(&authority.canonical_key(), |participant| {
                        session_participant_key(&participant.session)
                    })
                    .map(|index| Arc::clone(&self.participants[index].session))
                    .map_err(|_| {
                        invalid_resource(
                            "invocation participant is not a member of its execution frame",
                        )
                    })
            })
            .collect::<Result<Vec<_>, _>>()?;
        let mut participant_frames = Vec::with_capacity(participant_sessions.len());
        let mut flight_candidates = Vec::with_capacity(participant_sessions.len());
        let mut canonical_work = None;
        for (request_index, participant) in participant_sessions.iter().enumerate() {
            let key = session_participant_key(participant);
            let index = self
                .participants
                .binary_search_by_key(&key, |step_participant| {
                    session_participant_key(&step_participant.session)
                })
                .map_err(|_| {
                    invalid_resource("invocation participant lost its execution-frame assignment")
                })?;
            let step_participant = &self.participants[index];
            let frame = ActiveSequenceFrame {
                frame_id: step_participant.frame.frame_id,
                batch_step_id: self.batch_step_id,
            };
            let admitted = self.work_shape().participant_work()[index].token_span();
            // Imported Step admission requires retained token evidence and
            // stores that same immutable work Arc. Default untracked execution
            // keeps its existing flight-stage validation without another lock.
            if admitted.checkpoint_tokens().is_some() {
                // The held frame prevents frontier advancement or restore.
                let state =
                    participant.slot.state.lock().map_err(|_| {
                        invalid_resource("sequence session state mutex is poisoned")
                    })?;
                let super::SequenceSessionSlotState::Active(active) = &*state else {
                    return Err(invalid_resource("invocation session is no longer active"));
                };
                if active.epoch != participant.epoch
                    || active.fingerprint != participant.fingerprint
                    || active.active_frame != Some(frame)
                {
                    return Err(invalid_resource("invocation lost its exact Step frame"));
                }
                if let Some(span) = active.completed_boundary.bind_imported_invocation_work(
                    admitted,
                    work_shape.participant_work()[request_index].token_span(),
                    participant.resources().request.plan.plan_hash(),
                )? {
                    let work = canonical_work
                        .get_or_insert_with(|| work_shape.participant_work().to_vec());
                    work[request_index] = BatchParticipantTokenSpan::new(
                        work_shape.participants()[request_index],
                        span,
                    );
                }
            }
            participant_frames.push(StepParticipantFrameAssignment::new(
                participant.sequence_authority(),
                participant.request_authority(),
                step_participant.frame.frame_id,
            ));
            flight_candidates.push(ParticipantFlightCandidate {
                slot: Arc::clone(&participant.slot),
                epoch: participant.epoch,
                fingerprint: participant.fingerprint.clone(),
                frame,
                participant: BatchParticipantAuthority::new(
                    participant.sequence_authority(),
                    participant.request_authority(),
                ),
            });
        }
        let participants = participant_sessions
            .iter()
            .map(|participant| Arc::clone(participant.resources()))
            .collect::<Vec<_>>();
        let participant_count = u32::try_from(participants.len())
            .map_err(|_| invalid_resource("invocation participant count exceeds u32"))?;
        if participant_count == 0
            || immediate_shape.sequences() != participant_count
            || fit_shape.sequences() != participant_count
        {
            return Err(invalid_resource(
                "invocation shape sequence count differs from its exact participant set",
            ));
        }
        let participant_authorities = participants
            .iter()
            .map(|participant| {
                BatchParticipantAuthority::new(
                    participant.sequence_authority(),
                    participant.request_authority(),
                )
            })
            .collect::<Vec<_>>();
        if work_shape.participants() != participant_authorities {
            return Err(invalid_resource(
                "invocation work authority differs from selected participants",
            ));
        }
        let participant_session_identities = flight_candidates
            .iter()
            .map(|candidate| (candidate.epoch, candidate.fingerprint.clone()))
            .collect();
        let work_shape = match canonical_work {
            Some(work) => Arc::new(BatchWorkShape::new(work)?),
            None => work_shape,
        };
        Ok(PreparedParticipantAuthority {
            plan_evidence: self.plan_evidence(),
            participants,
            participant_frames,
            participant_session_identities,
            flight_candidates,
            work_shape,
        })
    }
}

impl<R> Drop for StepResourceLease<R>
where
    R: DeviceRuntime,
{
    fn drop(&mut self) {
        if !self.finalized {
            for participant in &self.participants {
                poison_session_frame(&participant.frame);
            }
        }
    }
}

pub enum InvocationResourceAdmissionDecision<R>
where
    R: DeviceRuntime,
{
    Admitted(InvocationResourceLease<R>),
    Deferred(AdmissionDeferred),
    BackingDeferred(InvocationAdmissionBackingDeferral<R>),
    PermanentRejected(AdmissionRejected),
    RequestStateDeferred(RequestStateHazardDeferral),
    RequestStateSplitRequired(RequestStateHazardSplitRequired),
    RequestStatePoisoned(RequestStateHazardPoison),
}

/// Non-cloneable backing authority for one exact node invocation under one
/// live step.
#[must_use = "invocation backing deferral retains its exact step parent"]
pub struct InvocationAdmissionBackingDeferral<R>
where
    R: DeviceRuntime,
{
    backing: PlanBackingDeferral<R>,
    step: Arc<StepResourceLease<R>>,
    node_id: NodeId,
    work_fingerprint: String,
}

impl<R> InvocationAdmissionBackingDeferral<R>
where
    R: DeviceRuntime,
{
    fn new(
        step: Arc<StepResourceLease<R>>,
        evidence: DynamicBackingDeferred,
        node_id: NodeId,
        work_fingerprint: String,
    ) -> Result<Self, VNextError> {
        let resources = Arc::clone(
            &step.participants[0]
                .session
                .resources()
                .request
                .plan
                .resources,
        );
        Ok(Self {
            backing: PlanBackingDeferral::new(resources, evidence)?,
            step,
            node_id,
            work_fingerprint,
        })
    }

    pub fn evidence(&self) -> &DynamicBackingDeferred {
        self.backing.evidence()
    }

    pub fn node_id(&self) -> &NodeId {
        &self.node_id
    }

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

    pub fn maintain(&self) -> Result<DynamicDeferredMaintenanceOutcome, VNextError> {
        if self.step.finalized {
            return Err(invalid_resource(
                "finalized step cannot maintain invocation backing",
            ));
        }
        self.backing.maintain()
    }

    pub fn register_waiter(&self) -> Result<PlanCapacityWaitRegistration<R>, VNextError> {
        self.backing.register_waiter()
    }
}

enum PreparedInvocationScopeDecision<R>
where
    R: DeviceRuntime,
{
    Prepared(PreparedInvocationScope<R>),
    Deferred(AdmissionDeferred),
    BackingDeferred(DynamicBackingDeferred),
    PermanentRejected(AdmissionRejected),
    RequestStateDeferred(RequestStateHazardDeferral),
    RequestStateSplitRequired(RequestStateHazardSplitRequired),
    RequestStatePoisoned(RequestStateHazardPoison),
}

struct PreparedInvocationScope<R>
where
    R: DeviceRuntime,
{
    claimed_backing: ClaimedBackingTransaction,
    participants: Vec<Arc<AdmittedSequenceResources<R>>>,
    participant_frames: Vec<StepParticipantFrameAssignment>,
    flight_candidates: Vec<ParticipantFlightCandidate>,
    node_id: NodeId,
    request_state_hazards: Option<RequestStateHazardPermit<Arc<AdmittedRequestResources<R>>>>,
}

struct PreparedInvocationMetadata<R>
where
    R: DeviceRuntime,
{
    participants: Vec<Arc<AdmittedSequenceResources<R>>>,
    participant_frames: Vec<StepParticipantFrameAssignment>,
    participant_session_identities: Vec<(SequenceSessionEpoch, SequenceSessionFingerprint)>,
    flight_candidates: Vec<ParticipantFlightCandidate>,
    node_id: NodeId,
    work_shape: Arc<BatchWorkShape>,
    fit_policy: AdmissionFitPolicy,
    pressure_action: AdmissionPressureAction,
}

struct PreparedParticipantAuthority<R>
where
    R: DeviceRuntime,
{
    plan_evidence: TrustedPlanRuntimeEvidence,
    participants: Vec<Arc<AdmittedSequenceResources<R>>>,
    participant_frames: Vec<StepParticipantFrameAssignment>,
    participant_session_identities: Vec<(SequenceSessionEpoch, SequenceSessionFingerprint)>,
    flight_candidates: Vec<ParticipantFlightCandidate>,
    work_shape: Arc<BatchWorkShape>,
}

fn submission_wave_fingerprint<R>(
    step: &StepResourceLease<R>,
    nodes: &[PreparedStepSubmissionNode<R>],
    claimed_backing: &ClaimedSubmissionWaveBacking,
    purpose: SubmissionWavePurpose,
) -> Result<String, VNextError>
where
    R: DeviceRuntime,
{
    #[derive(Serialize)]
    struct WaveNodeInput<'a> {
        plan_node_index: usize,
        node_id: &'a NodeId,
    }

    #[derive(Serialize)]
    struct WaveInput<'a> {
        domain: &'static str,
        purpose: SubmissionWavePurpose,
        batch_step_id: BatchStepId,
        plan_hash: &'a super::PlanHash,
        node_count: usize,
        nodes: &'a [WaveNodeInput<'a>],
        step_backing_fingerprint: &'a str,
        invocation_backing_fingerprint: &'a str,
        participant_frames: &'a [StepParticipantFrameAssignment],
        work_fingerprint: &'a str,
    }

    let first = nodes
        .first()
        .ok_or_else(|| invalid_resource("submission wave fingerprint requires plan nodes"))?;
    if nodes.len() != claimed_backing.node_count()
        || nodes
            .iter()
            .any(|node| !Arc::ptr_eq(&node.participant_authority, &first.participant_authority))
        || nodes
            .windows(2)
            .any(|pair| pair[0].plan_node_index() >= pair[1].plan_node_index())
        || first.work_shape() != claimed_backing.work_shape()
    {
        return Err(invalid_resource(
            "submission wave topology differs from its shared plan/work authority",
        ));
    }
    let node_scope = nodes
        .iter()
        .map(|node| WaveNodeInput {
            plan_node_index: node.plan_node_index(),
            node_id: node.node_id(),
        })
        .collect::<Vec<_>>();
    let bytes = serde_json::to_vec(&WaveInput {
        domain: "ferrum.runtime-vnext.step-submission-wave.v4",
        purpose,
        batch_step_id: step.batch_step_id,
        plan_hash: claimed_backing.plan_hash(),
        node_count: claimed_backing.node_count(),
        nodes: &node_scope,
        step_backing_fingerprint: step.claimed_backing.fingerprint(),
        invocation_backing_fingerprint: claimed_backing.fingerprint(),
        participant_frames: first.participant_frames(),
        work_fingerprint: claimed_backing.work_shape().fingerprint(),
    })
    .map_err(|error| {
        invalid_resource(format!(
            "submission wave fingerprint encode failed: {error}"
        ))
    })?;
    Ok(format!("{:x}", Sha256::digest(bytes)))
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "snake_case")]
pub(crate) enum SubmissionWavePurpose {
    FullPlan,
    DeterminismProbe,
}

pub enum StepSubmissionWaveAdmissionDecision<R>
where
    R: DeviceRuntime,
{
    Prepared(PreparedStepSubmissionWave<R>),
    Deferred(AdmissionDeferred),
    BackingDeferred(StepSubmissionWaveBackingDeferral<R>),
    PermanentRejected(AdmissionRejected),
    RequestStateDeferred(RequestStateHazardDeferral),
    RequestStateSplitRequired(RequestStateHazardSplitRequired),
    RequestStatePoisoned(RequestStateHazardPoison),
}

/// Non-cloneable backing authority for one immutable-plan submission wave.
#[must_use = "submission-wave backing deferral retains its exact step parent"]
pub struct StepSubmissionWaveBackingDeferral<R>
where
    R: DeviceRuntime,
{
    backing: PlanBackingDeferral<R>,
    step: Arc<StepResourceLease<R>>,
    node_work_fingerprints: Vec<(NodeId, String)>,
}

impl<R> StepSubmissionWaveBackingDeferral<R>
where
    R: DeviceRuntime,
{
    fn new(
        step: Arc<StepResourceLease<R>>,
        evidence: DynamicBackingDeferred,
        node_work_fingerprints: Vec<(NodeId, String)>,
    ) -> Result<Self, VNextError> {
        let resources = Arc::clone(
            &step.participants[0]
                .session
                .resources()
                .request
                .plan
                .resources,
        );
        Ok(Self {
            backing: PlanBackingDeferral::new(resources, evidence)?,
            step,
            node_work_fingerprints,
        })
    }

    pub fn evidence(&self) -> &DynamicBackingDeferred {
        self.backing.evidence()
    }

    pub fn node_work_fingerprints(&self) -> &[(NodeId, String)] {
        &self.node_work_fingerprints
    }

    pub fn maintain(&self) -> Result<DynamicDeferredMaintenanceOutcome, VNextError> {
        if self.step.finalized {
            return Err(invalid_resource(
                "finalized step cannot maintain submission-wave backing",
            ));
        }
        self.backing.maintain()
    }

    pub fn register_waiter(&self) -> Result<PlanCapacityWaitRegistration<R>, VNextError> {
        self.backing.register_waiter()
    }
}

/// One immutable-plan node projection inside a prepared physical submission
/// wave. The parent wave owns device-flight and retry authority for every node
/// as one atomic unit.
pub struct PreparedStepSubmissionNode<R>
where
    R: DeviceRuntime,
{
    participant_authority: Arc<PreparedParticipantAuthority<R>>,
    node_index: usize,
}

impl<R> PreparedStepSubmissionNode<R>
where
    R: DeviceRuntime,
{
    fn new(node_index: usize, participant_authority: Arc<PreparedParticipantAuthority<R>>) -> Self {
        Self {
            participant_authority,
            node_index,
        }
    }

    pub fn node_id(&self) -> &NodeId {
        self.participant_authority.participants[0]
            .request
            .plan
            .nodes()
            .get(self.node_index)
            .expect("prepared submission node index was derived from the immutable plan")
            .id()
    }

    pub(crate) const fn plan_node_index(&self) -> usize {
        self.node_index
    }

    pub fn participant_count(&self) -> u32 {
        u32::try_from(self.participant_authority.participants.len())
            .expect("wave participant count was validated before admission")
    }

    pub fn participants(
        &self,
    ) -> impl ExactSizeIterator<Item = &Arc<AdmittedSequenceResources<R>>> {
        self.participant_authority.participants.iter()
    }

    pub fn participant_frames(&self) -> &[StepParticipantFrameAssignment] {
        &self.participant_authority.participant_frames
    }

    pub fn work_shape(&self) -> &BatchWorkShape {
        self.participant_authority.work_shape.as_ref()
    }

    pub fn plan_evidence(&self) -> TrustedPlanRuntimeEvidence {
        self.participant_authority.plan_evidence.clone()
    }

    pub(crate) fn plan_evidence_ref(&self) -> &TrustedPlanRuntimeEvidence {
        &self.participant_authority.plan_evidence
    }

    pub(crate) fn runtime(&self) -> &Arc<R> {
        &self.participant_authority.participants[0]
            .request
            .plan
            .runtime()
    }

    pub(crate) fn participant_session_identities(
        &self,
    ) -> impl ExactSizeIterator<Item = (SequenceSessionEpoch, &SequenceSessionFingerprint)> {
        self.participant_authority
            .participant_session_identities
            .iter()
            .map(|(epoch, fingerprint)| (*epoch, fingerprint))
    }
}

/// Exact canonical command wave for one Step. Product waves cover the complete
/// plan; determinism probes carry a sealed purpose for a plan-ordered subset.
/// Node projections and shared Step backing remain owned until the one device
/// fence reaches a terminal state; dropping a prepared wave rolls back.
#[must_use = "a prepared submission wave must be dispatched or explicitly dropped"]
pub struct PreparedStepSubmissionWave<R>
where
    R: DeviceRuntime,
{
    // Drop wave backing and participant flights before releasing the Step.
    claimed_backing: ClaimedSubmissionWaveBacking,
    initializations: Option<PreparedBackingInitializations>,
    request_state_hazards: Option<RequestStateHazardPermit<Arc<AdmittedRequestResources<R>>>>,
    nodes: Vec<PreparedStepSubmissionNode<R>>,
    prepared_participant_flights: Vec<PreparedSubmissionWaveParticipantFlightHold>,
    active_wave: ActiveInvocationWaveGuard,
    step: Arc<StepResourceLease<R>>,
    execution_lane_id: ExecutionLaneId,
    batch_invocation_id: BatchInvocationId,
    fingerprint: String,
    purpose: SubmissionWavePurpose,
}

impl<R> PreparedStepSubmissionWave<R>
where
    R: DeviceRuntime,
{
    pub fn batch_step_id(&self) -> BatchStepId {
        self.step.batch_step_id()
    }

    pub const fn batch_invocation_id(&self) -> BatchInvocationId {
        self.batch_invocation_id
    }

    pub const fn execution_lane_id(&self) -> ExecutionLaneId {
        self.execution_lane_id
    }

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

    pub fn nodes(&self) -> &[PreparedStepSubmissionNode<R>] {
        &self.nodes
    }

    pub(crate) const fn purpose(&self) -> SubmissionWavePurpose {
        self.purpose
    }

    pub(crate) fn record_full_plan_success(
        &self,
        seal: &crate::vnext::SuccessfulWaveCompletionSeal,
    ) -> Result<(), VNextError> {
        super::record_completed_wave(self, seal)
    }

    pub fn claimed_backing(&self) -> &ClaimedSubmissionWaveBacking {
        &self.claimed_backing
    }

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

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

    pub fn node_participant_projection_count(&self) -> usize {
        self.nodes
            .iter()
            .map(|node| node.participant_count() as usize)
            .sum()
    }

    pub fn physical_invocation_ledger_entry_count(&self) -> usize {
        self.active_wave.physical_entry_count()
    }

    pub fn step_resources(&self) -> &Arc<StepResourceLease<R>> {
        &self.step
    }

    pub(crate) fn runtime(&self) -> &Arc<R> {
        self.nodes[0].runtime()
    }

    pub(crate) fn deferred_cleanup_domain(&self) -> DeferredDeviceCleanupDomainId {
        self.nodes[0].participant_authority.participants[0]
            .request
            .plan
            .resources
            .deferred_cleanup_domain
    }

    pub fn has_shared_step_backing(&self) -> bool {
        self.step.claimed_backing().has_shared_physical_claims()
    }

    pub(crate) fn backing_view(
        &self,
        node_index: usize,
        resource_id: &ResourceId,
    ) -> Result<LogicalBackingBufferView<'_, R::Buffer>, VNextError> {
        let node = self
            .nodes
            .get(node_index)
            .ok_or_else(|| invalid_resource("submission wave node index is out of bounds"))?;
        if let Some(authority) = self
            .claimed_backing
            .backing_slices()
            .iter()
            .find(|authority| authority.resource_id() == resource_id)
        {
            return node.participant_authority.participants[0]
                .request
                .plan
                .dynamic_pools()
                .view(authority);
        }
        self.step.backing_view(resource_id)
    }

    pub(crate) fn begin_dispatch(&mut self) -> Result<(), VNextError> {
        match &self.initializations {
            Some(initializations) => initializations.ensure_wave(&self.fingerprint)?,
            None => {
                self.initializations = Some(PreparedBackingInitializations::prepare(
                    &self.step,
                    &self.fingerprint,
                )?);
            }
        }
        begin_submission_wave_participant_flights_dispatch(&mut self.prepared_participant_flights)
    }

    pub(crate) fn encode_backing_initializations(
        &self,
        runtime: &R,
        commands: &mut DeviceCommandBatch<R::Command>,
    ) -> Result<usize, BackingInitializationEncodeError<R::Error>> {
        self.initializations
            .as_ref()
            .ok_or_else(|| {
                BackingInitializationEncodeError::Contract(invalid_resource(
                    "submission wave initialization was not prepared",
                ))
            })?
            .encode(&self.step, runtime, commands)
    }

    pub(crate) fn mark_submission_fence_installed(&mut self) -> Result<(), VNextError> {
        self.initializations
            .as_mut()
            .ok_or_else(|| invalid_resource("submission wave initialization was not prepared"))?
            .mark_in_flight()?;
        self.active_wave.mark_in_flight()?;
        if let Some(hazards) = &mut self.request_state_hazards {
            hazards.mark_submission_fence_installed()?;
        }
        Ok(())
    }

    pub(crate) fn mark_submission_indeterminate(&mut self) {
        if let Some(initializations) = &mut self.initializations {
            initializations.mark_indeterminate();
        }
        if let Some(hazards) = &mut self.request_state_hazards {
            hazards.mark_submission_indeterminate();
        }
    }

    pub(crate) fn finish_backing_initializations(
        &mut self,
        succeeded: bool,
    ) -> Result<(), VNextError> {
        self.initializations
            .as_mut()
            .ok_or_else(|| invalid_resource("submission wave initialization was not prepared"))?
            .finish(succeeded)
    }

    pub(crate) fn finish_request_state_hazards(
        &mut self,
        disposition: RequestStateHazardTerminalDisposition,
    ) -> Result<(), VNextError> {
        match &mut self.request_state_hazards {
            Some(hazards) => hazards.finish(disposition),
            None => Ok(()),
        }
    }

    pub(crate) fn definitely_not_submitted(
        mut self,
    ) -> Result<DefinitelyNotSubmittedWaveRetryAuthority<R>, VNextError> {
        self.active_wave.mark_not_submitted()?;
        reset_submission_wave_participant_flights_after_definitely_not_submitted(
            &mut self.prepared_participant_flights,
        )?;
        let topology_fingerprint = self.fingerprint.clone();
        let prior_attempt = self.batch_invocation_id;
        Ok(DefinitelyNotSubmittedWaveRetryAuthority {
            wave: Some(self),
            topology_fingerprint,
            prior_attempt,
        })
    }

    fn prepare_definitely_not_submitted_retry(
        &mut self,
        fresh_attempt: BatchInvocationId,
        topology_fingerprint: &str,
    ) -> Result<(), VNextError> {
        if self.fingerprint != topology_fingerprint
            || self
                .prepared_participant_flights
                .iter()
                .any(|hold| hold.phase != ParticipantFlightPhase::Prepared)
        {
            return Err(invalid_resource(
                "definitely-not-submitted wave retry topology changed",
            ));
        }
        self.initializations
            .as_ref()
            .ok_or_else(|| invalid_resource("wave retry lost backing initialization authority"))?
            .ensure_wave(topology_fingerprint)?;
        self.active_wave.prepare_retry(fresh_attempt)?;
        self.batch_invocation_id = fresh_attempt;
        Ok(())
    }
}

#[must_use = "a definitely-not-submitted wave retry must be retried or retired"]
pub struct DefinitelyNotSubmittedWaveRetryAuthority<R>
where
    R: DeviceRuntime,
{
    wave: Option<PreparedStepSubmissionWave<R>>,
    topology_fingerprint: String,
    prior_attempt: BatchInvocationId,
}

impl<R> fmt::Debug for DefinitelyNotSubmittedWaveRetryAuthority<R>
where
    R: DeviceRuntime,
{
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("DefinitelyNotSubmittedWaveRetryAuthority")
            .field("prior_attempt", &self.prior_attempt)
            .field("topology_fingerprint", &self.topology_fingerprint)
            .finish_non_exhaustive()
    }
}

impl<R> DefinitelyNotSubmittedWaveRetryAuthority<R>
where
    R: DeviceRuntime,
{
    pub const fn prior_attempt(&self) -> BatchInvocationId {
        self.prior_attempt
    }

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

    pub fn retry(mut self) -> Result<PreparedStepSubmissionWave<R>, VNextError> {
        let fresh_attempt = issue_batch_invocation_id()?;
        let wave = self
            .wave
            .as_mut()
            .ok_or_else(|| invalid_resource("wave retry authority no longer owns its wave"))?;
        wave.prepare_definitely_not_submitted_retry(fresh_attempt, &self.topology_fingerprint)?;
        self.wave
            .take()
            .ok_or_else(|| invalid_resource("validated wave retry authority lost its wave"))
    }
}

/// Exact prepared batch node/provider invocation authority. No device command
/// has been submitted at this layer; dropping it performs the typed
/// definitely-not-submitted participant-flight rollback.
#[must_use = "prepared invocation resources must be dispatched or explicitly dropped"]
pub struct InvocationResourceLease<R>
where
    R: DeviceRuntime,
{
    // Claimed backing is returned before participant-flight and parent frame
    // authorities. It retains the immutable work fingerprint even when empty.
    claimed_backing: ClaimedBackingTransaction,
    initializations: Option<PreparedBackingInitializations>,
    request_state_hazards: Option<RequestStateHazardPermit<Arc<AdmittedRequestResources<R>>>>,
    prepared_participant_flights: Vec<PreparedParticipantFlightHold>,
    active_wave: ActiveInvocationWaveGuard,
    participants: Vec<Arc<AdmittedSequenceResources<R>>>,
    participant_frames: Vec<StepParticipantFrameAssignment>,
    step: Arc<StepResourceLease<R>>,
    node_id: NodeId,
    batch_invocation_id: BatchInvocationId,
}

impl<R> InvocationResourceLease<R>
where
    R: DeviceRuntime,
{
    fn new(
        step: Arc<StepResourceLease<R>>,
        participants: Vec<Arc<AdmittedSequenceResources<R>>>,
        participant_frames: Vec<StepParticipantFrameAssignment>,
        node_id: NodeId,
        batch_invocation_id: BatchInvocationId,
        claimed_backing: ClaimedBackingTransaction,
        prepared_participant_flights: Vec<PreparedParticipantFlightHold>,
        active_wave: ActiveInvocationWaveGuard,
        request_state_hazards: Option<RequestStateHazardPermit<Arc<AdmittedRequestResources<R>>>>,
    ) -> Result<Self, VNextError> {
        if participants.is_empty() {
            return Err(invalid_resource(
                "invocation resources require a non-empty participant set",
            ));
        }
        if participant_frames.len() != participants.len()
            || prepared_participant_flights.len() != participants.len()
            || participant_frames
                .iter()
                .zip(&participants)
                .any(|(assignment, participant)| {
                    assignment.sequence_authority() != participant.sequence_authority()
                        || assignment.request_authority() != participant.request_authority()
                })
        {
            return Err(invalid_resource(
                "invocation frame mapping differs from its exact participant set",
            ));
        }
        if claimed_backing.work_shape().participants().len() != participants.len()
            || claimed_backing
                .work_shape()
                .participants()
                .iter()
                .zip(&participants)
                .any(|(authority, participant)| {
                    authority.sequence_authority() != participant.sequence_authority()
                        || authority.request_authority() != participant.request_authority()
                })
            || claimed_backing.work_shape().immediate_tokens()
                > step.work_shape().immediate_tokens()
            || claimed_backing.work_shape().immediate_pages() > step.work_shape().immediate_pages()
            || claimed_backing.work_shape().fit_tokens() > step.work_shape().fit_tokens()
            || claimed_backing.work_shape().fit_pages() > step.work_shape().fit_pages()
        {
            return Err(invalid_resource(
                "invocation work shape differs from participants or exceeds its step",
            ));
        }
        if let Some(capacity) = claimed_backing.logical_capacity() {
            let coordinator = participants[0].request.plan.logical_admission();
            let parents_match = capacity
                .parents()
                .iter()
                .map(|parent| (parent.sequence(), parent.request()))
                .eq(participants.iter().map(|participant| {
                    (
                        participant.sequence_authority(),
                        participant.request_authority(),
                    )
                }));
            if !coordinator.owns_batch_capacity_claim(capacity) || !parents_match {
                return Err(invalid_resource(
                    "invocation capacity authority differs from its exact participants",
                ));
            }
        }
        Ok(Self {
            claimed_backing,
            initializations: None,
            request_state_hazards,
            prepared_participant_flights,
            active_wave,
            participants,
            participant_frames,
            step,
            node_id,
            batch_invocation_id,
        })
    }

    pub fn node_id(&self) -> &NodeId {
        &self.node_id
    }

    pub const fn batch_invocation_id(&self) -> BatchInvocationId {
        self.batch_invocation_id
    }

    pub fn batch_step_id(&self) -> BatchStepId {
        self.step.batch_step_id()
    }

    pub fn participant_count(&self) -> u32 {
        u32::try_from(self.participants.len())
            .expect("invocation participant count was validated before admission")
    }

    pub fn prepared_participant_count(&self) -> u32 {
        u32::try_from(self.prepared_participant_flights.len())
            .expect("prepared participant count was validated at construction")
    }

    pub(crate) fn participant_session_identities(
        &self,
    ) -> impl ExactSizeIterator<Item = (SequenceSessionEpoch, &SequenceSessionFingerprint)> {
        self.prepared_participant_flights
            .iter()
            .map(|hold| (hold.epoch, &hold.fingerprint))
    }

    pub(crate) fn begin_dispatch(&mut self) -> Result<(), VNextError> {
        let topology_fingerprint = self.retry_topology_fingerprint()?;
        match &self.initializations {
            Some(initializations) => {
                initializations.ensure_wave(&topology_fingerprint)?;
            }
            None => {
                self.initializations = Some(PreparedBackingInitializations::prepare(
                    &self.step,
                    &topology_fingerprint,
                )?);
            }
        }
        begin_participant_flights_dispatch(&mut self.prepared_participant_flights)
    }

    pub(crate) fn encode_backing_initializations(
        &self,
        runtime: &R,
        commands: &mut DeviceCommandBatch<R::Command>,
    ) -> Result<usize, BackingInitializationEncodeError<R::Error>> {
        self.initializations
            .as_ref()
            .ok_or_else(|| {
                BackingInitializationEncodeError::Contract(invalid_resource(
                    "invocation backing initialization was not prepared",
                ))
            })?
            .encode(&self.step, runtime, commands)
    }

    pub(crate) fn mark_submission_fence_installed(&mut self) -> Result<(), VNextError> {
        self.initializations
            .as_mut()
            .ok_or_else(|| invalid_resource("invocation initialization was not prepared"))?
            .mark_in_flight()?;
        self.active_wave.mark_in_flight()?;
        if let Some(hazards) = &mut self.request_state_hazards {
            hazards.mark_submission_fence_installed()?;
        }
        Ok(())
    }

    pub(crate) fn mark_submission_indeterminate(&mut self) {
        if let Some(initializations) = &mut self.initializations {
            initializations.mark_indeterminate();
        }
        if let Some(hazards) = &mut self.request_state_hazards {
            hazards.mark_submission_indeterminate();
        }
    }

    pub(crate) fn finish_backing_initializations(
        &mut self,
        succeeded: bool,
    ) -> Result<(), VNextError> {
        self.initializations
            .as_mut()
            .ok_or_else(|| invalid_resource("invocation initialization was not prepared"))?
            .finish(succeeded)
    }

    pub(crate) fn finish_request_state_hazards(
        &mut self,
        disposition: RequestStateHazardTerminalDisposition,
    ) -> Result<(), VNextError> {
        match &mut self.request_state_hazards {
            Some(hazards) => hazards.finish(disposition),
            None => Ok(()),
        }
    }

    pub(crate) fn definitely_not_submitted(
        mut self,
    ) -> Result<DefinitelyNotSubmittedRetryAuthority<R>, VNextError> {
        self.active_wave.mark_not_submitted()?;
        reset_participant_flights_after_definitely_not_submitted(
            &mut self.prepared_participant_flights,
        )?;
        let topology_fingerprint = self.retry_topology_fingerprint()?;
        let work_fingerprint = self.work_shape().fingerprint().to_owned();
        let prior_attempt = self.batch_invocation_id;
        Ok(DefinitelyNotSubmittedRetryAuthority {
            invocation: Some(self),
            topology_fingerprint,
            work_fingerprint,
            prior_attempt,
        })
    }

    fn prepare_definitely_not_submitted_retry(
        &mut self,
        fresh_attempt: BatchInvocationId,
        topology_fingerprint: &str,
        work_fingerprint: &str,
    ) -> Result<(), VNextError> {
        if self.retry_topology_fingerprint()? != topology_fingerprint
            || self.work_shape().fingerprint() != work_fingerprint
            || self
                .prepared_participant_flights
                .iter()
                .any(|hold| hold.phase != ParticipantFlightPhase::Prepared)
        {
            return Err(invalid_resource(
                "definitely-not-submitted retry topology or work fingerprint changed",
            ));
        }
        self.initializations
            .as_ref()
            .ok_or_else(|| {
                invalid_resource("invocation retry lost backing initialization authority")
            })?
            .ensure_wave(topology_fingerprint)?;
        self.active_wave.prepare_retry(fresh_attempt)?;
        self.batch_invocation_id = fresh_attempt;
        Ok(())
    }

    fn retry_topology_fingerprint(&self) -> Result<String, VNextError> {
        #[derive(Serialize)]
        struct FingerprintInput<'a> {
            domain: &'static str,
            node_id: &'a NodeId,
            participant_frames: &'a [StepParticipantFrameAssignment],
            work_fingerprint: &'a str,
        }
        let bytes = serde_json::to_vec(&FingerprintInput {
            domain: "ferrum.runtime-vnext.invocation-retry-topology.v1",
            node_id: &self.node_id,
            participant_frames: &self.participant_frames,
            work_fingerprint: self.work_shape().fingerprint(),
        })
        .map_err(|error| {
            invalid_resource(format!(
                "invocation retry topology fingerprint encode failed: {error}"
            ))
        })?;
        Ok(format!("{:x}", Sha256::digest(bytes)))
    }

    pub fn coordinator_id(&self) -> LogicalAdmissionCoordinatorId {
        self.participants[0].coordinator_id()
    }

    pub fn participants(
        &self,
    ) -> impl ExactSizeIterator<Item = &Arc<AdmittedSequenceResources<R>>> {
        self.participants.iter()
    }

    pub fn participant_frames(&self) -> &[StepParticipantFrameAssignment] {
        &self.participant_frames
    }

    pub fn step_resources(&self) -> &Arc<StepResourceLease<R>> {
        &self.step
    }

    pub(crate) fn participant_backing_snapshot(
        &self,
        index: usize,
    ) -> Result<&Arc<SequenceBackingSnapshot<R>>, VNextError> {
        let participant = self.participants.get(index).ok_or_else(|| {
            invalid_resource("invocation backing participant index is out of range")
        })?;
        self.step
            .participant_backing_snapshot(BatchParticipantAuthority::new(
                participant.sequence_authority(),
                participant.request_authority(),
            ))
    }

    pub fn backing_slices(&self) -> &[LogicalBackingSliceAuthority] {
        self.claimed_backing.backing_slices()
    }

    pub fn logical_capacity(&self) -> Option<&LogicalBatchCapacityLease> {
        self.claimed_backing.logical_capacity()
    }

    pub fn work_shape(&self) -> &BatchWorkShape {
        self.claimed_backing.work_shape()
    }

    pub fn claimed_backing(&self) -> &ClaimedBackingTransaction {
        &self.claimed_backing
    }

    pub fn plan_evidence(&self) -> TrustedPlanRuntimeEvidence {
        self.step.plan_evidence()
    }

    pub(crate) fn runtime(&self) -> &Arc<R> {
        self.participants[0].request.plan.runtime()
    }

    pub(crate) fn deferred_cleanup_domain(&self) -> DeferredDeviceCleanupDomainId {
        self.participants[0]
            .request
            .plan
            .resources
            .deferred_cleanup_domain
    }

    pub(crate) fn backing_view(
        &self,
        resource_id: &ResourceId,
    ) -> Result<LogicalBackingBufferView<'_, R::Buffer>, VNextError> {
        if let Some(authority) = self
            .claimed_backing
            .backing_slices()
            .iter()
            .find(|authority| authority.resource_id() == resource_id)
        {
            return self.step.participants[0]
                .session
                .resources()
                .request
                .plan
                .dynamic_pools()
                .view(authority);
        }
        self.step.backing_view(resource_id)
    }

    pub(crate) fn participant_backing_views(
        &self,
        resource_id: &ResourceId,
    ) -> Result<Vec<(SequenceAuthorityId, LogicalBackingBufferView<'_, R::Buffer>)>, VNextError>
    {
        self.participants
            .iter()
            .map(|participant| {
                let authority = BatchParticipantAuthority::new(
                    participant.sequence_authority(),
                    participant.request_authority(),
                );
                Ok((
                    participant.sequence_authority(),
                    self.step.participant_backing_view(authority, resource_id)?,
                ))
            })
            .collect()
    }
}

/// The sole retry edge after a device runtime proves that submit did not
/// happen. It owns the exact invocation, topology and work evidence; dropping
/// it retires the ledger tombstone and cannot be relabeled as retryable later.
#[must_use = "a definitely-not-submitted retry authority must be retried or retired"]
pub struct DefinitelyNotSubmittedRetryAuthority<R>
where
    R: DeviceRuntime,
{
    invocation: Option<InvocationResourceLease<R>>,
    topology_fingerprint: String,
    work_fingerprint: String,
    prior_attempt: BatchInvocationId,
}

impl<R> fmt::Debug for DefinitelyNotSubmittedRetryAuthority<R>
where
    R: DeviceRuntime,
{
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("DefinitelyNotSubmittedRetryAuthority")
            .field("prior_attempt", &self.prior_attempt)
            .field("topology_fingerprint", &self.topology_fingerprint)
            .field("work_fingerprint", &self.work_fingerprint)
            .finish_non_exhaustive()
    }
}

impl<R> DefinitelyNotSubmittedRetryAuthority<R>
where
    R: DeviceRuntime,
{
    pub const fn prior_attempt(&self) -> BatchInvocationId {
        self.prior_attempt
    }

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

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

    pub fn retry(mut self) -> Result<InvocationResourceLease<R>, VNextError> {
        let fresh_attempt = issue_batch_invocation_id()?;
        let invocation = self
            .invocation
            .as_mut()
            .ok_or_else(|| invalid_resource("retry authority no longer owns its invocation"))?;
        invocation.prepare_definitely_not_submitted_retry(
            fresh_attempt,
            &self.topology_fingerprint,
            &self.work_fingerprint,
        )?;
        Ok(self
            .invocation
            .take()
            .expect("validated retry authority still owns its invocation"))
    }
}