eredu_runtime/
realtime_payload.rs

1//! Backend-neutral payload retention for delayed realtime coordinates.
2
3use std::{
4    collections::{BTreeMap, BTreeSet},
5    num::NonZeroUsize,
6};
7
8use eredu_core::{RealtimeFrameSlot, RealtimeSlotCoordinate, RealtimeSpeechConfig};
9
10use crate::generation::TokenDomain;
11
12/// Process-local identity of the canonical session state owning payload tensors.
13#[derive(Debug, Clone, Copy, Eq, Hash, Ord, PartialEq, PartialOrd)]
14pub struct RealtimePayloadOwnerIdentity(u64);
15
16impl RealtimePayloadOwnerIdentity {
17    /// Creates a nonzero owner identity.
18    pub const fn new(value: u64) -> Option<Self> {
19        if value == 0 {
20            None
21        } else {
22            Some(Self(value))
23        }
24    }
25
26    /// Returns the exact process-local owner value.
27    pub const fn value(self) -> u64 {
28        self.0
29    }
30}
31
32/// Process-local coordinate-history generation bound to every retained payload.
33#[derive(Debug, Clone, Copy, Eq, Hash, Ord, PartialEq, PartialOrd)]
34pub struct RealtimePayloadGeneration(u64);
35
36impl RealtimePayloadGeneration {
37    /// Creates a nonzero history-generation identity.
38    pub const fn new(value: u64) -> Option<Self> {
39        if value == 0 {
40            None
41        } else {
42            Some(Self(value))
43        }
44    }
45
46    /// Returns the exact process-local history-generation value.
47    pub const fn value(self) -> u64 {
48        self.0
49    }
50}
51
52/// Exact semantic identity carried by every realtime coordinate payload.
53#[derive(Debug, Clone, Eq, PartialEq)]
54pub struct RealtimePayloadContract {
55    schedule: RealtimeSpeechConfig,
56    batch: NonZeroUsize,
57    text_domain: TokenDomain,
58    audio_domain: TokenDomain,
59    generation: RealtimePayloadGeneration,
60    owner: RealtimePayloadOwnerIdentity,
61}
62
63impl RealtimePayloadContract {
64    /// Creates a contract bound to one schedule, batch, token geometry, generation, and owner.
65    pub fn new(
66        schedule: RealtimeSpeechConfig,
67        batch: usize,
68        text_domain: TokenDomain,
69        audio_domain: TokenDomain,
70        generation: RealtimePayloadGeneration,
71        owner: RealtimePayloadOwnerIdentity,
72    ) -> Result<Self, RealtimePayloadContractError> {
73        let batch = NonZeroUsize::new(batch).ok_or(RealtimePayloadContractError::EmptyBatch)?;
74        Ok(Self {
75            schedule,
76            batch,
77            text_domain,
78            audio_domain,
79            generation,
80            owner,
81        })
82    }
83
84    /// Returns the exact normalized speech schedule.
85    pub const fn schedule(&self) -> &RealtimeSpeechConfig {
86        &self.schedule
87    }
88
89    /// Returns the positive batch cardinality.
90    pub const fn batch(&self) -> NonZeroUsize {
91        self.batch
92    }
93
94    /// Returns the admitted text-token domain.
95    pub const fn text_domain(&self) -> TokenDomain {
96        self.text_domain
97    }
98
99    /// Returns the admitted audio-token domain shared by every audio slot.
100    pub const fn audio_domain(&self) -> TokenDomain {
101        self.audio_domain
102    }
103
104    /// Returns the request or session generation identity.
105    pub const fn generation(&self) -> RealtimePayloadGeneration {
106        self.generation
107    }
108
109    /// Returns the exact state or resource owner identity.
110    pub const fn owner(&self) -> RealtimePayloadOwnerIdentity {
111        self.owner
112    }
113
114    /// Resolves the token domain selected by one admitted text or audio slot.
115    pub fn slot_domain(
116        &self,
117        slot: RealtimeFrameSlot,
118    ) -> Result<TokenDomain, RealtimePayloadContractError> {
119        match slot {
120            RealtimeFrameSlot::Text => Ok(self.text_domain),
121            RealtimeFrameSlot::Audio(codebook)
122                if codebook < self.schedule.total_audio_codebooks() =>
123            {
124                Ok(self.audio_domain)
125            }
126            _ => Err(RealtimePayloadContractError::InvalidSlot {
127                slot,
128                total_audio_codebooks: self.schedule.total_audio_codebooks(),
129            }),
130        }
131    }
132
133    /// Validates another contract against every exact semantic identity field.
134    pub fn validate(&self, contract: &Self) -> Result<(), RealtimePayloadContractError> {
135        if self.schedule != contract.schedule {
136            return Err(RealtimePayloadContractError::ScheduleMismatch);
137        }
138        if self.batch != contract.batch {
139            return Err(RealtimePayloadContractError::BatchMismatch);
140        }
141        if self.text_domain != contract.text_domain {
142            return Err(RealtimePayloadContractError::TextDomainMismatch);
143        }
144        if self.audio_domain != contract.audio_domain {
145            return Err(RealtimePayloadContractError::AudioDomainMismatch);
146        }
147        if self.generation != contract.generation {
148            return Err(RealtimePayloadContractError::GenerationMismatch);
149        }
150        if self.owner != contract.owner {
151            return Err(RealtimePayloadContractError::OwnerMismatch);
152        }
153        Ok(())
154    }
155}
156
157/// One opaque payload bound to an exact coordinate and semantic contract.
158#[derive(Debug, Clone, Eq, PartialEq)]
159pub struct RealtimePayloadEnvelope<P> {
160    contract: RealtimePayloadContract,
161    coordinate: RealtimeSlotCoordinate,
162    domain: TokenDomain,
163    payload: P,
164}
165
166impl<P> RealtimePayloadEnvelope<P> {
167    /// Binds a payload to one admitted coordinate and its contract-derived token domain.
168    pub fn new(
169        contract: RealtimePayloadContract,
170        coordinate: RealtimeSlotCoordinate,
171        payload: P,
172    ) -> Result<Self, RealtimePayloadContractError> {
173        let domain = contract.slot_domain(coordinate.slot())?;
174        Ok(Self {
175            contract,
176            coordinate,
177            domain,
178            payload,
179        })
180    }
181
182    /// Returns the exact semantic payload contract.
183    pub const fn contract(&self) -> &RealtimePayloadContract {
184        &self.contract
185    }
186
187    /// Returns the exact delayed-frame coordinate.
188    pub const fn coordinate(&self) -> RealtimeSlotCoordinate {
189        self.coordinate
190    }
191
192    /// Returns the token domain derived from the coordinate slot and contract.
193    pub const fn domain(&self) -> TokenDomain {
194        self.domain
195    }
196
197    /// Returns the opaque payload.
198    pub const fn payload(&self) -> &P {
199        &self.payload
200    }
201
202    /// Validates handoff against another exact contract and coordinate.
203    pub fn validate(
204        &self,
205        contract: &RealtimePayloadContract,
206        coordinate: RealtimeSlotCoordinate,
207    ) -> Result<(), RealtimePayloadContractError> {
208        self.contract.validate(contract)?;
209        if self.coordinate != coordinate {
210            return Err(RealtimePayloadContractError::CoordinateMismatch);
211        }
212        let domain = contract.slot_domain(coordinate.slot())?;
213        if self.domain != domain {
214            return Err(RealtimePayloadContractError::SlotDomainMismatch);
215        }
216        Ok(())
217    }
218
219    /// Consumes the envelope into its contract, coordinate, derived domain, and payload.
220    pub fn into_parts(
221        self,
222    ) -> (
223        RealtimePayloadContract,
224        RealtimeSlotCoordinate,
225        TokenDomain,
226        P,
227    ) {
228        (self.contract, self.coordinate, self.domain, self.payload)
229    }
230}
231
232/// Stable failure while creating or validating exact realtime payload identity.
233#[derive(Debug, Clone, Eq, PartialEq, thiserror::Error)]
234#[non_exhaustive]
235pub enum RealtimePayloadContractError {
236    /// A payload contract cannot represent an empty batch.
237    #[error("realtime payload contract batch is empty")]
238    EmptyBatch,
239    /// The normalized speech schedule differs.
240    #[error("realtime payload contract schedule does not match")]
241    ScheduleMismatch,
242    /// The positive batch cardinality differs.
243    #[error("realtime payload contract batch does not match")]
244    BatchMismatch,
245    /// The text-token domain differs.
246    #[error("realtime payload contract text-token domain does not match")]
247    TextDomainMismatch,
248    /// The audio-token domain differs.
249    #[error("realtime payload contract audio-token domain does not match")]
250    AudioDomainMismatch,
251    /// The request or session generation identity differs.
252    #[error("realtime payload contract generation does not match")]
253    GenerationMismatch,
254    /// The state or resource owner identity differs.
255    #[error("realtime payload contract owner does not match")]
256    OwnerMismatch,
257    /// The delayed-frame coordinate differs.
258    #[error("realtime payload coordinate does not match")]
259    CoordinateMismatch,
260    /// The retained slot domain differs from the contract-derived domain.
261    #[error("realtime payload slot token domain does not match")]
262    SlotDomainMismatch,
263    /// A coordinate does not name text or an admitted audio codebook.
264    #[error(
265        "realtime payload slot {slot:?} is outside text plus {total_audio_codebooks} audio codebooks"
266    )]
267    InvalidSlot {
268        /// Invalid schedule slot.
269        slot: RealtimeFrameSlot,
270        /// Admitted audio-codebook count.
271        total_audio_codebooks: usize,
272    },
273}
274
275/// Payloads retained at exact text/audio coordinates for one speech schedule.
276#[derive(Debug, Clone, Eq, PartialEq)]
277pub struct RealtimePayloadHistory<P> {
278    schedule: RealtimeSpeechConfig,
279    contract: Option<RealtimePayloadContract>,
280    payloads: BTreeMap<RealtimeSlotCoordinate, RealtimePayloadEnvelope<P>>,
281}
282
283impl<P> RealtimePayloadHistory<P> {
284    /// Creates an empty pre-first-frame history bound only to one exact schedule.
285    ///
286    /// A payload contract must be bound before any coordinate payload can be
287    /// published. This unbound form exists because session batch is not known
288    /// until the first frame is accepted.
289    pub fn new(schedule: RealtimeSpeechConfig) -> Self {
290        Self {
291            schedule,
292            contract: None,
293            payloads: BTreeMap::new(),
294        }
295    }
296
297    /// Creates an empty history bound to one complete payload contract.
298    pub fn with_contract(contract: RealtimePayloadContract) -> Self {
299        Self {
300            schedule: contract.schedule().clone(),
301            contract: Some(contract),
302            payloads: BTreeMap::new(),
303        }
304    }
305
306    /// Returns the exact normalized schedule bound to this history.
307    pub const fn schedule(&self) -> &RealtimeSpeechConfig {
308        &self.schedule
309    }
310
311    /// Returns the exact payload contract after first-frame binding.
312    pub const fn contract(&self) -> Option<&RealtimePayloadContract> {
313        self.contract.as_ref()
314    }
315
316    /// Binds the first complete contract or validates an already-bound history.
317    ///
318    /// Failed validation never changes the current contract or payloads.
319    pub fn bind_or_validate_contract(
320        &mut self,
321        contract: &RealtimePayloadContract,
322    ) -> Result<(), RealtimePayloadHistoryError> {
323        if &self.schedule != contract.schedule() {
324            return Err(RealtimePayloadHistoryError::PayloadContract(
325                RealtimePayloadContractError::ScheduleMismatch,
326            ));
327        }
328        if let Some(current) = &self.contract {
329            current
330                .validate(contract)
331                .map_err(RealtimePayloadHistoryError::PayloadContract)
332        } else {
333            debug_assert!(self.payloads.is_empty());
334            self.contract = Some(contract.clone());
335            Ok(())
336        }
337    }
338
339    /// Validates whether a branch history may replace this canonical history.
340    pub fn validate_successor(&self, successor: &Self) -> Result<(), RealtimePayloadHistoryError> {
341        self.validate_schedule(&successor.schedule)?;
342        match (&self.contract, &successor.contract) {
343            (Some(current), Some(candidate)) => current
344                .validate(candidate)
345                .map_err(RealtimePayloadHistoryError::PayloadContract),
346            (None, Some(_)) if self.payloads.is_empty() => Ok(()),
347            (None, None) => Ok(()),
348            (Some(_), None) | (None, Some(_)) => Err(RealtimePayloadHistoryError::UnboundContract),
349        }
350    }
351
352    /// Returns the number of retained coordinate payloads.
353    pub fn len(&self) -> usize {
354        self.payloads.len()
355    }
356
357    /// Returns whether no payloads are retained.
358    pub fn is_empty(&self) -> bool {
359        self.payloads.is_empty()
360    }
361
362    /// Rejects handoff to any materially different normalized schedule.
363    pub fn validate_schedule(
364        &self,
365        schedule: &RealtimeSpeechConfig,
366    ) -> Result<(), RealtimePayloadHistoryError> {
367        if &self.schedule == schedule {
368            Ok(())
369        } else {
370            Err(RealtimePayloadHistoryError::ScheduleMismatch)
371        }
372    }
373
374    /// Resolves an absolute coordinate by adding the exact configured slot delay.
375    ///
376    /// This is only coordinate arithmetic; it does not assign temporal-input or
377    /// prediction-target meaning to the resulting slot.
378    pub fn delayed_coordinate(
379        &self,
380        schedule: &RealtimeSpeechConfig,
381        base_position: usize,
382        slot: RealtimeFrameSlot,
383    ) -> Result<RealtimeSlotCoordinate, RealtimePayloadHistoryError> {
384        self.validate_schedule(schedule)?;
385        let delay = self.slot_delay(slot)?;
386        let position = base_position.checked_add(delay).ok_or(
387            RealtimePayloadHistoryError::CoordinateOverflow {
388                base_position,
389                delay,
390            },
391        )?;
392        Ok(RealtimeSlotCoordinate::new(position, slot))
393    }
394
395    /// Inserts one payload at one exact validated coordinate.
396    pub fn insert(
397        &mut self,
398        schedule: &RealtimeSpeechConfig,
399        coordinate: RealtimeSlotCoordinate,
400        payload: P,
401    ) -> Result<(), RealtimePayloadHistoryError> {
402        self.insert_many(schedule, [(coordinate, payload)])
403    }
404
405    /// Calculates one delayed coordinate and inserts its payload atomically.
406    pub fn insert_delayed(
407        &mut self,
408        schedule: &RealtimeSpeechConfig,
409        base_position: usize,
410        slot: RealtimeFrameSlot,
411        payload: P,
412    ) -> Result<RealtimeSlotCoordinate, RealtimePayloadHistoryError> {
413        let coordinate = self.delayed_coordinate(schedule, base_position, slot)?;
414        self.insert(schedule, coordinate, payload)?;
415        Ok(coordinate)
416    }
417
418    /// Inserts a complete set of coordinate payloads as one publication.
419    ///
420    /// Invalid slots and duplicates are detected before any payload is visible.
421    pub fn insert_many(
422        &mut self,
423        schedule: &RealtimeSpeechConfig,
424        payloads: impl IntoIterator<Item = (RealtimeSlotCoordinate, P)>,
425    ) -> Result<(), RealtimePayloadHistoryError> {
426        self.validate_schedule(schedule)?;
427        let contract = self
428            .contract
429            .as_ref()
430            .ok_or(RealtimePayloadHistoryError::UnboundContract)?
431            .clone();
432        let payloads = payloads.into_iter().collect::<Vec<_>>();
433        let mut pending = BTreeSet::new();
434        for (coordinate, _) in &payloads {
435            self.validate_coordinate(*coordinate)?;
436            if self.payloads.contains_key(coordinate) || !pending.insert(*coordinate) {
437                return Err(RealtimePayloadHistoryError::DuplicatePayload {
438                    coordinate: *coordinate,
439                });
440            }
441        }
442        let envelopes = payloads
443            .into_iter()
444            .map(|(coordinate, payload)| {
445                RealtimePayloadEnvelope::new(contract.clone(), coordinate, payload)
446                    .map(|envelope| (coordinate, envelope))
447                    .map_err(RealtimePayloadHistoryError::PayloadContract)
448            })
449            .collect::<Result<Vec<_>, _>>()?;
450        self.payloads.extend(envelopes);
451        Ok(())
452    }
453
454    /// Atomically publishes validated coordinate payloads with explicit overwrite semantics.
455    ///
456    /// This is reserved for schedule-authorized placement and target updates.
457    /// Ordinary admission should use [`Self::insert_many`] so accidental
458    /// duplicate producers still fail closed.
459    pub fn overwrite_many(
460        &mut self,
461        schedule: &RealtimeSpeechConfig,
462        payloads: impl IntoIterator<Item = (RealtimeSlotCoordinate, P)>,
463    ) -> Result<(), RealtimePayloadHistoryError> {
464        self.validate_schedule(schedule)?;
465        let contract = self
466            .contract
467            .as_ref()
468            .ok_or(RealtimePayloadHistoryError::UnboundContract)?
469            .clone();
470        let payloads = payloads.into_iter().collect::<Vec<_>>();
471        for (coordinate, _) in &payloads {
472            self.validate_coordinate(*coordinate)?;
473        }
474        let envelopes = payloads
475            .into_iter()
476            .map(|(coordinate, payload)| {
477                RealtimePayloadEnvelope::new(contract.clone(), coordinate, payload)
478                    .map(|envelope| (coordinate, envelope))
479                    .map_err(RealtimePayloadHistoryError::PayloadContract)
480            })
481            .collect::<Result<Vec<_>, _>>()?;
482        self.payloads.extend(envelopes);
483        Ok(())
484    }
485
486    /// Returns a payload at one exact validated coordinate, when present.
487    pub fn get(
488        &self,
489        schedule: &RealtimeSpeechConfig,
490        coordinate: RealtimeSlotCoordinate,
491    ) -> Result<Option<&P>, RealtimePayloadHistoryError> {
492        self.validate_schedule(schedule)?;
493        self.validate_coordinate(coordinate)?;
494        Ok(self
495            .payloads
496            .get(&coordinate)
497            .map(RealtimePayloadEnvelope::payload))
498    }
499
500    /// Returns the typed envelope at one exact validated coordinate, when present.
501    pub fn envelope(
502        &self,
503        schedule: &RealtimeSpeechConfig,
504        coordinate: RealtimeSlotCoordinate,
505    ) -> Result<Option<&RealtimePayloadEnvelope<P>>, RealtimePayloadHistoryError> {
506        self.validate_schedule(schedule)?;
507        self.validate_coordinate(coordinate)?;
508        Ok(self.payloads.get(&coordinate))
509    }
510
511    /// Resolves one required payload or fails with its exact missing coordinate.
512    pub fn required(
513        &self,
514        schedule: &RealtimeSpeechConfig,
515        coordinate: RealtimeSlotCoordinate,
516    ) -> Result<&P, RealtimePayloadHistoryError> {
517        self.get(schedule, coordinate)?
518            .ok_or(RealtimePayloadHistoryError::MissingPayload { coordinate })
519    }
520
521    /// Resolves required coordinates in caller order, retaining duplicate reads.
522    pub fn resolve_required(
523        &self,
524        schedule: &RealtimeSpeechConfig,
525        coordinates: impl IntoIterator<Item = RealtimeSlotCoordinate>,
526    ) -> Result<Vec<&P>, RealtimePayloadHistoryError> {
527        self.validate_schedule(schedule)?;
528        coordinates
529            .into_iter()
530            .map(|coordinate| self.required(schedule, coordinate))
531            .collect()
532    }
533
534    /// Prunes coordinates older than the deterministic delayed-history window.
535    ///
536    /// For next frontier `n` and maximum delay `d`, positions before
537    /// `n - (d + 2)` are removed. The additional position retains the oldest
538    /// payload referenced by the just-submitted transition until its completion
539    /// has captured [`Self::retained_values`]. Warm-up clamps the minimum
540    /// position to zero. The return value is the number of payloads removed.
541    pub fn prune_for_next_frontier(
542        &mut self,
543        schedule: &RealtimeSpeechConfig,
544        next_frontier: usize,
545    ) -> Result<usize, RealtimePayloadHistoryError> {
546        self.validate_schedule(schedule)?;
547        let retained_positions = schedule.max_delay().checked_add(2).ok_or(
548            RealtimePayloadHistoryError::RetentionWindowOverflow {
549                max_delay: schedule.max_delay(),
550            },
551        )?;
552        let minimum = next_frontier.saturating_sub(retained_positions);
553        let previous = self.payloads.len();
554        self.payloads
555            .retain(|coordinate, _| coordinate.position() >= minimum);
556        Ok(previous - self.payloads.len())
557    }
558
559    /// Iterates retained payloads in stable coordinate order.
560    pub fn retained_values(&self) -> impl Iterator<Item = &P> {
561        self.payloads.values().map(RealtimePayloadEnvelope::payload)
562    }
563
564    /// Iterates retained typed envelopes in stable coordinate order.
565    pub fn envelopes(&self) -> impl Iterator<Item = &RealtimePayloadEnvelope<P>> {
566        self.payloads.values()
567    }
568
569    /// Iterates exact coordinates and retained payloads in stable order.
570    pub fn entries(&self) -> impl Iterator<Item = (RealtimeSlotCoordinate, &P)> {
571        self.payloads
572            .iter()
573            .map(|(coordinate, envelope)| (*coordinate, envelope.payload()))
574    }
575
576    fn validate_coordinate(
577        &self,
578        coordinate: RealtimeSlotCoordinate,
579    ) -> Result<(), RealtimePayloadHistoryError> {
580        self.slot_delay(coordinate.slot()).map(|_| ())
581    }
582
583    fn slot_delay(&self, slot: RealtimeFrameSlot) -> Result<usize, RealtimePayloadHistoryError> {
584        match slot {
585            RealtimeFrameSlot::Text => Ok(self.schedule.text_delay()),
586            RealtimeFrameSlot::Audio(codebook) => {
587                self.schedule.audio_delays().get(codebook).copied().ok_or(
588                    RealtimePayloadHistoryError::InvalidSlot {
589                        slot,
590                        total_audio_codebooks: self.schedule.total_audio_codebooks(),
591                    },
592                )
593            }
594            _ => Err(RealtimePayloadHistoryError::InvalidSlot {
595                slot,
596                total_audio_codebooks: self.schedule.total_audio_codebooks(),
597            }),
598        }
599    }
600}
601
602/// Stable failure while retaining or resolving delayed-coordinate payloads.
603#[derive(Debug, Clone, Eq, PartialEq, thiserror::Error)]
604#[non_exhaustive]
605pub enum RealtimePayloadHistoryError {
606    /// Coordinate payload operations require a complete exact contract.
607    #[error("realtime payload history has no bound payload contract")]
608    UnboundContract,
609    /// The complete payload identity differs from the bound history.
610    #[error(transparent)]
611    PayloadContract(RealtimePayloadContractError),
612    /// The caller supplied a materially different normalized schedule.
613    #[error("realtime payload history does not match the normalized schedule")]
614    ScheduleMismatch,
615    /// A coordinate does not name text or an admitted audio codebook.
616    #[error(
617        "realtime payload slot {slot:?} is outside text plus {total_audio_codebooks} audio codebooks"
618    )]
619    InvalidSlot {
620        /// Invalid schedule slot.
621        slot: RealtimeFrameSlot,
622        /// Admitted audio-codebook count.
623        total_audio_codebooks: usize,
624    },
625    /// A coordinate already has a payload in the current or pending publication.
626    #[error("realtime coordinate {coordinate:?} already contains a payload")]
627    DuplicatePayload {
628        /// Duplicate coordinate.
629        coordinate: RealtimeSlotCoordinate,
630    },
631    /// A required coordinate has no retained payload.
632    #[error("realtime coordinate {coordinate:?} has no retained payload")]
633    MissingPayload {
634        /// Missing coordinate.
635        coordinate: RealtimeSlotCoordinate,
636    },
637    /// Adding a schedule delay exceeded the coordinate representation.
638    #[error("realtime payload coordinate overflowed from base {base_position} plus delay {delay}")]
639    CoordinateOverflow {
640        /// Undelayed base position.
641        base_position: usize,
642        /// Exact configured slot delay.
643        delay: usize,
644    },
645    /// Maximum delay cannot be represented as a retained-position count.
646    #[error("realtime payload retention window overflowed for maximum delay {max_delay}")]
647    RetentionWindowOverflow {
648        /// Exact configured maximum delay.
649        max_delay: usize,
650    },
651}
652
653#[cfg(test)]
654mod tests {
655    use eredu_core::RealtimeFrameConvention;
656
657    use super::*;
658
659    fn schedule() -> RealtimeSpeechConfig {
660        RealtimeSpeechConfig::new(
661            2,
662            1,
663            1,
664            1,
665            0,
666            1,
667            RealtimeFrameConvention::FeedbackAlignedHistory,
668            vec![2, 0, 3],
669        )
670        .unwrap()
671    }
672
673    fn coordinate(position: usize, slot: RealtimeFrameSlot) -> RealtimeSlotCoordinate {
674        RealtimeSlotCoordinate::new(position, slot)
675    }
676
677    fn payload_contract(
678        schedule: RealtimeSpeechConfig,
679        batch: usize,
680        text_domain: TokenDomain,
681        audio_domain: TokenDomain,
682        generation: u64,
683        owner: u64,
684    ) -> RealtimePayloadContract {
685        RealtimePayloadContract::new(
686            schedule,
687            batch,
688            text_domain,
689            audio_domain,
690            RealtimePayloadGeneration::new(generation).unwrap(),
691            RealtimePayloadOwnerIdentity::new(owner).unwrap(),
692        )
693        .unwrap()
694    }
695
696    fn exact_payload_contract() -> RealtimePayloadContract {
697        payload_contract(
698            schedule(),
699            2,
700            TokenDomain::new(32),
701            TokenDomain::new(16),
702            7,
703            3,
704        )
705    }
706
707    fn payload_history<P>() -> RealtimePayloadHistory<P> {
708        RealtimePayloadHistory::with_contract(exact_payload_contract())
709    }
710
711    #[test]
712    fn payload_contract_requires_nonempty_batch_and_typed_nonempty_identities() {
713        assert_eq!(
714            RealtimePayloadContract::new(
715                schedule(),
716                0,
717                TokenDomain::new(32),
718                TokenDomain::new(16),
719                RealtimePayloadGeneration::new(7).unwrap(),
720                RealtimePayloadOwnerIdentity::new(3).unwrap(),
721            ),
722            Err(RealtimePayloadContractError::EmptyBatch)
723        );
724        assert_eq!(RealtimePayloadGeneration::new(0), None);
725        assert_eq!(RealtimePayloadOwnerIdentity::new(0), None);
726
727        let contract = exact_payload_contract();
728        assert_eq!(contract.schedule(), &schedule());
729        assert_eq!(contract.batch().get(), 2);
730        assert_eq!(contract.text_domain(), TokenDomain::new(32));
731        assert_eq!(contract.audio_domain(), TokenDomain::new(16));
732        assert_eq!(contract.generation().value(), 7);
733        assert_eq!(contract.owner().value(), 3);
734    }
735
736    #[test]
737    fn payload_contract_rejects_each_exact_identity_perturbation() {
738        let exact = exact_payload_contract();
739        let other_schedule = RealtimeSpeechConfig::new(
740            2,
741            1,
742            1,
743            1,
744            0,
745            1,
746            RealtimeFrameConvention::AbsoluteDelayedSlots,
747            vec![2, 0, 3],
748        )
749        .unwrap();
750
751        assert_eq!(
752            exact.validate(&payload_contract(
753                other_schedule,
754                2,
755                TokenDomain::new(32),
756                TokenDomain::new(16),
757                7,
758                3,
759            )),
760            Err(RealtimePayloadContractError::ScheduleMismatch)
761        );
762        assert_eq!(
763            exact.validate(&payload_contract(
764                schedule(),
765                3,
766                TokenDomain::new(32),
767                TokenDomain::new(16),
768                7,
769                3,
770            )),
771            Err(RealtimePayloadContractError::BatchMismatch)
772        );
773        assert_eq!(
774            exact.validate(&payload_contract(
775                schedule(),
776                2,
777                TokenDomain::new(33),
778                TokenDomain::new(16),
779                7,
780                3,
781            )),
782            Err(RealtimePayloadContractError::TextDomainMismatch)
783        );
784        assert_eq!(
785            exact.validate(&payload_contract(
786                schedule(),
787                2,
788                TokenDomain::new(32),
789                TokenDomain::new(17),
790                7,
791                3,
792            )),
793            Err(RealtimePayloadContractError::AudioDomainMismatch)
794        );
795        assert_eq!(
796            exact.validate(&payload_contract(
797                schedule(),
798                2,
799                TokenDomain::new(32),
800                TokenDomain::new(16),
801                8,
802                3,
803            )),
804            Err(RealtimePayloadContractError::GenerationMismatch)
805        );
806        assert_eq!(
807            exact.validate(&payload_contract(
808                schedule(),
809                2,
810                TokenDomain::new(32),
811                TokenDomain::new(16),
812                7,
813                4,
814            )),
815            Err(RealtimePayloadContractError::OwnerMismatch)
816        );
817    }
818
819    #[test]
820    fn payload_envelope_derives_slot_domain_and_validates_exact_coordinate() {
821        let contract = exact_payload_contract();
822        let text_coordinate = coordinate(5, RealtimeFrameSlot::Text);
823        let text = RealtimePayloadEnvelope::new(contract.clone(), text_coordinate, "text").unwrap();
824        assert_eq!(text.contract(), &contract);
825        assert_eq!(text.coordinate(), text_coordinate);
826        assert_eq!(text.domain(), TokenDomain::new(32));
827        assert_eq!(text.payload(), &"text");
828        assert_eq!(text.validate(&contract, text_coordinate), Ok(()));
829
830        let audio_coordinate = coordinate(5, RealtimeFrameSlot::Audio(1));
831        let audio =
832            RealtimePayloadEnvelope::new(contract.clone(), audio_coordinate, "audio").unwrap();
833        assert_eq!(audio.domain(), TokenDomain::new(16));
834        assert_eq!(audio.validate(&contract, audio_coordinate), Ok(()));
835        assert_eq!(
836            audio.validate(&contract, coordinate(6, RealtimeFrameSlot::Audio(1))),
837            Err(RealtimePayloadContractError::CoordinateMismatch)
838        );
839
840        assert_eq!(
841            RealtimePayloadEnvelope::new(
842                contract,
843                coordinate(5, RealtimeFrameSlot::Audio(2)),
844                "invalid",
845            ),
846            Err(RealtimePayloadContractError::InvalidSlot {
847                slot: RealtimeFrameSlot::Audio(2),
848                total_audio_codebooks: 2,
849            })
850        );
851    }
852
853    #[test]
854    fn history_requires_first_contract_binding_and_stores_only_typed_envelopes() {
855        let schedule = schedule();
856        let text = coordinate(0, RealtimeFrameSlot::Text);
857        let audio = coordinate(0, RealtimeFrameSlot::Audio(0));
858        let mut history = RealtimePayloadHistory::new(schedule.clone());
859        assert!(history.contract().is_none());
860        assert_eq!(history.get(&schedule, text), Ok(None));
861        assert_eq!(
862            history.insert(&schedule, text, 3),
863            Err(RealtimePayloadHistoryError::UnboundContract)
864        );
865        assert!(history.is_empty());
866
867        let contract = exact_payload_contract();
868        history.bind_or_validate_contract(&contract).unwrap();
869        history
870            .insert_many(&schedule, [(text, 3), (audio, 5)])
871            .unwrap();
872        assert_eq!(history.contract(), Some(&contract));
873        let envelopes = history.envelopes().collect::<Vec<_>>();
874        assert_eq!(envelopes.len(), 2);
875        assert_eq!(envelopes[0].contract(), &contract);
876        assert_eq!(envelopes[0].coordinate(), text);
877        assert_eq!(envelopes[0].domain(), TokenDomain::new(32));
878        assert_eq!(envelopes[0].payload(), &3);
879        assert_eq!(envelopes[1].contract(), &contract);
880        assert_eq!(envelopes[1].coordinate(), audio);
881        assert_eq!(envelopes[1].domain(), TokenDomain::new(16));
882        assert_eq!(
883            history.envelope(&schedule, audio).unwrap(),
884            Some(envelopes[1])
885        );
886    }
887
888    #[test]
889    fn bound_history_rejects_every_contract_perturbation_without_mutation() {
890        let exact = exact_payload_contract();
891        let schedule = schedule();
892        let text = coordinate(0, RealtimeFrameSlot::Text);
893        let mut history = RealtimePayloadHistory::with_contract(exact.clone());
894        history.insert(&schedule, text, 3).unwrap();
895        let other_schedule = RealtimeSpeechConfig::new(
896            2,
897            1,
898            1,
899            1,
900            0,
901            1,
902            RealtimeFrameConvention::AbsoluteDelayedSlots,
903            vec![2, 0, 3],
904        )
905        .unwrap();
906        let perturbations = [
907            (
908                payload_contract(
909                    other_schedule,
910                    2,
911                    TokenDomain::new(32),
912                    TokenDomain::new(16),
913                    7,
914                    3,
915                ),
916                RealtimePayloadContractError::ScheduleMismatch,
917            ),
918            (
919                payload_contract(
920                    schedule.clone(),
921                    3,
922                    TokenDomain::new(32),
923                    TokenDomain::new(16),
924                    7,
925                    3,
926                ),
927                RealtimePayloadContractError::BatchMismatch,
928            ),
929            (
930                payload_contract(
931                    schedule.clone(),
932                    2,
933                    TokenDomain::new(33),
934                    TokenDomain::new(16),
935                    7,
936                    3,
937                ),
938                RealtimePayloadContractError::TextDomainMismatch,
939            ),
940            (
941                payload_contract(
942                    schedule.clone(),
943                    2,
944                    TokenDomain::new(32),
945                    TokenDomain::new(17),
946                    7,
947                    3,
948                ),
949                RealtimePayloadContractError::AudioDomainMismatch,
950            ),
951            (
952                payload_contract(
953                    schedule.clone(),
954                    2,
955                    TokenDomain::new(32),
956                    TokenDomain::new(16),
957                    8,
958                    3,
959                ),
960                RealtimePayloadContractError::GenerationMismatch,
961            ),
962            (
963                payload_contract(
964                    schedule.clone(),
965                    2,
966                    TokenDomain::new(32),
967                    TokenDomain::new(16),
968                    7,
969                    4,
970                ),
971                RealtimePayloadContractError::OwnerMismatch,
972            ),
973        ];
974
975        for (candidate, expected) in perturbations {
976            assert_eq!(
977                history.bind_or_validate_contract(&candidate),
978                Err(RealtimePayloadHistoryError::PayloadContract(expected))
979            );
980            assert_eq!(history.contract(), Some(&exact));
981            assert_eq!(history.required(&schedule, text), Ok(&3));
982            assert_eq!(history.len(), 1);
983        }
984    }
985
986    #[test]
987    fn history_clone_and_prune_preserve_envelope_identity() {
988        let schedule = schedule();
989        let contract = exact_payload_contract();
990        let mut history = RealtimePayloadHistory::with_contract(contract.clone());
991        for position in 0..=6 {
992            history
993                .insert(
994                    &schedule,
995                    coordinate(position, RealtimeFrameSlot::Text),
996                    position,
997                )
998                .unwrap();
999        }
1000        let mut branch = history.clone();
1001        assert_eq!(branch.contract(), Some(&contract));
1002        assert!(branch
1003            .envelopes()
1004            .all(|envelope| envelope.contract() == &contract));
1005
1006        assert_eq!(branch.prune_for_next_frontier(&schedule, 7), Ok(2));
1007        assert_eq!(branch.len(), 5);
1008        assert_eq!(history.len(), 7);
1009        assert!(branch.envelopes().all(|envelope| {
1010            envelope.coordinate().position() >= 2 && envelope.contract() == &contract
1011        }));
1012    }
1013
1014    #[test]
1015    fn exact_insert_get_and_required_resolution_preserve_coordinate_order() {
1016        let schedule = schedule();
1017        let mut history = payload_history();
1018        let text = coordinate(2, RealtimeFrameSlot::Text);
1019        let audio_zero = coordinate(2, RealtimeFrameSlot::Audio(0));
1020        let audio_one = coordinate(2, RealtimeFrameSlot::Audio(1));
1021        history
1022            .insert_many(
1023                &schedule,
1024                [(audio_one, "a1"), (text, "text"), (audio_zero, "a0")],
1025            )
1026            .unwrap();
1027
1028        assert_eq!(history.get(&schedule, text).unwrap(), Some(&"text"));
1029        assert_eq!(history.required(&schedule, audio_one).unwrap(), &"a1");
1030        assert_eq!(
1031            history
1032                .resolve_required(&schedule, [audio_one, text, audio_one])
1033                .unwrap(),
1034            vec![&"a1", &"text", &"a1"]
1035        );
1036        assert_eq!(
1037            history.retained_values().copied().collect::<Vec<_>>(),
1038            vec!["text", "a0", "a1"]
1039        );
1040        assert_eq!(history.schedule(), &schedule);
1041    }
1042
1043    #[test]
1044    fn invalid_or_duplicate_publications_are_atomic() {
1045        let schedule = schedule();
1046        let mut history = payload_history();
1047        let retained = coordinate(0, RealtimeFrameSlot::Text);
1048        history.insert(&schedule, retained, 7).unwrap();
1049        let invalid = coordinate(1, RealtimeFrameSlot::Audio(2));
1050
1051        assert_eq!(
1052            history.insert_many(
1053                &schedule,
1054                [
1055                    (coordinate(1, RealtimeFrameSlot::Audio(0)), 8),
1056                    (invalid, 9)
1057                ]
1058            ),
1059            Err(RealtimePayloadHistoryError::InvalidSlot {
1060                slot: RealtimeFrameSlot::Audio(2),
1061                total_audio_codebooks: 2,
1062            })
1063        );
1064        assert_eq!(history.len(), 1);
1065        assert_eq!(
1066            history.insert_many(&schedule, [(invalid, 8), (invalid, 9)]),
1067            Err(RealtimePayloadHistoryError::InvalidSlot {
1068                slot: RealtimeFrameSlot::Audio(2),
1069                total_audio_codebooks: 2,
1070            })
1071        );
1072        let pending = coordinate(1, RealtimeFrameSlot::Audio(0));
1073        assert_eq!(
1074            history.insert_many(&schedule, [(pending, 8), (pending, 9)]),
1075            Err(RealtimePayloadHistoryError::DuplicatePayload {
1076                coordinate: pending,
1077            })
1078        );
1079        assert_eq!(history.len(), 1);
1080        assert_eq!(
1081            history.insert(&schedule, retained, 10),
1082            Err(RealtimePayloadHistoryError::DuplicatePayload {
1083                coordinate: retained,
1084            })
1085        );
1086        assert_eq!(history.required(&schedule, retained), Ok(&7));
1087    }
1088
1089    #[test]
1090    fn mismatch_missing_and_arithmetic_fail_without_mutation() {
1091        let schedule = schedule();
1092        let other = RealtimeSpeechConfig::new(
1093            2,
1094            1,
1095            1,
1096            1,
1097            0,
1098            1,
1099            RealtimeFrameConvention::AbsoluteDelayedSlots,
1100            vec![2, 0, 3],
1101        )
1102        .unwrap();
1103        let mut history = payload_history();
1104        let retained = coordinate(0, RealtimeFrameSlot::Text);
1105        history.insert(&schedule, retained, 7).unwrap();
1106
1107        assert_eq!(
1108            history.insert(&other, coordinate(1, RealtimeFrameSlot::Text), 8),
1109            Err(RealtimePayloadHistoryError::ScheduleMismatch)
1110        );
1111        let missing = coordinate(1, RealtimeFrameSlot::Audio(0));
1112        assert_eq!(
1113            history.required(&schedule, missing),
1114            Err(RealtimePayloadHistoryError::MissingPayload {
1115                coordinate: missing,
1116            })
1117        );
1118        assert_eq!(
1119            history.insert_delayed(&schedule, usize::MAX, RealtimeFrameSlot::Text, 9,),
1120            Err(RealtimePayloadHistoryError::CoordinateOverflow {
1121                base_position: usize::MAX,
1122                delay: 2,
1123            })
1124        );
1125        assert_eq!(history.len(), 1);
1126        assert_eq!(history.required(&schedule, retained), Ok(&7));
1127    }
1128
1129    #[test]
1130    fn pruning_uses_next_frontier_and_maximum_delay_deterministically() {
1131        let schedule = schedule();
1132        let mut history = payload_history();
1133        for position in 0..=5 {
1134            history
1135                .insert(
1136                    &schedule,
1137                    coordinate(position, RealtimeFrameSlot::Text),
1138                    position,
1139                )
1140                .unwrap();
1141        }
1142
1143        assert_eq!(history.prune_for_next_frontier(&schedule, 3), Ok(0));
1144        assert_eq!(history.prune_for_next_frontier(&schedule, 6), Ok(1));
1145        assert_eq!(
1146            history
1147                .entries()
1148                .map(|(coordinate, payload)| (coordinate.position(), *payload))
1149                .collect::<Vec<_>>(),
1150            vec![(1, 1), (2, 2), (3, 3), (4, 4), (5, 5)]
1151        );
1152    }
1153}
eredu_runtime/realtime_payload.rs

eredu_runtime/
realtime_payload.rs
