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#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
67pub struct SubmissionWaveDeterminismLogicalRange {
68 logical_offset_bytes: u64,
69 length_bytes: u64,
70}
71
72#[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 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#[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#[derive(Debug, Clone, PartialEq, Eq)]
589pub struct SubmissionWaveDeterminismRestore {
590 layout: SubmissionWaveDeterminismRestoreLayout,
591 participant_payloads: Vec<Vec<Vec<u8>>>,
592}
593
594#[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 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 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#[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#[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#[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#[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#[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#[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 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}