Skip to main content

ferrum_interfaces/vnext/operation/
determinism.rs

1use std::collections::{BTreeMap, BTreeSet};
2use std::sync::Arc;
3
4use serde::Serialize;
5use sha2::{Digest, Sha256};
6
7use super::dispatch_contract::{BoundDeviceSubmissionAttribution, ProfiledSubmissionHandle};
8use super::foundation::invalid_operation;
9use super::{
10    packed_step_token_range_to_participant_local_readback, BatchOperationIdentity,
11    BatchedOperationInvocation, BoundOperationProvider, ElementType,
12};
13use crate::vnext::{
14    AllocationKind, AllocationLifetime, BatchInvocationId, BatchParticipantAuthority, BufferUsage,
15    CompletionHandle, CompletionReadbackBatchRequest, CompletionReadbackCollectionObservation,
16    CompletionReadbackCollectionRequest, CompletionReadbackDisposition, CompletionReadbackRequest,
17    DeviceCommandPhase, DeviceComputePathRequirement, DeviceExecutionPath,
18    DeviceReusableExecutionProgramId, DeviceRuntime, ExecutablePlanView,
19    ExecutionDeterminismInitializationKind, ExecutionDeterminismInitializationSpec,
20    ExecutionDeterminismValueExtent, ExecutionDeterminismValueLocation,
21    ExecutionDeterminismWitnessKind, ExecutionDeterminismWitnessPlan,
22    ExecutionDeterminismWitnessSpec, HostTransferLayout, NodeId, OperationCompletionDisposition,
23    PlanHash, PreparedStepSubmissionWave, ResourceId, ResourceWorkShape, SequenceSession,
24    SubmittedOperationReceipt, TrustedActiveSequenceBinding, VNextError,
25};
26
27const LOGICAL_RESTORE_FINGERPRINT_DOMAIN: &[u8] =
28    b"ferrum.runtime-vnext.determinism-logical-restore.v2";
29const EXTERNAL_INPUT_FINGERPRINT_DOMAIN: &[u8] =
30    b"ferrum.runtime-vnext.determinism-external-input.v2";
31const INITIAL_STATE_FINGERPRINT_DOMAIN: &[u8] =
32    b"ferrum.runtime-vnext.determinism-initial-state.v2";
33const NO_RNG_STATE_FINGERPRINT_DOMAIN: &[u8] = b"ferrum.runtime-vnext.determinism-no-rng-state.v1";
34
35fn hash_u64(hasher: &mut Sha256, value: u64) {
36    hasher.update(value.to_le_bytes());
37}
38
39fn hash_bytes(hasher: &mut Sha256, bytes: &[u8]) -> Result<(), VNextError> {
40    hash_u64(
41        hasher,
42        u64::try_from(bytes.len())
43            .map_err(|_| invalid_operation("determinism restore fingerprint input exceeds u64"))?,
44    );
45    hasher.update(bytes);
46    Ok(())
47}
48
49fn hash_ranges(
50    hasher: &mut Sha256,
51    ranges: &[SubmissionWaveDeterminismLogicalRange],
52) -> Result<(), VNextError> {
53    hash_u64(
54        hasher,
55        u64::try_from(ranges.len())
56            .map_err(|_| invalid_operation("determinism restore range count exceeds u64"))?,
57    );
58    for range in ranges {
59        hash_u64(hasher, range.logical_offset_bytes());
60        hash_u64(hasher, range.length_bytes());
61    }
62    Ok(())
63}
64
65/// One exact logical range as seen by a provider for one prepared participant.
66#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
67pub struct SubmissionWaveDeterminismLogicalRange {
68    logical_offset_bytes: u64,
69    length_bytes: u64,
70}
71
72/// Stable semantic participant authority for one physically canonical batch.
73///
74/// Runtime resource ownership and provider bindings remain indexed by physical
75/// batch position. Determinism comparisons use the logical position so fresh
76/// admissions may choose different sequence authorities without changing the
77/// semantic execution identity.
78#[derive(Debug, Clone, PartialEq, Eq)]
79pub struct SubmissionWaveDeterminismParticipantOrder {
80    physical_authorities: Vec<BatchParticipantAuthority>,
81    logical_authorities: Vec<BatchParticipantAuthority>,
82    physical_to_logical: Vec<u32>,
83    logical_to_physical: Vec<u32>,
84}
85
86impl SubmissionWaveDeterminismParticipantOrder {
87    /// Binds a caller-defined semantic order to the exact physical authorities
88    /// carried by one prepared wave. The logical order cannot be expressed as
89    /// a naked permutation: every position must name one authority in the wave.
90    pub fn from_logical_participant_sessions<R: DeviceRuntime>(
91        wave: &PreparedStepSubmissionWave<R>,
92        logical_sessions: &[Arc<SequenceSession<R>>],
93    ) -> Result<Self, VNextError> {
94        let logical_authorities = logical_sessions
95            .iter()
96            .map(|session| {
97                session.ensure_open_identity()?;
98                Ok(BatchParticipantAuthority::new(
99                    session.sequence_authority(),
100                    session.request_authority(),
101                ))
102            })
103            .collect::<Result<Vec<_>, VNextError>>()?;
104        Self::from_logical_participant_authorities(wave, logical_authorities)
105    }
106
107    fn from_logical_participant_authorities<R: DeviceRuntime>(
108        wave: &PreparedStepSubmissionWave<R>,
109        logical_authorities: Vec<BatchParticipantAuthority>,
110    ) -> Result<Self, VNextError> {
111        let physical_authorities = prepared_wave_participant_authorities(wave)?;
112        if logical_authorities.len() != physical_authorities.len()
113            || logical_authorities
114                .iter()
115                .enumerate()
116                .any(|(index, authority)| {
117                    logical_authorities[..index].contains(authority)
118                        || !physical_authorities.contains(authority)
119                })
120        {
121            return Err(invalid_operation(
122                "determinism logical participant authorities differ from the prepared wave",
123            ));
124        }
125        let physical_to_logical = physical_authorities
126            .iter()
127            .map(|authority| {
128                logical_authorities
129                    .iter()
130                    .position(|candidate| candidate == authority)
131                    .and_then(|index| u32::try_from(index).ok())
132                    .ok_or_else(|| {
133                        invalid_operation(
134                            "determinism physical participant lacks a logical authority",
135                        )
136                    })
137            })
138            .collect::<Result<Vec<_>, VNextError>>()?;
139        let logical_to_physical = validate_participant_permutation(&physical_to_logical)?;
140        let order = Self {
141            physical_authorities,
142            logical_authorities,
143            physical_to_logical,
144            logical_to_physical,
145        };
146        order.validate_for_wave(wave)?;
147        Ok(order)
148    }
149
150    fn identity_for_prepared_wave<R: DeviceRuntime>(
151        wave: &PreparedStepSubmissionWave<R>,
152    ) -> Result<Self, VNextError> {
153        let authorities = prepared_wave_participant_authorities(wave)?;
154        Self::from_logical_participant_authorities(wave, authorities)
155    }
156
157    pub fn participant_count(&self) -> u32 {
158        u32::try_from(self.physical_to_logical.len())
159            .expect("determinism participant order count was validated")
160    }
161
162    pub fn logical_index_for_physical(&self, physical_index: u32) -> Option<u32> {
163        usize::try_from(physical_index)
164            .ok()
165            .and_then(|index| self.physical_to_logical.get(index))
166            .copied()
167    }
168
169    pub fn physical_index_for_logical(&self, logical_index: u32) -> Option<u32> {
170        usize::try_from(logical_index)
171            .ok()
172            .and_then(|index| self.logical_to_physical.get(index))
173            .copied()
174    }
175
176    fn validate_for_wave<R: DeviceRuntime>(
177        &self,
178        wave: &PreparedStepSubmissionWave<R>,
179    ) -> Result<(), VNextError> {
180        let actual_physical = prepared_wave_participant_authorities(wave)?;
181        let actual_logical = self.reorder_physical_to_logical(&actual_physical)?;
182        if actual_physical != self.physical_authorities
183            || actual_logical != self.logical_authorities
184        {
185            return Err(invalid_operation(
186                "determinism participant order is not bound to this prepared wave",
187            ));
188        }
189        Ok(())
190    }
191
192    fn reorder_physical_to_logical<T: Clone>(&self, physical: &[T]) -> Result<Vec<T>, VNextError> {
193        if physical.len() != self.physical_to_logical.len() {
194            return Err(invalid_operation(
195                "determinism participant order differs from its physical values",
196            ));
197        }
198        self.logical_to_physical
199            .iter()
200            .map(|physical_index| {
201                physical
202                    .get(usize::try_from(*physical_index).expect("u32 physical index fits usize"))
203                    .cloned()
204                    .ok_or_else(|| {
205                        invalid_operation(
206                            "determinism participant order references an absent physical value",
207                        )
208                    })
209            })
210            .collect()
211    }
212}
213
214fn validate_participant_permutation(physical_to_logical: &[u32]) -> Result<Vec<u32>, VNextError> {
215    if physical_to_logical.is_empty() {
216        return Err(invalid_operation(
217            "determinism participant order requires a non-empty batch",
218        ));
219    }
220    let participant_count = u32::try_from(physical_to_logical.len())
221        .map_err(|_| invalid_operation("determinism participant order count exceeds u32"))?;
222    let mut logical_to_physical = vec![u32::MAX; physical_to_logical.len()];
223    for (physical_index, logical_index) in physical_to_logical.iter().copied().enumerate() {
224        if logical_index >= participant_count {
225            return Err(invalid_operation(
226                "determinism participant order contains an out-of-range logical index",
227            ));
228        }
229        let inverse = logical_to_physical
230            .get_mut(usize::try_from(logical_index).expect("u32 logical index fits usize"))
231            .expect("logical index was range checked");
232        if *inverse != u32::MAX {
233            return Err(invalid_operation(
234                "determinism participant order is not a one-to-one permutation",
235            ));
236        }
237        *inverse = u32::try_from(physical_index)
238            .map_err(|_| invalid_operation("determinism physical participant index exceeds u32"))?;
239    }
240    if logical_to_physical
241        .iter()
242        .any(|physical_index| *physical_index == u32::MAX)
243    {
244        return Err(invalid_operation(
245            "determinism participant order does not cover every logical participant",
246        ));
247    }
248    Ok(logical_to_physical)
249}
250
251fn prepared_wave_participant_authorities<R: DeviceRuntime>(
252    wave: &PreparedStepSubmissionWave<R>,
253) -> Result<Vec<BatchParticipantAuthority>, VNextError> {
254    let authorities = wave
255        .nodes()
256        .first()
257        .map(|node| node.work_shape().participants().to_vec())
258        .filter(|participants| !participants.is_empty())
259        .ok_or_else(|| {
260            invalid_operation("determinism participant order requires a non-empty prepared wave")
261        })?;
262    if wave
263        .nodes()
264        .iter()
265        .any(|node| node.work_shape().participants() != authorities)
266    {
267        return Err(invalid_operation(
268            "determinism prepared-wave nodes disagree on physical participant authorities",
269        ));
270    }
271    Ok(authorities)
272}
273
274impl SubmissionWaveDeterminismLogicalRange {
275    pub const fn logical_offset_bytes(self) -> u64 {
276        self.logical_offset_bytes
277    }
278
279    pub const fn length_bytes(self) -> u64 {
280        self.length_bytes
281    }
282}
283
284/// Provider-visible deterministic I/O layout for one exact prepared wave.
285///
286/// The immutable plan defines which semantic values are required. This layout
287/// closes the remaining runtime boundary: packed versus participant-local
288/// token offsets and the exact page-quantized state prefix visible to each
289/// provider invocation.
290#[derive(Debug, Clone, PartialEq, Eq)]
291pub struct SubmissionWaveDeterminismRestoreLayout {
292    witness_plan: ExecutionDeterminismWitnessPlan,
293    participant_order: SubmissionWaveDeterminismParticipantOrder,
294    batch_invocation_id: BatchInvocationId,
295    claimed_backing_fingerprint: String,
296    node_logical_work_fingerprints: Vec<String>,
297    participant_initialization_ranges: Vec<Vec<SubmissionWaveDeterminismLogicalRange>>,
298    logical_participant_initialization_ranges: Vec<Vec<SubmissionWaveDeterminismLogicalRange>>,
299    witness_participant_ranges: Vec<Vec<SubmissionWaveDeterminismLogicalRange>>,
300    logical_witness_participant_ranges: Vec<Vec<SubmissionWaveDeterminismLogicalRange>>,
301}
302
303impl SubmissionWaveDeterminismRestoreLayout {
304    #[allow(clippy::too_many_arguments)]
305    pub fn from_prepared_wave<'binding, R, I>(
306        runtime: &R,
307        providers: &[BoundOperationProvider<'_, R>],
308        resolved: &dyn ExecutablePlanView,
309        batch_identity: &BatchOperationIdentity,
310        active_bindings: I,
311        wave: &PreparedStepSubmissionWave<R>,
312    ) -> Result<Self, VNextError>
313    where
314        R: DeviceRuntime,
315        I: Clone + ExactSizeIterator<Item = &'binding TrustedActiveSequenceBinding>,
316    {
317        Self::from_prepared_wave_with_participant_order(
318            runtime,
319            providers,
320            resolved,
321            batch_identity,
322            active_bindings,
323            SubmissionWaveDeterminismParticipantOrder::identity_for_prepared_wave(wave)?,
324            wave,
325        )
326    }
327
328    #[allow(clippy::too_many_arguments)]
329    pub fn from_prepared_wave_with_participant_order<'binding, R, I>(
330        runtime: &R,
331        providers: &[BoundOperationProvider<'_, R>],
332        resolved: &dyn ExecutablePlanView,
333        batch_identity: &BatchOperationIdentity,
334        active_bindings: I,
335        participant_order: SubmissionWaveDeterminismParticipantOrder,
336        wave: &PreparedStepSubmissionWave<R>,
337    ) -> Result<Self, VNextError>
338    where
339        R: DeviceRuntime,
340        I: Clone + ExactSizeIterator<Item = &'binding TrustedActiveSequenceBinding>,
341    {
342        let node_ids = wave
343            .nodes()
344            .iter()
345            .map(|node| node.node_id().clone())
346            .collect::<Vec<_>>();
347        let witness_plan = resolved
348            .execution_plan()
349            .determinism_witness_plan_for_nodes(&node_ids)?;
350        let participant_count = wave
351            .nodes()
352            .first()
353            .map(|node| node.participant_count() as usize)
354            .filter(|count| *count > 0)
355            .ok_or_else(|| {
356                invalid_operation("determinism restore layout requires a non-empty prepared wave")
357            })?;
358        if providers.len() != wave.nodes().len()
359            || providers.len() != batch_identity.node_count()
360            || active_bindings.len() != participant_count
361            || usize::try_from(participant_order.participant_count()).ok()
362                != Some(participant_count)
363            || participant_order.validate_for_wave(wave).is_err()
364            || active_bindings
365                .clone()
366                .zip(&participant_order.physical_authorities)
367                .any(|(binding, authority)| {
368                    binding.sequence_authority() != authority.sequence_authority()
369                })
370            || batch_identity.nodes().iter().any(|node| {
371                node.participants().len() != participant_count
372                    || node
373                        .participants()
374                        .iter()
375                        .zip(&participant_order.physical_authorities)
376                        .any(|(participant, authority)| {
377                            participant.node_key().sequence_authority()
378                                != authority.sequence_authority()
379                                || participant.node_key().request_authority()
380                                    != authority.request_authority()
381                        })
382            })
383            || witness_plan.plan_hash() != resolved.execution_plan().plan_hash()
384            || witness_plan.plan_hash() != batch_identity.plan_hash()
385            || batch_identity.batch_invocation_id() != wave.batch_invocation_id()
386            || batch_identity.claimed_backing_fingerprint() != wave.fingerprint()
387            || wave.nodes().iter().enumerate().any(|(node_index, node)| {
388                node.plan_evidence_ref().plan_hash() != witness_plan.plan_hash()
389                    || node.participant_count() as usize != participant_count
390                    || node.work_shape().participant_work().len() != participant_count
391                    || node.work_shape().participant_token_ranges().len() != participant_count
392                    || batch_identity.node_id_at(node_index) != Some(node.node_id())
393                    || batch_identity.node_participant_count(node_index) != Some(participant_count)
394            })
395        {
396            return Err(invalid_operation(
397                "determinism restore layout differs from its immutable plan or participant topology",
398            ));
399        }
400
401        let invocations = providers
402            .iter()
403            .zip(batch_identity.nodes())
404            .enumerate()
405            .map(|(node_index, (provider, node_identity))| {
406                provider.validate_binding(resolved, node_identity.node_id())?;
407                BatchedOperationInvocation::from_wave_node(
408                    runtime,
409                    resolved,
410                    provider.dispatch(),
411                    batch_identity,
412                    node_identity,
413                    wave,
414                    node_index,
415                    active_bindings.clone(),
416                )
417            })
418            .collect::<Result<Vec<_>, VNextError>>()?;
419
420        let mut participant_initialization_ranges =
421            vec![Vec::with_capacity(witness_plan.initializations().len()); participant_count];
422        for initialization in witness_plan.initializations() {
423            for (participant_index, ranges) in
424                participant_initialization_ranges.iter_mut().enumerate()
425            {
426                let mut bound_range = None;
427                for consumer_node_id in initialization.consumer_node_ids() {
428                    let invocation = invocations
429                        .iter()
430                        .find(|invocation| invocation.node_id() == consumer_node_id)
431                        .ok_or_else(|| {
432                            invalid_operation(
433                                "determinism initialization consumer is absent from its prepared wave",
434                            )
435                        })?;
436                    let candidate = prepared_location_range(
437                        wave,
438                        invocation,
439                        participant_index,
440                        initialization.location(),
441                    )?;
442                    if bound_range
443                        .replace(candidate)
444                        .is_some_and(|bound| bound != candidate)
445                    {
446                        return Err(invalid_operation(
447                            "determinism initialization consumers disagree on the provider-visible range",
448                        ));
449                    }
450                }
451                ranges.push(bound_range.ok_or_else(|| {
452                    invalid_operation(
453                        "determinism initialization has no prepared-wave consumer range",
454                    )
455                })?);
456            }
457        }
458
459        let witness_participant_ranges = witness_plan
460            .witnesses()
461            .iter()
462            .map(|witness| {
463                let invocation = invocations
464                    .iter()
465                    .find(|invocation| invocation.node_id() == witness.node_id())
466                    .ok_or_else(|| {
467                        invalid_operation(
468                            "determinism witness node is absent from its prepared wave",
469                        )
470                    })?;
471                (0..participant_count)
472                    .map(|participant_index| {
473                        prepared_location_range(
474                            wave,
475                            invocation,
476                            participant_index,
477                            witness.location(),
478                        )
479                    })
480                    .collect::<Result<Vec<_>, VNextError>>()
481            })
482            .collect::<Result<Vec<_>, VNextError>>()?;
483
484        let logical_participant_initialization_ranges = logical_initialization_ranges(
485            wave,
486            &participant_order,
487            witness_plan.initializations(),
488            &participant_initialization_ranges,
489        )?;
490        let logical_witness_participant_ranges = witness_plan
491            .witnesses()
492            .iter()
493            .zip(&witness_participant_ranges)
494            .map(|(witness, ranges)| {
495                logical_participant_ranges(wave, &participant_order, witness.location(), ranges)
496            })
497            .collect::<Result<Vec<_>, VNextError>>()?;
498        let node_logical_work_fingerprints = logical_work_fingerprints(wave, &participant_order)?;
499
500        Ok(Self {
501            witness_plan,
502            participant_order,
503            batch_invocation_id: batch_identity.batch_invocation_id(),
504            claimed_backing_fingerprint: batch_identity.claimed_backing_fingerprint().to_owned(),
505            node_logical_work_fingerprints,
506            participant_initialization_ranges,
507            logical_participant_initialization_ranges,
508            witness_participant_ranges,
509            logical_witness_participant_ranges,
510        })
511    }
512
513    pub fn witness_plan(&self) -> &ExecutionDeterminismWitnessPlan {
514        &self.witness_plan
515    }
516
517    pub fn participant_order(&self) -> &SubmissionWaveDeterminismParticipantOrder {
518        &self.participant_order
519    }
520
521    pub fn participant_count(&self) -> u32 {
522        u32::try_from(self.participant_initialization_ranges.len())
523            .expect("determinism restore layout participant count was validated")
524    }
525
526    pub fn participant_initialization_ranges(
527        &self,
528        participant_index: u32,
529    ) -> Option<&[SubmissionWaveDeterminismLogicalRange]> {
530        usize::try_from(participant_index)
531            .ok()
532            .and_then(|index| self.participant_initialization_ranges.get(index))
533            .map(Vec::as_slice)
534    }
535
536    pub fn witness_participant_ranges(
537        &self,
538        witness_index: usize,
539    ) -> Option<&[SubmissionWaveDeterminismLogicalRange]> {
540        self.witness_participant_ranges
541            .get(witness_index)
542            .map(Vec::as_slice)
543    }
544
545    pub fn bind(
546        self,
547        participant_payloads: Vec<Vec<Vec<u8>>>,
548    ) -> Result<SubmissionWaveDeterminismRestore, VNextError> {
549        if participant_payloads.len() != self.participant_initialization_ranges.len()
550            || participant_payloads
551                .iter()
552                .zip(&self.participant_initialization_ranges)
553                .any(|(payloads, ranges)| payloads.len() != ranges.len())
554        {
555            return Err(invalid_operation(
556                "determinism restore must cover every work-bound initialization and participant",
557            ));
558        }
559        for ((payloads, ranges), initializations) in participant_payloads
560            .iter()
561            .zip(&self.participant_initialization_ranges)
562            .zip(std::iter::repeat(self.witness_plan.initializations()))
563        {
564            for ((initialization, bytes), range) in initializations.iter().zip(payloads).zip(ranges)
565            {
566                let location = initialization.location();
567                if u64::try_from(bytes.len()).ok() != Some(range.length_bytes())
568                    || bytes.len()
569                        % usize::try_from(location.element_type().size_bytes())
570                            .expect("element width fits usize")
571                        != 0
572                {
573                    return Err(invalid_operation(
574                        "determinism restore payload differs from its complete typed initialization range",
575                    ));
576                }
577            }
578        }
579        Ok(SubmissionWaveDeterminismRestore {
580            layout: self,
581            participant_payloads,
582        })
583    }
584}
585
586/// Complete participant-major input/state restoration bound to one immutable
587/// plan and one exact prepared-wave work topology.
588#[derive(Debug, Clone, PartialEq, Eq)]
589pub struct SubmissionWaveDeterminismRestore {
590    layout: SubmissionWaveDeterminismRestoreLayout,
591    participant_payloads: Vec<Vec<Vec<u8>>>,
592}
593
594/// Domain-separated semantic initialization identities for one deterministic
595/// execution. These values are derived from the exact typed restore payloads;
596/// artifact producers cannot substitute independently computed digests.
597#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
598pub struct SubmissionWaveDeterminismInitializationIdentity {
599    input_sha256: String,
600    rng_sha256: String,
601    initial_state_sha256: String,
602}
603
604impl SubmissionWaveDeterminismInitializationIdentity {
605    pub fn input_sha256(&self) -> &str {
606        &self.input_sha256
607    }
608
609    pub fn rng_sha256(&self) -> &str {
610        &self.rng_sha256
611    }
612
613    pub fn initial_state_sha256(&self) -> &str {
614        &self.initial_state_sha256
615    }
616}
617
618impl SubmissionWaveDeterminismRestore {
619    pub fn layout(&self) -> &SubmissionWaveDeterminismRestoreLayout {
620        &self.layout
621    }
622
623    pub fn plan_hash(&self) -> &PlanHash {
624        self.layout.witness_plan.plan_hash()
625    }
626
627    pub fn node_ids(&self) -> &[NodeId] {
628        self.layout.witness_plan.node_ids()
629    }
630
631    pub fn initializations(&self) -> &[ExecutionDeterminismInitializationSpec] {
632        self.layout.witness_plan.initializations()
633    }
634
635    pub fn participant_count(&self) -> u32 {
636        u32::try_from(self.participant_payloads.len())
637            .expect("determinism restore participant count was validated")
638    }
639
640    pub fn participant_payloads(&self, participant_index: u32) -> Option<&[Vec<u8>]> {
641        usize::try_from(participant_index)
642            .ok()
643            .and_then(|index| self.participant_payloads.get(index))
644            .map(Vec::as_slice)
645    }
646
647    fn initialization_fingerprint(
648        &self,
649        domain: &[u8],
650        include: impl Fn(&ExecutionDeterminismInitializationKind) -> bool,
651    ) -> Result<String, VNextError> {
652        let initializations = self.initializations();
653        let selected_count = initializations
654            .iter()
655            .filter(|initialization| include(initialization.kind()))
656            .count()
657            .checked_mul(self.participant_payloads.len())
658            .ok_or_else(|| invalid_operation("determinism initialization count overflows"))?;
659        let mut hasher = Sha256::new();
660        hash_bytes(&mut hasher, domain)?;
661        hash_bytes(
662            &mut hasher,
663            self.layout.witness_plan.fingerprint()?.as_bytes(),
664        )?;
665        hash_u64(
666            &mut hasher,
667            u64::try_from(selected_count)
668                .map_err(|_| invalid_operation("determinism initialization count exceeds u64"))?,
669        );
670        for (participant_index, ranges) in self
671            .layout
672            .logical_participant_initialization_ranges
673            .iter()
674            .enumerate()
675        {
676            let physical_index = self
677                .layout
678                .participant_order
679                .physical_index_for_logical(u32::try_from(participant_index).map_err(|_| {
680                    invalid_operation("determinism logical participant index exceeds u32")
681                })?)
682                .and_then(|index| usize::try_from(index).ok())
683                .ok_or_else(|| {
684                    invalid_operation("determinism logical participant lacks a physical payload")
685                })?;
686            let payloads = self
687                .participant_payloads
688                .get(physical_index)
689                .ok_or_else(|| invalid_operation("determinism physical payload is absent"))?;
690            for (initialization_index, ((initialization, payload), range)) in
691                initializations.iter().zip(payloads).zip(ranges).enumerate()
692            {
693                if !include(initialization.kind()) {
694                    continue;
695                }
696                hash_u64(
697                    &mut hasher,
698                    u64::try_from(participant_index).map_err(|_| {
699                        invalid_operation("determinism participant index exceeds u64")
700                    })?,
701                );
702                hash_u64(
703                    &mut hasher,
704                    u64::try_from(initialization_index).map_err(|_| {
705                        invalid_operation("determinism initialization index exceeds u64")
706                    })?,
707                );
708                let encoded = serde_json::to_vec(initialization).map_err(|error| {
709                    invalid_operation(format!(
710                        "determinism initialization identity serialization failed: {error}"
711                    ))
712                })?;
713                hash_bytes(&mut hasher, &encoded)?;
714                hash_u64(&mut hasher, range.logical_offset_bytes());
715                hash_u64(&mut hasher, range.length_bytes());
716                hash_bytes(&mut hasher, payload)?;
717            }
718        }
719        Ok(format!("{:x}", hasher.finalize()))
720    }
721
722    /// Exact input/state digests restored for this wave. The current model
723    /// execution graph contains no stochastic operation input, so the RNG
724    /// identity is an explicit domain-separated empty state rather than a
725    /// collector-supplied placeholder.
726    pub fn initialization_identity(
727        &self,
728    ) -> Result<SubmissionWaveDeterminismInitializationIdentity, VNextError> {
729        let input_sha256 =
730            self.initialization_fingerprint(EXTERNAL_INPUT_FINGERPRINT_DOMAIN, |kind| {
731                matches!(
732                    kind,
733                    ExecutionDeterminismInitializationKind::ExternalInput { .. }
734                )
735            })?;
736        let initial_state_sha256 = self
737            .initialization_fingerprint(INITIAL_STATE_FINGERPRINT_DOMAIN, |kind| {
738                matches!(kind, ExecutionDeterminismInitializationKind::State { .. })
739            })?;
740        let mut rng = Sha256::new();
741        hash_bytes(&mut rng, NO_RNG_STATE_FINGERPRINT_DOMAIN)?;
742        hash_bytes(&mut rng, self.layout.witness_plan.fingerprint()?.as_bytes())?;
743        hash_u64(&mut rng, 0);
744        Ok(SubmissionWaveDeterminismInitializationIdentity {
745            input_sha256,
746            rng_sha256: format!("{:x}", rng.finalize()),
747            initial_state_sha256,
748        })
749    }
750
751    /// Stable identity of the logical input/state image restored before each
752    /// execution.
753    ///
754    /// Physical batch, backing, lane, and allocation identities are excluded
755    /// intentionally: repeated executions must allocate fresh submission
756    /// authority while proving that they restored identical semantic bytes.
757    /// Scratch poison is also excluded because it is the independent variable
758    /// compared by the determinism gate.
759    pub fn logical_fingerprint(&self) -> Result<String, VNextError> {
760        let mut hasher = Sha256::new();
761        hash_bytes(&mut hasher, LOGICAL_RESTORE_FINGERPRINT_DOMAIN)?;
762        hash_bytes(
763            &mut hasher,
764            self.layout.witness_plan.fingerprint()?.as_bytes(),
765        )?;
766
767        hash_u64(
768            &mut hasher,
769            u64::try_from(self.layout.node_logical_work_fingerprints.len())
770                .map_err(|_| invalid_operation("determinism restore node count exceeds u64"))?,
771        );
772        for fingerprint in &self.layout.node_logical_work_fingerprints {
773            hash_bytes(&mut hasher, fingerprint.as_bytes())?;
774        }
775
776        hash_u64(
777            &mut hasher,
778            u64::try_from(self.layout.participant_initialization_ranges.len()).map_err(|_| {
779                invalid_operation("determinism restore participant count exceeds u64")
780            })?,
781        );
782        for ranges in &self.layout.logical_participant_initialization_ranges {
783            hash_ranges(&mut hasher, ranges)?;
784        }
785
786        hash_u64(
787            &mut hasher,
788            u64::try_from(self.layout.witness_participant_ranges.len())
789                .map_err(|_| invalid_operation("determinism restore witness count exceeds u64"))?,
790        );
791        for ranges in &self.layout.logical_witness_participant_ranges {
792            hash_ranges(&mut hasher, ranges)?;
793        }
794
795        hash_u64(
796            &mut hasher,
797            u64::try_from(self.participant_payloads.len()).map_err(|_| {
798                invalid_operation("determinism restore payload participant count exceeds u64")
799            })?,
800        );
801        for logical_index in 0..self.participant_payloads.len() {
802            let physical_index = self
803                .layout
804                .participant_order
805                .physical_index_for_logical(u32::try_from(logical_index).map_err(|_| {
806                    invalid_operation("determinism logical payload index exceeds u32")
807                })?)
808                .and_then(|index| usize::try_from(index).ok())
809                .ok_or_else(|| {
810                    invalid_operation("determinism logical payload lacks a physical participant")
811                })?;
812            let payloads = self
813                .participant_payloads
814                .get(physical_index)
815                .ok_or_else(|| invalid_operation("determinism physical payload is absent"))?;
816            hash_u64(
817                &mut hasher,
818                u64::try_from(payloads.len()).map_err(|_| {
819                    invalid_operation("determinism restore payload count exceeds u64")
820                })?,
821            );
822            for payload in payloads {
823                hash_bytes(&mut hasher, payload)?;
824            }
825        }
826
827        Ok(format!("{:x}", hasher.finalize()))
828    }
829
830    pub(super) fn validate_for(&self, resolved: &dyn ExecutablePlanView) -> Result<(), VNextError> {
831        let actual = resolved
832            .execution_plan()
833            .determinism_witness_plan_for_nodes(self.node_ids())?;
834        if self.plan_hash() != resolved.execution_plan().plan_hash()
835            || self.layout.witness_plan != actual
836        {
837            return Err(invalid_operation(
838                "determinism restore differs from the exact immutable plan initialization denominator",
839            ));
840        }
841        Ok(())
842    }
843
844    pub(super) fn validate_for_submission<'binding, R: DeviceRuntime>(
845        &self,
846        runtime: &R,
847        providers: &[BoundOperationProvider<'_, R>],
848        resolved: &dyn ExecutablePlanView,
849        batch_identity: &BatchOperationIdentity,
850        active_bindings: impl Clone + ExactSizeIterator<Item = &'binding TrustedActiveSequenceBinding>,
851        wave: &PreparedStepSubmissionWave<R>,
852    ) -> Result<(), VNextError> {
853        self.validate_for(resolved)?;
854        let actual_layout =
855            SubmissionWaveDeterminismRestoreLayout::from_prepared_wave_with_participant_order(
856                runtime,
857                providers,
858                resolved,
859                batch_identity,
860                active_bindings,
861                self.layout.participant_order.clone(),
862                wave,
863            )?;
864        if self.layout != actual_layout
865            || self.node_ids().len() != wave.nodes().len()
866            || self.node_ids().len() != batch_identity.node_count()
867            || self.node_ids().iter().zip(wave.nodes()).enumerate().any(
868                |(node_index, (node_id, prepared_node))| {
869                    prepared_node.node_id() != node_id
870                        || batch_identity.node_id_at(node_index) != Some(node_id)
871                },
872            )
873        {
874            return Err(invalid_operation(
875                "determinism restore node scope differs from its exact prepared wave",
876            ));
877        }
878        Ok(())
879    }
880
881    pub(super) fn initialization_range(
882        &self,
883        participant_index: u32,
884        initialization_index: usize,
885    ) -> Option<SubmissionWaveDeterminismLogicalRange> {
886        self.layout
887            .participant_initialization_ranges(participant_index)
888            .and_then(|ranges| ranges.get(initialization_index))
889            .copied()
890    }
891}
892
893fn logical_initialization_ranges<R: DeviceRuntime>(
894    wave: &PreparedStepSubmissionWave<R>,
895    participant_order: &SubmissionWaveDeterminismParticipantOrder,
896    initializations: &[ExecutionDeterminismInitializationSpec],
897    physical_ranges: &[Vec<SubmissionWaveDeterminismLogicalRange>],
898) -> Result<Vec<Vec<SubmissionWaveDeterminismLogicalRange>>, VNextError> {
899    let participant_count = usize::try_from(participant_order.participant_count())
900        .expect("determinism participant count fits usize");
901    if physical_ranges.len() != participant_count
902        || physical_ranges
903            .iter()
904            .any(|ranges| ranges.len() != initializations.len())
905    {
906        return Err(invalid_operation(
907            "determinism initialization range matrix differs from its participant order",
908        ));
909    }
910    let mut logical_ranges = vec![Vec::with_capacity(initializations.len()); participant_count];
911    for (initialization_index, initialization) in initializations.iter().enumerate() {
912        let physical = physical_ranges
913            .iter()
914            .map(|ranges| ranges[initialization_index])
915            .collect::<Vec<_>>();
916        let logical = logical_participant_ranges(
917            wave,
918            participant_order,
919            initialization.location(),
920            &physical,
921        )?;
922        for (participant_ranges, range) in logical_ranges.iter_mut().zip(logical) {
923            participant_ranges.push(range);
924        }
925    }
926    Ok(logical_ranges)
927}
928
929fn logical_participant_ranges<R: DeviceRuntime>(
930    wave: &PreparedStepSubmissionWave<R>,
931    participant_order: &SubmissionWaveDeterminismParticipantOrder,
932    location: &ExecutionDeterminismValueLocation,
933    physical_ranges: &[SubmissionWaveDeterminismLogicalRange],
934) -> Result<Vec<SubmissionWaveDeterminismLogicalRange>, VNextError> {
935    let mut logical_ranges = participant_order.reorder_physical_to_logical(physical_ranges)?;
936    if resource_is_shared(wave, location.resource_id()) {
937        if let ExecutionDeterminismValueExtent::ImmediateTokenSpan {
938            bytes_per_token,
939            maximum_tokens,
940        } = location.extent()
941        {
942            let maximum_bytes = bytes_per_token
943                .checked_mul(maximum_tokens)
944                .ok_or_else(|| invalid_operation("determinism logical packed range overflows"))?;
945            let mut next_offset = 0_u64;
946            for range in &mut logical_ranges {
947                let end = next_offset
948                    .checked_add(range.length_bytes())
949                    .ok_or_else(|| {
950                        invalid_operation("determinism logical participant ranges overflow")
951                    })?;
952                if end > maximum_bytes {
953                    return Err(invalid_operation(
954                        "determinism logical participant ranges exceed shared capacity",
955                    ));
956                }
957                *range = SubmissionWaveDeterminismLogicalRange {
958                    logical_offset_bytes: next_offset,
959                    length_bytes: range.length_bytes(),
960                };
961                next_offset = end;
962            }
963        }
964    }
965    Ok(logical_ranges)
966}
967
968fn resource_is_shared<R: DeviceRuntime>(
969    wave: &PreparedStepSubmissionWave<R>,
970    resource_id: &ResourceId,
971) -> bool {
972    wave.claimed_backing()
973        .backing_slices()
974        .iter()
975        .chain(wave.step_resources().backing_slices())
976        .any(|authority| authority.resource_id() == resource_id)
977}
978
979fn logical_work_fingerprints<R: DeviceRuntime>(
980    wave: &PreparedStepSubmissionWave<R>,
981    participant_order: &SubmissionWaveDeterminismParticipantOrder,
982) -> Result<Vec<String>, VNextError> {
983    wave.nodes()
984        .iter()
985        .map(|node| {
986            let physical_spans = node
987                .work_shape()
988                .participant_work()
989                .iter()
990                .map(|work| work.token_span().clone())
991                .collect::<Vec<_>>();
992            let logical_spans = participant_order.reorder_physical_to_logical(&physical_spans)?;
993            Ok(ResourceWorkShape::from_token_spans(logical_spans)?
994                .fingerprint()
995                .to_owned())
996        })
997        .collect()
998}
999
1000fn prepared_location_range<R: DeviceRuntime>(
1001    wave: &PreparedStepSubmissionWave<R>,
1002    invocation: &BatchedOperationInvocation<'_, R::Buffer>,
1003    participant_index: usize,
1004    location: &ExecutionDeterminismValueLocation,
1005) -> Result<SubmissionWaveDeterminismLogicalRange, VNextError> {
1006    let participant = invocation
1007        .participants()
1008        .get(participant_index)
1009        .ok_or_else(|| {
1010            invalid_operation("determinism location participant is absent from its invocation")
1011        })?;
1012    let component_index = usize::try_from(location.storage_component_ordinal())
1013        .map_err(|_| invalid_operation("determinism component ordinal exceeds usize"))?;
1014    let mut bindings = participant.bindings().iter().filter(|binding| {
1015        let Some(component) = binding.storage().components().get(component_index) else {
1016            return false;
1017        };
1018        binding.value_id() == location.value_id()
1019            && binding.usage() == location.usage()
1020            && component.resource_id() == location.resource_id()
1021            && component.component_id() == location.storage_component_id()
1022            && component.offset_bytes() == location.logical_offset_bytes()
1023            && component.element_type() == location.element_type()
1024    });
1025    let binding = bindings.next().ok_or_else(|| {
1026        invalid_operation(
1027            "determinism semantic value has no exact provider-visible binding component",
1028        )
1029    })?;
1030    if bindings.next().is_some() {
1031        return Err(invalid_operation(
1032            "determinism semantic value maps to multiple provider-visible bindings",
1033        ));
1034    }
1035    let component = &binding.storage().components()[component_index];
1036    let declared_length = if binding.storage().components().len() == 1 {
1037        binding.tensor().minimum_storage_bytes()?
1038    } else {
1039        component.length_bytes()
1040    };
1041    if declared_length != location.declared_length_bytes() {
1042        return Err(invalid_operation(
1043            "determinism provider binding differs from its plan-declared byte range",
1044        ));
1045    }
1046    let mut views = participant
1047        .views()
1048        .iter()
1049        .filter(|view| view.resource_id() == location.resource_id());
1050    let view = views.next().ok_or_else(|| {
1051        invalid_operation("determinism provider binding has no exact operation buffer view")
1052    })?;
1053    if views.next().is_some()
1054        || view.descriptor().usage != location.usage()
1055        || view.descriptor().element_type != location.element_type()
1056    {
1057        return Err(invalid_operation(
1058            "determinism operation buffer view is ambiguous or differs from its typed location",
1059        ));
1060    }
1061
1062    let token_range = invocation
1063        .participant_token_ranges()
1064        .get(participant_index)
1065        .ok_or_else(|| {
1066            invalid_operation("determinism invocation participant token range is missing")
1067        })?;
1068    let participant_work = invocation
1069        .work_shape()
1070        .participant_work()
1071        .get(participant_index)
1072        .ok_or_else(|| invalid_operation("determinism participant work is missing"))?;
1073    if token_range.immediate_tokens() != participant_work.token_span().immediate_tokens()
1074        || token_range.source_token_range() != participant_work.token_span().immediate_token_range()
1075    {
1076        return Err(invalid_operation(
1077            "determinism provider token range differs from its prepared work",
1078        ));
1079    }
1080    let resource_is_shared = resource_is_shared(wave, location.resource_id());
1081
1082    let (logical_offset_bytes, length_bytes) = match location.extent() {
1083        ExecutionDeterminismValueExtent::Fixed => (
1084            location.logical_offset_bytes(),
1085            location.declared_length_bytes(),
1086        ),
1087        ExecutionDeterminismValueExtent::ImmediateTokenSpan {
1088            bytes_per_token,
1089            maximum_tokens,
1090        } => {
1091            let projection = participant
1092                .work()
1093                .token_projection(binding.role(), binding.ordinal())
1094                .ok_or_else(|| {
1095                    invalid_operation(
1096                        "determinism immediate value lacks its provider token projection",
1097                    )
1098                })?;
1099            if binding.usage() != BufferUsage::Activations
1100                || component.offset_bytes() != 0
1101                || component.length_bytes()
1102                    != bytes_per_token
1103                        .checked_mul(projection.canonical_extent())
1104                        .ok_or_else(|| {
1105                            invalid_operation("determinism immediate canonical extent overflows")
1106                        })?
1107            {
1108                return Err(invalid_operation(
1109                    "determinism immediate value differs from its provider token projection",
1110                ));
1111            }
1112            let token_start = if resource_is_shared {
1113                token_range.immediate_token_range().start
1114            } else {
1115                token_range.source_token_range().start
1116            };
1117            immediate_token_logical_range(
1118                token_start,
1119                token_range.immediate_tokens(),
1120                bytes_per_token,
1121                maximum_tokens,
1122            )?
1123        }
1124        ExecutionDeterminismValueExtent::ActiveTokenPrefix {
1125            bytes_per_token,
1126            maximum_tokens,
1127            maximum_storage_bytes,
1128        } => {
1129            let source_end = token_range.source_token_range().end;
1130            let minimum_logical_bytes = bytes_per_token
1131                .checked_mul(source_end)
1132                .ok_or_else(|| invalid_operation("determinism state logical prefix overflows"))?;
1133            if resource_is_shared
1134                || binding.usage() != BufferUsage::State
1135                || component.offset_bytes() != 0
1136                || participant
1137                    .work()
1138                    .token_projection(binding.role(), binding.ordinal())
1139                    .is_some()
1140                || source_end > maximum_tokens
1141                || view.descriptor().size_bytes < minimum_logical_bytes
1142                || view.descriptor().size_bytes > maximum_storage_bytes
1143            {
1144                return Err(invalid_operation(
1145                    "determinism state prefix differs from its participant-local provider view",
1146                ));
1147            }
1148            (0, view.descriptor().size_bytes)
1149        }
1150    };
1151    let element_bytes = location.element_type().size_bytes();
1152    if logical_offset_bytes % element_bytes != 0 || length_bytes % element_bytes != 0 {
1153        return Err(invalid_operation(
1154            "determinism provider-visible range is not element aligned",
1155        ));
1156    }
1157    let translated = view.translate(logical_offset_bytes, length_bytes)?;
1158    let translated_bytes = translated.iter().try_fold(0_u64, |total, region| {
1159        total
1160            .checked_add(region.length_bytes())
1161            .ok_or_else(|| invalid_operation("determinism translated range overflows"))
1162    })?;
1163    if translated_bytes != length_bytes {
1164        return Err(invalid_operation(
1165            "determinism provider-visible range is not fully backed",
1166        ));
1167    }
1168    Ok(SubmissionWaveDeterminismLogicalRange {
1169        logical_offset_bytes,
1170        length_bytes,
1171    })
1172}
1173
1174fn immediate_token_logical_range(
1175    token_start: u64,
1176    immediate_tokens: u64,
1177    bytes_per_token: u64,
1178    maximum_tokens: u64,
1179) -> Result<(u64, u64), VNextError> {
1180    let token_end = token_start
1181        .checked_add(immediate_tokens)
1182        .ok_or_else(|| invalid_operation("determinism immediate token range overflows"))?;
1183    if immediate_tokens == 0 || token_end > maximum_tokens {
1184        return Err(invalid_operation(
1185            "determinism immediate token range exceeds its scheduled resource capacity",
1186        ));
1187    }
1188    Ok((
1189        bytes_per_token
1190            .checked_mul(token_start)
1191            .ok_or_else(|| invalid_operation("determinism immediate byte offset overflows"))?,
1192        bytes_per_token
1193            .checked_mul(immediate_tokens)
1194            .ok_or_else(|| invalid_operation("determinism immediate byte length overflows"))?,
1195    ))
1196}
1197
1198#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
1199struct PhysicalReadbackKey {
1200    node_id: NodeId,
1201    resource_id: ResourceId,
1202    expected_usage: BufferUsage,
1203    participant_layouts: Vec<(u64, ElementType, u64)>,
1204}
1205
1206/// One physical readback group and every semantic witness that projects onto
1207/// that exact range.
1208#[derive(Debug, Clone, PartialEq, Eq)]
1209pub struct SubmissionWaveDeterminismReadbackTarget {
1210    witnesses: Vec<ExecutionDeterminismWitnessSpec>,
1211    batch: CompletionReadbackBatchRequest,
1212}
1213
1214impl SubmissionWaveDeterminismReadbackTarget {
1215    pub fn witnesses(&self) -> &[ExecutionDeterminismWitnessSpec] {
1216        &self.witnesses
1217    }
1218
1219    pub fn batch(&self) -> &CompletionReadbackBatchRequest {
1220        &self.batch
1221    }
1222}
1223
1224/// Canonical terminal readback denominator for one prepared deterministic
1225/// submission wave.
1226#[derive(Debug, Clone, PartialEq, Eq)]
1227#[must_use = "the exact plan-derived witness readback must be collected"]
1228pub struct SubmissionWaveDeterminismReadbackPlan {
1229    plan_hash: PlanHash,
1230    node_ids: Vec<NodeId>,
1231    participant_order: SubmissionWaveDeterminismParticipantOrder,
1232    witnesses: Vec<ExecutionDeterminismWitnessSpec>,
1233    logical_witness_participant_ranges: Vec<Vec<SubmissionWaveDeterminismLogicalRange>>,
1234    collection: CompletionReadbackCollectionRequest,
1235    targets: Vec<SubmissionWaveDeterminismReadbackTarget>,
1236    witness_count: usize,
1237}
1238
1239impl SubmissionWaveDeterminismReadbackPlan {
1240    pub fn from_restore<R: DeviceRuntime>(
1241        resolved: &dyn ExecutablePlanView,
1242        batch_identity: &BatchOperationIdentity,
1243        wave: &PreparedStepSubmissionWave<R>,
1244        restore: &SubmissionWaveDeterminismRestore,
1245    ) -> Result<Self, VNextError> {
1246        let node_ids = wave
1247            .nodes()
1248            .iter()
1249            .map(|node| node.node_id().clone())
1250            .collect::<Vec<_>>();
1251        restore.validate_for(resolved)?;
1252        let witness_plan = restore.layout().witness_plan();
1253        let node_logical_work_fingerprints =
1254            logical_work_fingerprints(wave, restore.layout().participant_order())?;
1255        if witness_plan.plan_hash() != batch_identity.plan_hash()
1256            || restore.layout.batch_invocation_id != batch_identity.batch_invocation_id()
1257            || restore.layout.claimed_backing_fingerprint
1258                != batch_identity.claimed_backing_fingerprint()
1259            || restore.layout.node_logical_work_fingerprints != node_logical_work_fingerprints
1260            || wave.nodes().len() != batch_identity.node_count()
1261            || wave.nodes().iter().enumerate().any(|(node_index, node)| {
1262                node.plan_evidence_ref().plan_hash() != witness_plan.plan_hash()
1263                    || batch_identity.node_id_at(node_index) != Some(node.node_id())
1264                    || batch_identity.node_participant_count(node_index)
1265                        != Some(node.participant_count() as usize)
1266            })
1267        {
1268            return Err(invalid_operation(
1269                "determinism readback plan differs from its prepared wave or physical batch identity",
1270            ));
1271        }
1272        validate_terminal_witness_stability(resolved, &witness_plan)?;
1273
1274        let mut grouped = BTreeMap::<
1275            PhysicalReadbackKey,
1276            (
1277                CompletionReadbackBatchRequest,
1278                Vec<ExecutionDeterminismWitnessSpec>,
1279            ),
1280        >::new();
1281
1282        for (witness_index, witness) in witness_plan.witnesses().iter().enumerate() {
1283            let node_index = batch_identity
1284                .node_index(witness.node_id())
1285                .ok_or_else(|| {
1286                    invalid_operation(
1287                        "determinism witness node is absent from its physical batch identity",
1288                    )
1289                })?;
1290            let node = wave.nodes().get(node_index).ok_or_else(|| {
1291                invalid_operation("determinism witness node is absent from its prepared wave")
1292            })?;
1293            if node.node_id() != witness.node_id() {
1294                return Err(invalid_operation(
1295                    "determinism witness node index differs from its prepared wave",
1296                ));
1297            }
1298
1299            let ranges = restore
1300                .layout()
1301                .witness_participant_ranges(witness_index)
1302                .ok_or_else(|| {
1303                    invalid_operation("determinism witness lacks its prepared participant ranges")
1304                })?;
1305            let element_bytes = witness.element_type().size_bytes();
1306            let requests = ranges
1307                .iter()
1308                .enumerate()
1309                .map(|(participant_index, range)| {
1310                    if range.length_bytes() % element_bytes != 0 {
1311                        return Err(invalid_operation(
1312                            "determinism witness provider-visible range is not element aligned",
1313                        ));
1314                    }
1315                    let completion_range = completion_witness_readback_range(
1316                        wave,
1317                        node,
1318                        participant_index,
1319                        witness,
1320                        *range,
1321                    )?;
1322                    CompletionReadbackRequest::new_typed(
1323                        witness.node_id().clone(),
1324                        u32::try_from(participant_index).map_err(|_| {
1325                            invalid_operation("determinism readback participant index exceeds u32")
1326                        })?,
1327                        witness.resource_id().clone(),
1328                        witness.location().usage(),
1329                        completion_range.logical_offset_bytes(),
1330                        HostTransferLayout::new(
1331                            witness.element_type(),
1332                            completion_range.length_bytes() / element_bytes,
1333                        )?,
1334                    )
1335                })
1336                .collect::<Result<Vec<_>, VNextError>>()?;
1337            let batch = CompletionReadbackBatchRequest::new(requests)?;
1338            let first = batch
1339                .requests()
1340                .first()
1341                .expect("determinism readback batches are non-empty");
1342            let key = PhysicalReadbackKey {
1343                node_id: first.node_id().clone(),
1344                resource_id: first.resource_id().clone(),
1345                expected_usage: first.expected_usage(),
1346                participant_layouts: batch
1347                    .requests()
1348                    .iter()
1349                    .map(|request| {
1350                        (
1351                            request.logical_offset_bytes(),
1352                            request.output_layout().element_type(),
1353                            request.output_layout().element_count(),
1354                        )
1355                    })
1356                    .collect(),
1357            };
1358            match grouped.entry(key) {
1359                std::collections::btree_map::Entry::Vacant(entry) => {
1360                    entry.insert((batch, vec![witness.clone()]));
1361                }
1362                std::collections::btree_map::Entry::Occupied(mut entry) => {
1363                    entry.get_mut().1.push(witness.clone());
1364                }
1365            }
1366        }
1367
1368        let targets = grouped
1369            .into_values()
1370            .map(|(batch, witnesses)| SubmissionWaveDeterminismReadbackTarget { witnesses, batch })
1371            .collect::<Vec<_>>();
1372        let collection = CompletionReadbackCollectionRequest::new(
1373            targets.iter().map(|target| target.batch.clone()).collect(),
1374        )?;
1375        let witness_count = targets.iter().map(|target| target.witnesses.len()).sum();
1376        if witness_count != witness_plan.witnesses().len()
1377            || collection
1378                .batches()
1379                .iter()
1380                .zip(&targets)
1381                .any(|(batch, target)| batch != &target.batch)
1382        {
1383            return Err(invalid_operation(
1384                "determinism readback canonicalization lost a semantic witness mapping",
1385            ));
1386        }
1387
1388        Ok(Self {
1389            plan_hash: witness_plan.plan_hash().clone(),
1390            node_ids,
1391            participant_order: restore.layout.participant_order.clone(),
1392            witnesses: witness_plan.witnesses().to_vec(),
1393            logical_witness_participant_ranges: restore
1394                .layout
1395                .logical_witness_participant_ranges
1396                .clone(),
1397            collection,
1398            targets,
1399            witness_count,
1400        })
1401    }
1402
1403    pub fn plan_hash(&self) -> &PlanHash {
1404        &self.plan_hash
1405    }
1406
1407    pub fn node_ids(&self) -> &[NodeId] {
1408        &self.node_ids
1409    }
1410
1411    pub fn collection_request(&self) -> &CompletionReadbackCollectionRequest {
1412        &self.collection
1413    }
1414
1415    pub fn targets(&self) -> &[SubmissionWaveDeterminismReadbackTarget] {
1416        &self.targets
1417    }
1418
1419    pub const fn witness_count(&self) -> usize {
1420        self.witness_count
1421    }
1422
1423    fn logical_witness_range(
1424        &self,
1425        witness: &ExecutionDeterminismWitnessSpec,
1426        logical_participant_index: u32,
1427    ) -> Result<SubmissionWaveDeterminismLogicalRange, VNextError> {
1428        let mut matches = self
1429            .witnesses
1430            .iter()
1431            .enumerate()
1432            .filter(|(_, candidate)| *candidate == witness);
1433        let (witness_index, _) = matches.next().ok_or_else(|| {
1434            invalid_operation("determinism readback witness is absent from its semantic plan")
1435        })?;
1436        if matches.next().is_some() {
1437            return Err(invalid_operation(
1438                "determinism readback witness is ambiguous in its semantic plan",
1439            ));
1440        }
1441        self.logical_witness_participant_ranges
1442            .get(witness_index)
1443            .and_then(|ranges| {
1444                usize::try_from(logical_participant_index)
1445                    .ok()
1446                    .and_then(|index| ranges.get(index))
1447            })
1448            .copied()
1449            .ok_or_else(|| {
1450                invalid_operation(
1451                    "determinism logical witness range is absent from its participant order",
1452                )
1453            })
1454    }
1455}
1456
1457fn completion_witness_readback_range<R: DeviceRuntime>(
1458    wave: &PreparedStepSubmissionWave<R>,
1459    node: &crate::vnext::PreparedStepSubmissionNode<R>,
1460    participant_index: usize,
1461    witness: &ExecutionDeterminismWitnessSpec,
1462    provider_range: SubmissionWaveDeterminismLogicalRange,
1463) -> Result<SubmissionWaveDeterminismLogicalRange, VNextError> {
1464    let ExecutionDeterminismValueExtent::ImmediateTokenSpan {
1465        bytes_per_token,
1466        maximum_tokens,
1467    } = witness.location().extent()
1468    else {
1469        return Ok(provider_range);
1470    };
1471    if !resource_is_shared(wave, witness.resource_id()) {
1472        return Ok(provider_range);
1473    }
1474
1475    let descriptor = wave
1476        .step_resources()
1477        .dynamic_descriptor(witness.resource_id())?;
1478    if descriptor.lifetime() != AllocationLifetime::Step
1479        || descriptor.kind() != &AllocationKind::Value
1480        || !matches!(
1481            descriptor.demand(),
1482            crate::vnext::DynamicResourceDemand::Tokens {
1483                bytes_per_token: descriptor_bytes_per_token,
1484                maximum_tokens: descriptor_maximum_tokens,
1485            } if *descriptor_bytes_per_token == bytes_per_token
1486                && *descriptor_maximum_tokens == maximum_tokens
1487        )
1488    {
1489        return Err(invalid_operation(
1490            "shared immediate determinism witness lacks matching Step token backing",
1491        ));
1492    }
1493    let provider_end = provider_range
1494        .logical_offset_bytes()
1495        .checked_add(provider_range.length_bytes())
1496        .ok_or_else(|| invalid_operation("determinism provider readback range overflows"))?;
1497    let participant_local = packed_step_token_range_to_participant_local_readback(
1498        descriptor.demand(),
1499        node.work_shape(),
1500        participant_index,
1501        provider_range.logical_offset_bytes()..provider_end,
1502    )?;
1503    Ok(SubmissionWaveDeterminismLogicalRange {
1504        logical_offset_bytes: participant_local.start,
1505        length_bytes: participant_local.end - participant_local.start,
1506    })
1507}
1508
1509fn validate_terminal_witness_stability(
1510    resolved: &dyn ExecutablePlanView,
1511    witness_plan: &ExecutionDeterminismWitnessPlan,
1512) -> Result<(), VNextError> {
1513    if witness_plan.node_ids().len() > 1 {
1514        let retained = resolved
1515            .execution_plan()
1516            .payload()
1517            .retained_completion_values();
1518        for witness in witness_plan.witnesses() {
1519            let ExecutionDeterminismWitnessKind::Output {
1520                value_id,
1521                output_ordinal,
1522            } = witness.kind()
1523            else {
1524                continue;
1525            };
1526            let exact = retained.iter().filter(|candidate| {
1527                candidate.value_id() == value_id
1528                    && candidate.producer_node_id() == witness.node_id()
1529                    && candidate.output_ordinal() == *output_ordinal
1530                    && candidate.resource_id() == witness.resource_id()
1531                    && candidate.logical_offset_bytes() == witness.logical_offset_bytes()
1532                    && candidate.tensor().element_type() == witness.element_type()
1533                    && candidate.tensor().minimum_storage_bytes().ok()
1534                        == Some(witness.declared_length_bytes())
1535            });
1536            if exact.count() != 1 {
1537                return Err(invalid_operation(format!(
1538                    "determinism output `{value_id}` from node `{}` is not retained as one exact terminal witness",
1539                    witness.node_id()
1540                )));
1541            }
1542        }
1543    }
1544
1545    let node_order = witness_plan
1546        .node_ids()
1547        .iter()
1548        .enumerate()
1549        .map(|(index, node_id)| (node_id, index))
1550        .collect::<BTreeMap<_, _>>();
1551    for witness in witness_plan.witnesses() {
1552        if !matches!(
1553            witness.kind(),
1554            ExecutionDeterminismWitnessKind::StateEffect { .. }
1555        ) {
1556            continue;
1557        }
1558        let witness_order = node_order[witness.node_id()];
1559        let witness_end = witness
1560            .logical_offset_bytes()
1561            .checked_add(witness.maximum_bound_length_bytes()?)
1562            .ok_or_else(|| invalid_operation("determinism state witness range overflows"))?;
1563        let overwritten = witness_plan.witnesses().iter().any(|later| {
1564            let Some(later_order) = node_order.get(later.node_id()) else {
1565                return false;
1566            };
1567            if *later_order <= witness_order || later.resource_id() != witness.resource_id() {
1568                return false;
1569            }
1570            let Some(later_end) = later
1571                .logical_offset_bytes()
1572                .checked_add(later.maximum_bound_length_bytes().unwrap_or(u64::MAX))
1573            else {
1574                return true;
1575            };
1576            later.logical_offset_bytes() < witness_end && witness.logical_offset_bytes() < later_end
1577        });
1578        if overwritten {
1579            return Err(invalid_operation(format!(
1580                "determinism state witness from node `{}` is overwritten later in the same terminal readback scope",
1581                witness.node_id()
1582            )));
1583        }
1584    }
1585    Ok(())
1586}
1587
1588/// One physical readback buffer retained once even when multiple semantic
1589/// witnesses intentionally project onto the same bytes.
1590#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
1591pub struct SubmissionWaveDeterminismPhysicalReadback {
1592    request: CompletionReadbackRequest,
1593    raw_sha256: String,
1594    #[serde(skip)]
1595    bytes: Vec<u8>,
1596}
1597
1598impl SubmissionWaveDeterminismPhysicalReadback {
1599    pub fn request(&self) -> &CompletionReadbackRequest {
1600        &self.request
1601    }
1602
1603    pub fn raw_sha256(&self) -> &str {
1604        &self.raw_sha256
1605    }
1606
1607    pub fn bytes(&self) -> &[u8] {
1608        &self.bytes
1609    }
1610}
1611
1612/// Semantic witness mapped to exactly one physical participant readback.
1613#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
1614pub struct SubmissionWaveDeterminismWitnessReadback {
1615    witness: ExecutionDeterminismWitnessSpec,
1616    participant_index: u32,
1617    #[serde(skip)]
1618    physical_participant_index: u32,
1619    #[serde(skip)]
1620    logical_range: SubmissionWaveDeterminismLogicalRange,
1621    physical_readback_index: u32,
1622}
1623
1624impl SubmissionWaveDeterminismWitnessReadback {
1625    pub fn witness(&self) -> &ExecutionDeterminismWitnessSpec {
1626        &self.witness
1627    }
1628
1629    pub const fn participant_index(&self) -> u32 {
1630        self.participant_index
1631    }
1632
1633    pub const fn physical_participant_index(&self) -> u32 {
1634        self.physical_participant_index
1635    }
1636
1637    pub const fn logical_range(&self) -> SubmissionWaveDeterminismLogicalRange {
1638        self.logical_range
1639    }
1640
1641    pub const fn physical_readback_index(&self) -> u32 {
1642        self.physical_readback_index
1643    }
1644}
1645
1646#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
1647#[serde(rename_all = "snake_case")]
1648enum SubmissionWaveDeterminismComputeExpectation {
1649    EagerOnly,
1650    Replayed {
1651        program_id: DeviceReusableExecutionProgramId,
1652        declared_eager_boundary_node_ids: Vec<NodeId>,
1653    },
1654}
1655
1656impl SubmissionWaveDeterminismComputeExpectation {
1657    fn replayed(
1658        program_id: DeviceReusableExecutionProgramId,
1659        declared_eager_boundary_node_ids: Vec<NodeId>,
1660    ) -> Result<Self, VNextError> {
1661        if declared_eager_boundary_node_ids
1662            .windows(2)
1663            .any(|pair| pair[0] >= pair[1])
1664        {
1665            return Err(invalid_operation(
1666                "determinism eager boundary node ids are not canonical",
1667            ));
1668        }
1669        Ok(Self::Replayed {
1670            program_id,
1671            declared_eager_boundary_node_ids,
1672        })
1673    }
1674
1675    const fn execution_path(&self) -> DeviceExecutionPath {
1676        match self {
1677            Self::EagerOnly => DeviceExecutionPath::Eager,
1678            Self::Replayed { .. } => DeviceExecutionPath::Replayed,
1679        }
1680    }
1681
1682    const fn compute_path_requirement(&self) -> DeviceComputePathRequirement {
1683        match self {
1684            Self::EagerOnly => DeviceComputePathRequirement::EagerOnly,
1685            Self::Replayed {
1686                declared_eager_boundary_node_ids,
1687                ..
1688            } if declared_eager_boundary_node_ids.is_empty() => {
1689                DeviceComputePathRequirement::ReplayedOnly
1690            }
1691            Self::Replayed { .. } => {
1692                DeviceComputePathRequirement::ReplayedWithDeclaredEagerBoundaries
1693            }
1694        }
1695    }
1696
1697    fn reusable_program_id(&self) -> Option<&DeviceReusableExecutionProgramId> {
1698        match self {
1699            Self::EagerOnly => None,
1700            Self::Replayed { program_id, .. } => Some(program_id),
1701        }
1702    }
1703
1704    fn declared_eager_boundary_node_ids(&self) -> &[NodeId] {
1705        match self {
1706            Self::EagerOnly => &[],
1707            Self::Replayed {
1708                declared_eager_boundary_node_ids,
1709                ..
1710            } => declared_eager_boundary_node_ids,
1711        }
1712    }
1713}
1714
1715/// Fail-closed terminal evidence for one forced eager or replay execution.
1716///
1717/// The physical buffers retain raw bytes for in-process comparison. Their
1718/// serialized form contains only SHA256 digests so a detached artifact cannot
1719/// accidentally duplicate large device outputs.
1720#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
1721pub struct SubmissionWaveDeterminismEvidence {
1722    restore_fingerprint: String,
1723    initialization_identity: SubmissionWaveDeterminismInitializationIdentity,
1724    compute_expectation: SubmissionWaveDeterminismComputeExpectation,
1725    submission_receipt_fingerprint: String,
1726    terminal_receipt_fingerprint: String,
1727    attribution: BoundDeviceSubmissionAttribution,
1728    physical_readbacks: Vec<SubmissionWaveDeterminismPhysicalReadback>,
1729    witnesses: Vec<SubmissionWaveDeterminismWitnessReadback>,
1730}
1731
1732impl SubmissionWaveDeterminismEvidence {
1733    pub fn restore_fingerprint(&self) -> &str {
1734        &self.restore_fingerprint
1735    }
1736
1737    pub fn initialization_identity(&self) -> &SubmissionWaveDeterminismInitializationIdentity {
1738        &self.initialization_identity
1739    }
1740
1741    pub const fn expected_execution_path(&self) -> DeviceExecutionPath {
1742        self.compute_expectation.execution_path()
1743    }
1744
1745    pub const fn expected_compute_path_requirement(&self) -> DeviceComputePathRequirement {
1746        self.compute_expectation.compute_path_requirement()
1747    }
1748
1749    pub fn reusable_program_fingerprint(&self) -> Option<String> {
1750        self.compute_expectation
1751            .reusable_program_id()
1752            .map(DeviceReusableExecutionProgramId::fingerprint)
1753    }
1754
1755    pub fn declared_eager_boundary_node_ids(&self) -> &[NodeId] {
1756        self.compute_expectation.declared_eager_boundary_node_ids()
1757    }
1758
1759    pub fn submission_receipt_fingerprint(&self) -> &str {
1760        &self.submission_receipt_fingerprint
1761    }
1762
1763    pub fn terminal_receipt_fingerprint(&self) -> &str {
1764        &self.terminal_receipt_fingerprint
1765    }
1766
1767    pub fn attribution(&self) -> &BoundDeviceSubmissionAttribution {
1768        &self.attribution
1769    }
1770
1771    pub fn physical_readbacks(&self) -> &[SubmissionWaveDeterminismPhysicalReadback] {
1772        &self.physical_readbacks
1773    }
1774
1775    pub fn witnesses(&self) -> &[SubmissionWaveDeterminismWitnessReadback] {
1776        &self.witnesses
1777    }
1778}
1779
1780fn validate_determinism_attribution(
1781    attribution: &BoundDeviceSubmissionAttribution,
1782    node_ids: &[NodeId],
1783    compute_expectation: &SubmissionWaveDeterminismComputeExpectation,
1784) -> Result<(), VNextError> {
1785    let expected_nodes = node_ids.iter().collect::<BTreeSet<_>>();
1786    let declared_eager_boundary_nodes = compute_expectation
1787        .declared_eager_boundary_node_ids()
1788        .iter()
1789        .collect::<BTreeSet<_>>();
1790    if !declared_eager_boundary_nodes.is_subset(&expected_nodes) {
1791        return Err(invalid_operation(
1792            "determinism eager boundary is absent from the requested plan nodes",
1793        ));
1794    }
1795    let mut observed_nodes = BTreeSet::new();
1796    match compute_expectation.execution_path() {
1797        DeviceExecutionPath::Eager => {
1798            for replayed_segment in attribution.device().replayed_segments() {
1799                for command in replayed_segment.logical_commands() {
1800                    let node_index = usize::try_from(command.node_index()).map_err(|_| {
1801                        invalid_operation("determinism replay node index exceeds usize")
1802                    })?;
1803                    let node_id = attribution
1804                        .batch_identity()
1805                        .node_id_at(node_index)
1806                        .ok_or_else(|| {
1807                            invalid_operation(
1808                                "determinism replay attribution references a node absent from its batch",
1809                            )
1810                        })?;
1811                    if expected_nodes.contains(node_id) {
1812                        return Err(invalid_operation(format!(
1813                            "determinism node `{node_id}` replayed while eager execution was required"
1814                        )));
1815                    }
1816                }
1817            }
1818            for command in attribution.device().commands() {
1819                if command.command_phase() != DeviceCommandPhase::Compute {
1820                    continue;
1821                }
1822                let Some(node_index) = command.node_index() else {
1823                    continue;
1824                };
1825                let node_index = usize::try_from(node_index).map_err(|_| {
1826                    invalid_operation("determinism attribution node index exceeds usize")
1827                })?;
1828                let node_id = attribution
1829                    .batch_identity()
1830                    .node_id_at(node_index)
1831                    .ok_or_else(|| {
1832                        invalid_operation(
1833                            "determinism attribution references a node absent from its batch",
1834                        )
1835                    })?;
1836                if !expected_nodes.contains(node_id) {
1837                    continue;
1838                }
1839                if command.execution_path() != DeviceExecutionPath::Eager
1840                    || command.reusable_graph_node_count().is_some()
1841                {
1842                    return Err(invalid_operation(format!(
1843                        "determinism node `{node_id}` did not execute through the required eager path"
1844                    )));
1845                }
1846                if !observed_nodes.insert(node_id) {
1847                    return Err(invalid_operation(format!(
1848                        "determinism node `{node_id}` has duplicate eager compute attribution"
1849                    )));
1850                }
1851            }
1852        }
1853        DeviceExecutionPath::Replayed => {
1854            let expected_program_id =
1855                compute_expectation.reusable_program_id().ok_or_else(|| {
1856                    invalid_operation("determinism replay expectation lacks a reusable program")
1857                })?;
1858            if attribution.device().replayed_segments().is_empty() {
1859                return Err(invalid_operation(
1860                    "determinism replay attribution contains no resident segment",
1861                ));
1862            }
1863            for command in attribution.device().commands() {
1864                if command.command_phase() != DeviceCommandPhase::Compute {
1865                    continue;
1866                }
1867                let Some(node_index) = command.node_index() else {
1868                    continue;
1869                };
1870                let node_index = usize::try_from(node_index).map_err(|_| {
1871                    invalid_operation("determinism attribution node index exceeds usize")
1872                })?;
1873                let node_id = attribution
1874                    .batch_identity()
1875                    .node_id_at(node_index)
1876                    .ok_or_else(|| {
1877                        invalid_operation(
1878                            "determinism attribution references a node absent from its batch",
1879                        )
1880                    })?;
1881                if !expected_nodes.contains(node_id) {
1882                    continue;
1883                }
1884                if declared_eager_boundary_nodes.contains(node_id) {
1885                    if command.execution_path() != DeviceExecutionPath::Eager
1886                        || command.reusable_graph_node_count().is_some()
1887                    {
1888                        return Err(invalid_operation(format!(
1889                            "determinism declared eager boundary `{node_id}` did not execute eagerly"
1890                        )));
1891                    }
1892                    if !observed_nodes.insert(node_id) {
1893                        return Err(invalid_operation(format!(
1894                            "determinism declared eager boundary `{node_id}` has duplicate compute attribution"
1895                        )));
1896                    }
1897                } else if command.execution_path() != DeviceExecutionPath::Replayed {
1898                    return Err(invalid_operation(format!(
1899                        "determinism node `{node_id}` executed eagerly without a declared topology boundary"
1900                    )));
1901                }
1902            }
1903            for replayed_segment in attribution.device().replayed_segments() {
1904                if replayed_segment.program_id() != expected_program_id {
1905                    return Err(invalid_operation(
1906                        "determinism replay attribution references another reusable program",
1907                    ));
1908                }
1909                for command in replayed_segment.logical_commands() {
1910                    let node_index = usize::try_from(command.node_index()).map_err(|_| {
1911                        invalid_operation("determinism replay node index exceeds usize")
1912                    })?;
1913                    let node_id = attribution
1914                        .batch_identity()
1915                        .node_id_at(node_index)
1916                        .ok_or_else(|| {
1917                            invalid_operation(
1918                                "determinism replay attribution references a node absent from its batch",
1919                            )
1920                        })?;
1921                    if declared_eager_boundary_nodes.contains(node_id) {
1922                        return Err(invalid_operation(format!(
1923                            "determinism declared eager boundary `{node_id}` appeared in replay attribution"
1924                        )));
1925                    }
1926                    if expected_nodes.contains(node_id) && !observed_nodes.insert(node_id) {
1927                        return Err(invalid_operation(format!(
1928                            "determinism node `{node_id}` has duplicate replay attribution"
1929                        )));
1930                    }
1931                }
1932            }
1933        }
1934    }
1935    if observed_nodes != expected_nodes {
1936        return Err(invalid_operation(
1937            "determinism attribution does not cover every requested plan node",
1938        ));
1939    }
1940    Ok(())
1941}
1942
1943/// Submitted deterministic work whose terminal observation remains bound to
1944/// the exact immutable-plan witness denominator.
1945#[must_use = "deterministic submission must collect its exact witness readback"]
1946pub struct SubmissionWaveDeterminismHandle<R: DeviceRuntime> {
1947    completion: CompletionHandle<R>,
1948    attribution: Option<BoundDeviceSubmissionAttribution>,
1949    readback_plan: SubmissionWaveDeterminismReadbackPlan,
1950    restore_fingerprint: String,
1951    initialization_identity: SubmissionWaveDeterminismInitializationIdentity,
1952    compute_expectation: SubmissionWaveDeterminismComputeExpectation,
1953}
1954
1955impl<R: DeviceRuntime> SubmissionWaveDeterminismHandle<R> {
1956    pub(super) fn from_profiled_eager(
1957        profiled: ProfiledSubmissionHandle<R>,
1958        readback_plan: SubmissionWaveDeterminismReadbackPlan,
1959        restore_fingerprint: String,
1960        initialization_identity: SubmissionWaveDeterminismInitializationIdentity,
1961    ) -> Self {
1962        Self::from_profiled(
1963            profiled,
1964            readback_plan,
1965            restore_fingerprint,
1966            initialization_identity,
1967            SubmissionWaveDeterminismComputeExpectation::EagerOnly,
1968        )
1969    }
1970
1971    pub(super) fn from_profiled_replayed(
1972        profiled: ProfiledSubmissionHandle<R>,
1973        readback_plan: SubmissionWaveDeterminismReadbackPlan,
1974        restore_fingerprint: String,
1975        initialization_identity: SubmissionWaveDeterminismInitializationIdentity,
1976        program_id: DeviceReusableExecutionProgramId,
1977        declared_eager_boundary_node_ids: Vec<NodeId>,
1978    ) -> Result<Self, VNextError> {
1979        let expectation = SubmissionWaveDeterminismComputeExpectation::replayed(
1980            program_id,
1981            declared_eager_boundary_node_ids,
1982        )?;
1983        Ok(Self::from_profiled(
1984            profiled,
1985            readback_plan,
1986            restore_fingerprint,
1987            initialization_identity,
1988            expectation,
1989        ))
1990    }
1991
1992    fn from_profiled(
1993        profiled: ProfiledSubmissionHandle<R>,
1994        readback_plan: SubmissionWaveDeterminismReadbackPlan,
1995        restore_fingerprint: String,
1996        initialization_identity: SubmissionWaveDeterminismInitializationIdentity,
1997        compute_expectation: SubmissionWaveDeterminismComputeExpectation,
1998    ) -> Self {
1999        let (completion, attribution) = profiled.into_parts();
2000        Self {
2001            completion,
2002            attribution,
2003            readback_plan,
2004            restore_fingerprint,
2005            initialization_identity,
2006            compute_expectation,
2007        }
2008    }
2009
2010    pub fn receipt(&self) -> &SubmittedOperationReceipt {
2011        self.completion.receipt()
2012    }
2013
2014    pub fn attribution(&self) -> Option<&BoundDeviceSubmissionAttribution> {
2015        self.attribution.as_ref()
2016    }
2017
2018    pub fn readback_plan(&self) -> &SubmissionWaveDeterminismReadbackPlan {
2019        &self.readback_plan
2020    }
2021
2022    pub fn restore_fingerprint(&self) -> &str {
2023        &self.restore_fingerprint
2024    }
2025
2026    pub fn wait_with_determinism_readback(
2027        &self,
2028    ) -> Result<CompletionReadbackCollectionObservation, VNextError> {
2029        self.completion
2030            .wait_with_readback_collection(self.readback_plan.collection_request().clone())
2031    }
2032
2033    /// Waits for terminal completion and consumes the handle into one typed
2034    /// evidence object. Missing actual-path attribution, nonterminal
2035    /// completion, failed readback, or any semantic/physical mapping drift is
2036    /// rejected after the submitted work has been observed to completion.
2037    pub fn wait_into_evidence(self) -> Result<SubmissionWaveDeterminismEvidence, VNextError> {
2038        let Self {
2039            completion,
2040            attribution,
2041            readback_plan,
2042            restore_fingerprint,
2043            initialization_identity,
2044            compute_expectation,
2045        } = self;
2046        let submission_receipt_fingerprint = completion.receipt().fingerprint().to_owned();
2047        let observation =
2048            completion.wait_with_readback_collection(readback_plan.collection_request().clone())?;
2049        let receipt = match observation {
2050            CompletionReadbackCollectionObservation::Terminal(receipt) => receipt,
2051            other => {
2052                return Err(invalid_operation(format!(
2053                    "determinism readback did not reach a terminal observation: {other:?}"
2054                )))
2055            }
2056        };
2057        if !matches!(
2058            receipt.completion().disposition(),
2059            OperationCompletionDisposition::Succeeded
2060        ) {
2061            return Err(invalid_operation(
2062                "determinism submission completed without a successful disposition",
2063            ));
2064        }
2065
2066        let attribution = attribution
2067            .ok_or_else(|| {
2068                invalid_operation("determinism submission lacks actual-path device attribution")
2069            })?
2070            .bind_terminal_timing(receipt.completion().submission_timing().clone())?;
2071        validate_determinism_attribution(
2072            &attribution,
2073            readback_plan.node_ids(),
2074            &compute_expectation,
2075        )?;
2076
2077        let expected_readbacks = readback_plan
2078            .targets()
2079            .iter()
2080            .map(|target| target.batch().requests().len())
2081            .sum::<usize>();
2082        if receipt.dispositions().len() != expected_readbacks {
2083            return Err(invalid_operation(
2084                "determinism terminal readback count differs from its typed plan",
2085            ));
2086        }
2087
2088        let mut physical_readbacks = Vec::with_capacity(expected_readbacks);
2089        let mut witnesses = Vec::with_capacity(
2090            readback_plan.witness_count().saturating_mul(
2091                readback_plan
2092                    .targets()
2093                    .first()
2094                    .map_or(0, |target| target.batch().requests().len()),
2095            ),
2096        );
2097        let mut disposition_index = 0_usize;
2098        for target in readback_plan.targets() {
2099            let first_physical_index = physical_readbacks.len();
2100            for expected_request in target.batch().requests() {
2101                let disposition =
2102                    receipt
2103                        .dispositions()
2104                        .get(disposition_index)
2105                        .ok_or_else(|| {
2106                            invalid_operation("determinism terminal readback disappeared")
2107                        })?;
2108                let CompletionReadbackDisposition::Succeeded(output) = disposition else {
2109                    return Err(invalid_operation(format!(
2110                        "determinism terminal readback failed: {disposition:?}"
2111                    )));
2112                };
2113                if output.request() != expected_request {
2114                    return Err(invalid_operation(
2115                        "determinism terminal readback differs from its exact request",
2116                    ));
2117                }
2118                physical_readbacks.push(SubmissionWaveDeterminismPhysicalReadback {
2119                    request: output.request().clone(),
2120                    raw_sha256: output.sha256().to_owned(),
2121                    bytes: output.bytes().to_vec(),
2122                });
2123                disposition_index += 1;
2124            }
2125            for witness in target.witnesses() {
2126                for physical_participant_index in 0..target.batch().requests().len() {
2127                    let physical_readback_index = first_physical_index
2128                        .checked_add(physical_participant_index)
2129                        .and_then(|index| u32::try_from(index).ok())
2130                        .ok_or_else(|| {
2131                            invalid_operation("determinism physical readback index exceeds u32")
2132                        })?;
2133                    let physical_participant_index = u32::try_from(physical_participant_index)
2134                        .map_err(|_| {
2135                            invalid_operation(
2136                                "determinism physical witness participant index exceeds u32",
2137                            )
2138                        })?;
2139                    let participant_index = readback_plan
2140                        .participant_order
2141                        .logical_index_for_physical(physical_participant_index)
2142                        .ok_or_else(|| {
2143                            invalid_operation(
2144                                "determinism physical witness lacks a logical participant",
2145                            )
2146                        })?;
2147                    let logical_range =
2148                        readback_plan.logical_witness_range(witness, participant_index)?;
2149                    let physical_request = target
2150                        .batch()
2151                        .requests()
2152                        .get(
2153                            usize::try_from(physical_participant_index)
2154                                .expect("u32 physical participant fits usize"),
2155                        )
2156                        .ok_or_else(|| {
2157                            invalid_operation("determinism physical witness request disappeared")
2158                        })?;
2159                    if physical_request.output_layout().byte_len()? != logical_range.length_bytes()
2160                    {
2161                        return Err(invalid_operation(
2162                            "determinism logical witness length differs from its physical readback",
2163                        ));
2164                    }
2165                    witnesses.push(SubmissionWaveDeterminismWitnessReadback {
2166                        witness: witness.clone(),
2167                        participant_index,
2168                        physical_participant_index,
2169                        logical_range,
2170                        physical_readback_index,
2171                    });
2172                }
2173            }
2174        }
2175        if disposition_index != receipt.dispositions().len() {
2176            return Err(invalid_operation(
2177                "determinism terminal readback left unowned physical outputs",
2178            ));
2179        }
2180        let terminal_receipt_fingerprint = receipt.fingerprint().to_owned();
2181        Ok(SubmissionWaveDeterminismEvidence {
2182            restore_fingerprint,
2183            initialization_identity,
2184            compute_expectation,
2185            submission_receipt_fingerprint,
2186            terminal_receipt_fingerprint,
2187            attribution,
2188            physical_readbacks,
2189            witnesses,
2190        })
2191    }
2192}
2193
2194#[cfg(test)]
2195mod tests {
2196    use super::{immediate_token_logical_range, validate_participant_permutation};
2197
2198    #[test]
2199    fn participant_order_requires_an_exact_bijection() {
2200        assert_eq!(
2201            validate_participant_permutation(&[2, 0, 1]).unwrap(),
2202            [1, 2, 0]
2203        );
2204        assert!(validate_participant_permutation(&[]).is_err());
2205        assert!(validate_participant_permutation(&[0, 0]).is_err());
2206        assert!(validate_participant_permutation(&[0, 2]).is_err());
2207    }
2208
2209    #[test]
2210    fn immediate_token_range_uses_scheduled_capacity_not_canonical_extent() {
2211        assert_eq!(
2212            immediate_token_logical_range(1, 3, 16, 4).unwrap(),
2213            (16, 48)
2214        );
2215        assert!(immediate_token_logical_range(2, 3, 16, 4).is_err());
2216        assert!(immediate_token_logical_range(u64::MAX, 1, 16, u64::MAX).is_err());
2217    }
2218}