hydro_lang/sim/runtime/
tick_input.rs

1//! Tick-input hooks ([`TickInputHook`]): the per-kind hook types that buffer input for
2//! a tick across scheduling boundaries and decide, when the tick runs, what to release
3//! into that execution — batch hooks ([`StreamHook`], [`KeyedStreamHook`]) and snapshot
4//! hooks ([`SingletonHook`], [`KeyedSingletonHook`], and the choice-free
5//! [`PassthroughSingletonHook`]). Their scripted decision/status types and
6//! [`ScriptableHook`] impls live alongside them.
7
8use std::cell::RefCell;
9use std::collections::VecDeque;
10use std::hash::Hash;
11use std::rc::Rc;
12
13use bolero::generator::bolero_generator::driver::object::Borrowed;
14use bolero::{ValueGenerator, produce};
15use dfir_rs::rustc_hash::FxHashMap;
16use dfir_rs::util::unsync::mpsc::Sender;
17
18use super::{
19    HookLocationMeta, ManualDebug, RuntimeHook, ScriptDecision, ScriptableHook,
20    ScriptableTickInputHook, TickInputHook, TruncatedVecDebug, abort, log_release,
21};
22use crate::live_collections::stream::{NoOrder, Ordering, TotalOrder};
23
24pub struct StreamHook<T, Order: Ordering> {
25    pub input: Rc<RefCell<VecDeque<T>>>,
26    pub to_release: Option<Vec<T>>,
27    pub output: Sender<T>,
28    pub batch_location: HookLocationMeta,
29    pub format_item_debug: fn(&T) -> Option<String>,
30    pub _order: std::marker::PhantomData<Order>,
31}
32
33impl<T> RuntimeHook for StreamHook<T, TotalOrder> {
34    fn has_pending_input(&self) -> bool {
35        !self.input.borrow().is_empty()
36    }
37
38    fn only_one_possible_decision(&self) -> bool {
39        // One buffered element is still a real choice: release it or not.
40        self.input.borrow().is_empty()
41    }
42
43    fn release_decision(&mut self, log_writer: Option<&mut dyn std::fmt::Write>) {
44        if let Some(to_release) = self.to_release.take() {
45            if let Some(log_writer) = log_writer {
46                let HookLocationMeta {
47                    location: batch_location,
48                    line,
49                    caret_indent,
50                } = self.batch_location;
51                let note_str = if to_release.is_empty() {
52                    "^ releasing no items".to_owned()
53                } else {
54                    format!(
55                        "^ releasing items: {:?}",
56                        TruncatedVecDebug(
57                            RefCell::new(Some(to_release.iter())),
58                            8,
59                            self.format_item_debug
60                        )
61                    )
62                };
63
64                log_release(
65                    log_writer,
66                    batch_location,
67                    line,
68                    caret_indent,
69                    &note_str,
70                    colored::Color::Green,
71                );
72            }
73
74            for item in to_release {
75                self.output.try_send(item).unwrap();
76            }
77        } else {
78            panic!("No decision to release");
79        }
80    }
81
82    fn location_meta(&self) -> HookLocationMeta {
83        self.batch_location
84    }
85}
86
87impl<T> TickInputHook for StreamHook<T, TotalOrder> {
88    fn can_trigger_tick(&self) -> bool {
89        !self.input.borrow().is_empty()
90    }
91
92    fn autonomous_decision<'a>(&mut self, driver: &mut Borrowed<'a>, force_trigger: bool) -> bool {
93        let mut current_input = self.input.borrow_mut();
94        let count = ((if force_trigger { 1 } else { 0 })..=current_input.len())
95            .generate(driver)
96            .unwrap();
97
98        self.to_release = Some(current_input.drain(0..count).collect());
99        count > 0
100    }
101}
102
103impl<T> RuntimeHook for StreamHook<T, NoOrder> {
104    fn has_pending_input(&self) -> bool {
105        !self.input.borrow().is_empty()
106    }
107
108    fn only_one_possible_decision(&self) -> bool {
109        // One buffered element is still a real choice: release it or not.
110        self.input.borrow().is_empty()
111    }
112
113    fn release_decision(&mut self, log_writer: Option<&mut dyn std::fmt::Write>) {
114        if let Some(to_release) = self.to_release.take() {
115            if let Some(log_writer) = log_writer {
116                let HookLocationMeta {
117                    location: batch_location,
118                    line,
119                    caret_indent,
120                } = self.batch_location;
121                let note_str = if to_release.is_empty() {
122                    "^ releasing no items".to_owned()
123                } else {
124                    format!(
125                        "^ releasing unordered items: {:?}",
126                        TruncatedVecDebug(
127                            RefCell::new(Some(to_release.iter())),
128                            8,
129                            self.format_item_debug
130                        )
131                    )
132                };
133
134                log_release(
135                    log_writer,
136                    batch_location,
137                    line,
138                    caret_indent,
139                    &note_str,
140                    colored::Color::Green,
141                );
142            }
143
144            for item in to_release {
145                self.output.try_send(item).unwrap();
146            }
147        } else {
148            panic!("No decision to release");
149        }
150    }
151
152    fn location_meta(&self) -> HookLocationMeta {
153        self.batch_location
154    }
155}
156
157impl<T> TickInputHook for StreamHook<T, NoOrder> {
158    fn can_trigger_tick(&self) -> bool {
159        !self.input.borrow().is_empty()
160    }
161
162    fn autonomous_decision<'a>(&mut self, driver: &mut Borrowed<'a>, force_trigger: bool) -> bool {
163        let mut current_input = self.input.borrow_mut();
164        let mut out = vec![];
165        let mut min_index = 0;
166        while !current_input.is_empty() {
167            let must_release = force_trigger && out.is_empty();
168            if !must_release && produce().generate(driver).unwrap() {
169                break;
170            }
171
172            let idx = (min_index..current_input.len()).generate(driver).unwrap();
173            let item = current_input.remove(idx).unwrap();
174            out.push(item);
175
176            min_index = idx;
177            // Next time, only consider items at or after this index. The reason this is safe is
178            // because batching a `NoOrder` streams results in batches with a `NoOrder` guarantee.
179            // Therefore, simulating different order of elements _within_ a batch is redundant.
180
181            if min_index == current_input.len() {
182                break;
183            }
184        }
185
186        let triggered = !out.is_empty();
187        self.to_release = Some(out);
188        triggered
189    }
190}
191
192/// A scripted decision for a totally ordered batch hook.
193#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
194pub enum BatchDecision<T> {
195    /// Release the next `n` buffered elements.
196    Prefix(usize),
197    /// Release this exact sequence of values from the front of the buffer.
198    Values(Vec<T>),
199    /// Release everything that has arrived by the time the tick fires.
200    All,
201}
202
203impl<T> ScriptDecision for BatchDecision<T>
204where
205    T: serde::Serialize + serde::de::DeserializeOwned,
206{
207    fn describe(&self) -> String {
208        match self {
209            BatchDecision::Prefix(n) => format!("release({})", n),
210            BatchDecision::Values(values) => {
211                format!("release_values({} value(s))", values.len())
212            }
213            BatchDecision::All => "release_all()".to_owned(),
214        }
215    }
216}
217
218/// A scripted decision for an unordered batch hook.
219#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
220pub enum UnorderedBatchDecision<T> {
221    /// Release this multiset of values. Duplicate values name duplicate buffered items.
222    Values(Vec<T>),
223    /// Release everything that has arrived by the time the tick fires.
224    All,
225}
226
227impl<T> ScriptDecision for UnorderedBatchDecision<T>
228where
229    T: serde::Serialize + serde::de::DeserializeOwned,
230{
231    fn describe(&self) -> String {
232        match self {
233            UnorderedBatchDecision::Values(values) => {
234                format!("release_values({} value(s))", values.len())
235            }
236            UnorderedBatchDecision::All => "release_all()".to_owned(),
237        }
238    }
239}
240
241/// The pending-input view a batch hook reports to its test-side handle (see
242/// [`ScriptableHook::status`]), used by `pause_until_*` predicates.
243#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
244pub struct BatchStatus {
245    /// The number of buffered elements.
246    pub buffered: usize,
247}
248
249impl<T> ScriptableHook for StreamHook<T, TotalOrder>
250where
251    T: serde::Serialize + serde::de::DeserializeOwned + PartialEq,
252{
253    type Decision = BatchDecision<T>;
254    type Status = BatchStatus;
255
256    fn is_honorable(&self, decision: &BatchDecision<T>) -> Result<bool, String> {
257        let input = self.input.borrow();
258        match decision {
259            BatchDecision::Prefix(n) => Ok(input.len() >= *n),
260            BatchDecision::Values(values) => {
261                if let Some(idx) = input
262                    .iter()
263                    .zip(values)
264                    .position(|(buffered, expected)| buffered != expected)
265                {
266                    Err(format!(
267                        "release_values: buffered item at prefix position {} did not match the expected value",
268                        idx
269                    ))
270                } else {
271                    Ok(input.len() >= values.len())
272                }
273            }
274            BatchDecision::All => Ok(true),
275        }
276    }
277
278    fn apply(&mut self, decision: BatchDecision<T>) {
279        let mut input = self.input.borrow_mut();
280        let out: Vec<T> = match decision {
281            BatchDecision::Prefix(n) => input.drain(0..n).collect(),
282            BatchDecision::Values(values) => input.drain(0..values.len()).collect(),
283            BatchDecision::All => input.drain(..).collect(),
284        };
285
286        self.to_release = Some(out);
287    }
288
289    fn implicit(&mut self) {
290        self.to_release = Some(vec![]);
291    }
292
293    fn status(&self) -> BatchStatus {
294        BatchStatus {
295            buffered: self.input.borrow().len(),
296        }
297    }
298
299    fn describe_pending(&self) -> Option<String> {
300        let input = self.input.borrow();
301        (!input.is_empty()).then(|| {
302            format!(
303                "{} buffered item(s): {:?}",
304                input.len(),
305                TruncatedVecDebug(RefCell::new(Some(input.iter())), 8, self.format_item_debug)
306            )
307        })
308    }
309}
310
311impl<T> ScriptableTickInputHook for StreamHook<T, TotalOrder>
312where
313    T: serde::Serialize + serde::de::DeserializeOwned + PartialEq,
314{
315    fn decision_triggers_tick(&self, decision: &BatchDecision<T>) -> bool {
316        match decision {
317            BatchDecision::Prefix(n) => *n > 0,
318            BatchDecision::Values(values) => !values.is_empty(),
319            BatchDecision::All => !self.input.borrow().is_empty(),
320        }
321    }
322}
323
324impl<T> ScriptableHook for StreamHook<T, NoOrder>
325where
326    T: serde::Serialize + serde::de::DeserializeOwned + PartialEq,
327{
328    type Decision = UnorderedBatchDecision<T>;
329    type Status = BatchStatus;
330
331    fn is_honorable(&self, decision: &UnorderedBatchDecision<T>) -> Result<bool, String> {
332        Ok(match decision {
333            UnorderedBatchDecision::Values(values) => {
334                let mut unmatched: Vec<&T> = values.iter().collect();
335                for buffered in self.input.borrow().iter() {
336                    if let Some(idx) = unmatched
337                        .iter()
338                        .position(|requested| *requested == buffered)
339                    {
340                        unmatched.swap_remove(idx);
341                    }
342                }
343                unmatched.is_empty()
344            }
345            UnorderedBatchDecision::All => true,
346        })
347    }
348
349    fn apply(&mut self, decision: UnorderedBatchDecision<T>) {
350        let mut input = self.input.borrow_mut();
351        let out: Vec<T> = match decision {
352            UnorderedBatchDecision::Values(mut values) => {
353                let (selected, remaining): (Vec<_>, Vec<_>) =
354                    input.drain(..).partition(|buffered| {
355                        values
356                            .iter()
357                            .position(|requested| requested == buffered)
358                            .is_some_and(|idx| {
359                                values.swap_remove(idx);
360                                true
361                            })
362                    });
363                assert!(
364                    values.is_empty(),
365                    "scripted unordered batch decision was not honorable"
366                );
367                *input = remaining.into();
368                selected
369            }
370            UnorderedBatchDecision::All => input.drain(..).collect(),
371        };
372
373        self.to_release = Some(out);
374    }
375
376    fn implicit(&mut self) {
377        self.to_release = Some(vec![]);
378    }
379
380    fn status(&self) -> BatchStatus {
381        BatchStatus {
382            buffered: self.input.borrow().len(),
383        }
384    }
385
386    fn describe_pending(&self) -> Option<String> {
387        let input = self.input.borrow();
388        (!input.is_empty()).then(|| {
389            format!(
390                "{} buffered item(s): {:?}",
391                input.len(),
392                TruncatedVecDebug(RefCell::new(Some(input.iter())), 8, self.format_item_debug)
393            )
394        })
395    }
396}
397
398impl<T> ScriptableTickInputHook for StreamHook<T, NoOrder>
399where
400    T: serde::Serialize + serde::de::DeserializeOwned + PartialEq,
401{
402    fn decision_triggers_tick(&self, decision: &UnorderedBatchDecision<T>) -> bool {
403        match decision {
404            UnorderedBatchDecision::Values(values) => !values.is_empty(),
405            UnorderedBatchDecision::All => !self.input.borrow().is_empty(),
406        }
407    }
408}
409
410pub struct KeyedStreamHook<K: Hash + Eq + Clone, V, Order: Ordering> {
411    pub input: Rc<RefCell<FxHashMap<K, VecDeque<V>>>>, // FxHasher is deterministic
412    pub to_release: Option<Vec<(K, V)>>,
413    pub output: Sender<(K, V)>,
414    pub batch_location: HookLocationMeta,
415    pub format_item_debug: fn(&(K, V)) -> Option<String>,
416    pub _order: std::marker::PhantomData<Order>,
417}
418
419impl<K: Hash + Eq + Clone, V> RuntimeHook for KeyedStreamHook<K, V, TotalOrder> {
420    fn has_pending_input(&self) -> bool {
421        #[expect(clippy::disallowed_methods, reason = "FxHasher is deterministic")]
422        !self.input.borrow().values().all(|q| q.is_empty())
423    }
424
425    fn only_one_possible_decision(&self) -> bool {
426        // One buffered element is still a real choice: release it or not.
427        #[expect(clippy::disallowed_methods, reason = "FxHasher is deterministic")]
428        self.input.borrow().values().all(|q| q.is_empty())
429    }
430
431    fn release_decision(&mut self, log_writer: Option<&mut dyn std::fmt::Write>) {
432        if let Some(to_release) = self.to_release.take() {
433            if let Some(log_writer) = log_writer {
434                let HookLocationMeta {
435                    location: batch_location,
436                    line,
437                    caret_indent,
438                } = self.batch_location;
439                let note_str = if to_release.is_empty() {
440                    "^ releasing no items".to_owned()
441                } else {
442                    format!(
443                        "^ releasing items: {:?}",
444                        TruncatedVecDebug(
445                            RefCell::new(Some(to_release.iter())),
446                            8,
447                            self.format_item_debug
448                        )
449                    )
450                };
451
452                log_release(
453                    log_writer,
454                    batch_location,
455                    line,
456                    caret_indent,
457                    &note_str,
458                    colored::Color::Green,
459                );
460            }
461
462            for item in to_release {
463                self.output.try_send(item).unwrap();
464            }
465        } else {
466            panic!("No decision to release");
467        }
468    }
469
470    fn location_meta(&self) -> HookLocationMeta {
471        self.batch_location
472    }
473}
474
475impl<K: Hash + Eq + Clone, V> TickInputHook for KeyedStreamHook<K, V, TotalOrder> {
476    fn can_trigger_tick(&self) -> bool {
477        #[expect(clippy::disallowed_methods, reason = "FxHasher is deterministic")]
478        !self.input.borrow().values().all(|q| q.is_empty())
479    }
480
481    fn autonomous_decision<'a>(
482        &mut self,
483        driver: &mut Borrowed<'a>,
484        mut force_trigger: bool,
485    ) -> bool {
486        let mut current_input = self.input.borrow_mut();
487        self.to_release = Some(vec![]);
488        #[expect(clippy::disallowed_methods, reason = "FxHasher is deterministic")]
489        let nonempty_key_count = current_input.values().filter(|q| !q.is_empty()).count();
490
491        let mut remaining_nonempty_keys = nonempty_key_count;
492        #[expect(clippy::disallowed_methods, reason = "FxHasher is deterministic")]
493        for (key, queue) in current_input.iter_mut() {
494            if queue.is_empty() {
495                continue;
496            }
497
498            remaining_nonempty_keys -= 1;
499
500            let count = ((if force_trigger && remaining_nonempty_keys == 0 {
501                1
502            } else {
503                0
504            })..=queue.len())
505                .generate(driver)
506                .unwrap();
507
508            let items: Vec<(K, V)> = queue.drain(0..count).map(|v| (key.clone(), v)).collect();
509            self.to_release.as_mut().unwrap().extend(items);
510
511            if count > 0 {
512                force_trigger = false;
513            }
514        }
515
516        !self.to_release.as_ref().unwrap().is_empty()
517    }
518}
519
520impl<K: Hash + Eq + Clone, V> RuntimeHook for KeyedStreamHook<K, V, NoOrder> {
521    fn has_pending_input(&self) -> bool {
522        #[expect(clippy::disallowed_methods, reason = "FxHasher is deterministic")]
523        !self.input.borrow().values().all(|q| q.is_empty())
524    }
525
526    fn only_one_possible_decision(&self) -> bool {
527        // One buffered element is still a real choice: release it or not.
528        #[expect(clippy::disallowed_methods, reason = "FxHasher is deterministic")]
529        self.input.borrow().values().all(|q| q.is_empty())
530    }
531
532    fn release_decision(&mut self, log_writer: Option<&mut dyn std::fmt::Write>) {
533        if let Some(to_release) = self.to_release.take() {
534            if let Some(log_writer) = log_writer {
535                let HookLocationMeta {
536                    location: batch_location,
537                    line,
538                    caret_indent,
539                } = self.batch_location;
540                let note_str = if to_release.is_empty() {
541                    "^ releasing no items".to_owned()
542                } else {
543                    format!(
544                        "^ releasing unordered items: {:?}",
545                        TruncatedVecDebug(
546                            RefCell::new(Some(to_release.iter())),
547                            8,
548                            self.format_item_debug
549                        )
550                    )
551                };
552
553                log_release(
554                    log_writer,
555                    batch_location,
556                    line,
557                    caret_indent,
558                    &note_str,
559                    colored::Color::Green,
560                );
561            }
562
563            for item in to_release {
564                self.output.try_send(item).unwrap();
565            }
566        } else {
567            panic!("No decision to release");
568        }
569    }
570
571    fn location_meta(&self) -> HookLocationMeta {
572        self.batch_location
573    }
574}
575
576impl<K: Hash + Eq + Clone, V> TickInputHook for KeyedStreamHook<K, V, NoOrder> {
577    fn can_trigger_tick(&self) -> bool {
578        #[expect(clippy::disallowed_methods, reason = "FxHasher is deterministic")]
579        !self.input.borrow().values().all(|q| q.is_empty())
580    }
581
582    fn autonomous_decision<'a>(
583        &mut self,
584        driver: &mut Borrowed<'a>,
585        mut force_trigger: bool,
586    ) -> bool {
587        let mut current_input = self.input.borrow_mut();
588        self.to_release = Some(vec![]);
589        #[expect(clippy::disallowed_methods, reason = "FxHasher is deterministic")]
590        let nonempty_key_count = current_input.values().filter(|q| !q.is_empty()).count();
591
592        let mut remaining_nonempty_keys = nonempty_key_count;
593        #[expect(clippy::disallowed_methods, reason = "FxHasher is deterministic")]
594        for (key, queue) in current_input.iter_mut() {
595            if queue.is_empty() {
596                continue;
597            }
598
599            remaining_nonempty_keys -= 1;
600
601            let mut min_index = 0;
602            while !queue.is_empty() {
603                let must_release = force_trigger && remaining_nonempty_keys == 0;
604                if !must_release && produce().generate(driver).unwrap() {
605                    break;
606                }
607
608                let idx = (min_index..queue.len()).generate(driver).unwrap();
609                let item = queue.remove(idx).unwrap();
610                self.to_release.as_mut().unwrap().push((key.clone(), item));
611                force_trigger = false;
612
613                min_index = idx;
614                // Next time, only consider items at or after this index. The reason this is safe is
615                // because batching a `NoOrder` stream results in batches with a `NoOrder` guarantee.
616                // Therefore, simulating different order of elements _within_ a batch is redundant.
617
618                if min_index == queue.len() {
619                    break;
620                }
621            }
622        }
623
624        !self.to_release.as_ref().unwrap().is_empty()
625    }
626}
627
628pub struct SingletonHook<T> {
629    input: Rc<RefCell<VecDeque<T>>>,
630    to_release: Option<(T, bool)>, // (data, is new)
631    last_released: Option<T>,
632    skipped_states: Vec<T>,
633    output: Sender<T>,
634    batch_location: HookLocationMeta,
635    format_item_debug: fn(&T) -> Option<String>,
636}
637
638impl<T: Clone> SingletonHook<T> {
639    pub fn new(
640        input: Rc<RefCell<VecDeque<T>>>,
641        output: Sender<T>,
642        batch_location: HookLocationMeta,
643        format_item_debug: fn(&T) -> Option<String>,
644    ) -> Self {
645        Self {
646            input,
647            to_release: None,
648            last_released: None,
649            skipped_states: vec![],
650            output,
651            batch_location,
652            format_item_debug,
653        }
654    }
655}
656
657impl<T: Clone> RuntimeHook for SingletonHook<T> {
658    fn has_pending_input(&self) -> bool {
659        !self.input.borrow().is_empty()
660    }
661
662    fn only_one_possible_decision(&self) -> bool {
663        // With no previously revealed value, a sole buffered version has exactly one
664        // resolution: reveal it (there is nothing to keep). Once a value has been
665        // revealed, any buffered version is a real choice (keep vs reveal); with two or
666        // more buffered versions the choice of which to reveal is real either way.
667        let input = self.input.borrow();
668        input.is_empty() || (input.len() == 1 && self.last_released.is_none())
669    }
670
671    fn release_decision(&mut self, log_writer: Option<&mut dyn std::fmt::Write>) {
672        if let Some((to_release, is_new)) = self.to_release.take() {
673            self.last_released = Some(to_release.clone());
674
675            if let Some(log_writer) = log_writer {
676                let HookLocationMeta {
677                    location: batch_location,
678                    line,
679                    caret_indent,
680                } = self.batch_location;
681                let note_str = if self.skipped_states.is_empty() {
682                    if is_new {
683                        format!(
684                            "^ releasing snapshot: {:?}",
685                            ManualDebug(&to_release, self.format_item_debug)
686                        )
687                    } else {
688                        format!(
689                            "^ releasing unchanged snapshot: {:?}",
690                            ManualDebug(&to_release, self.format_item_debug)
691                        )
692                    }
693                } else {
694                    format!(
695                        "^ releasing snapshot: {:?} (skipping earlier states: {:?})",
696                        ManualDebug(&to_release, self.format_item_debug),
697                        self.skipped_states
698                            .iter()
699                            .map(|s| ManualDebug(s, self.format_item_debug))
700                            .collect::<Vec<_>>()
701                    )
702                };
703
704                log_release(
705                    log_writer,
706                    batch_location,
707                    line,
708                    caret_indent,
709                    &note_str,
710                    colored::Color::Green,
711                );
712            }
713
714            self.output.try_send(to_release).unwrap();
715        } else {
716            panic!("No decision to release");
717        }
718    }
719
720    fn location_meta(&self) -> HookLocationMeta {
721        self.batch_location
722    }
723}
724
725impl<T: Clone> TickInputHook for SingletonHook<T> {
726    fn can_trigger_tick(&self) -> bool {
727        // TODO(mingwei): Singletons/Optionals will soon not trigger a tick, so then this will always return false.
728        !self.input.borrow().is_empty()
729    }
730
731    fn autonomous_decision<'a>(&mut self, driver: &mut Borrowed<'a>, force_trigger: bool) -> bool {
732        let mut current_input = self.input.borrow_mut();
733        if current_input.is_empty() {
734            if force_trigger {
735                panic!("Cannot make a triggering decision when there is no input");
736            }
737
738            if let Some(last) = &self.last_released {
739                // Re-release the last item
740                self.to_release = Some((last.clone(), false));
741                false
742            } else {
743                panic!("No input and no last released item to re-release");
744            }
745        } else if !force_trigger
746            && let Some(last) = &self.last_released
747            && produce().generate(driver).unwrap()
748        {
749            // Re-release the last item
750            self.to_release = Some((last.clone(), false));
751            false
752        } else {
753            // Release a new item
754            let idx_to_release = (0..current_input.len()).generate(driver).unwrap();
755            self.skipped_states = current_input.drain(0..idx_to_release).collect(); // Drop earlier items
756            let item = current_input.pop_front().unwrap();
757            self.to_release = Some((item, true));
758            true
759        }
760    }
761}
762
763/// A hook for batching / snapshotting an [`Optional`](crate::live_collections::Optional) with
764/// `InitNone` boundedness into a tick.
765///
766/// This is the [`SingletonHook`] analog for optionals whose *presence* is monotone (the
767/// `InitNone` bound): the optional starts null and, once it becomes non-null, stays non-null.
768/// It differs from [`SingletonHook`] in that its released value is optional. Before the first
769/// non-null value it releases *null* (sending nothing into the tick's `source_stream`, so the
770/// downstream optional is empty). Once it has released a value, presence is monotone, so it only
771/// ever re-releases or advances to a newer value — never back to null.
772pub struct OptionalInitNoneHook<T> {
773    input: Rc<RefCell<VecDeque<T>>>,
774    to_release: Option<(Option<T>, bool)>, // (value or null, is new)
775    last_released: Option<T>,              // last non-null value released (None until the first)
776    skipped_states: Vec<T>,
777    output: Sender<T>,
778    batch_location: HookLocationMeta,
779    format_item_debug: fn(&T) -> Option<String>,
780}
781
782impl<T: Clone> OptionalInitNoneHook<T> {
783    pub fn new(
784        input: Rc<RefCell<VecDeque<T>>>,
785        output: Sender<T>,
786        batch_location: HookLocationMeta,
787        format_item_debug: fn(&T) -> Option<String>,
788    ) -> Self {
789        Self {
790            input,
791            to_release: None,
792            last_released: None,
793            skipped_states: vec![],
794            output,
795            batch_location,
796            format_item_debug,
797        }
798    }
799}
800
801impl<T: Clone> RuntimeHook for OptionalInitNoneHook<T> {
802    fn has_pending_input(&self) -> bool {
803        !self.input.borrow().is_empty()
804    }
805
806    fn only_one_possible_decision(&self) -> bool {
807        // If there is no input, the decision is trivial.
808        // Even if the input is a single item, we can always either take it or leave it, and
809        // remain at the existing value (which may be `None`).
810        self.input.borrow().is_empty()
811    }
812
813    fn release_decision(&mut self, log_writer: Option<&mut dyn std::fmt::Write>) {
814        let Some((to_release, is_new)) = self.to_release.take() else {
815            panic!("No decision to release");
816        };
817
818        if let Some(value) = &to_release {
819            self.last_released = Some(value.clone());
820        }
821
822        if let Some(log_writer) = log_writer {
823            let HookLocationMeta {
824                location: batch_location,
825                line,
826                caret_indent,
827            } = self.batch_location;
828            let note_str = match (&to_release, is_new) {
829                (None, _) => "^ releasing null snapshot".to_owned(),
830                (Some(value), true) => {
831                    if self.skipped_states.is_empty() {
832                        format!(
833                            "^ releasing snapshot: {:?}",
834                            ManualDebug(value, self.format_item_debug)
835                        )
836                    } else {
837                        format!(
838                            "^ releasing snapshot: {:?} (skipping earlier states: {:?})",
839                            ManualDebug(value, self.format_item_debug),
840                            self.skipped_states
841                                .iter()
842                                .map(|s| ManualDebug(s, self.format_item_debug))
843                                .collect::<Vec<_>>()
844                        )
845                    }
846                }
847                (Some(value), false) => format!(
848                    "^ releasing unchanged snapshot: {:?}",
849                    ManualDebug(value, self.format_item_debug)
850                ),
851            };
852
853            log_release(
854                log_writer,
855                batch_location,
856                line,
857                caret_indent,
858                &note_str,
859                colored::Color::Green,
860            );
861        }
862
863        if let Some(value) = to_release {
864            self.output.try_send(value).unwrap();
865        }
866    }
867
868    fn location_meta(&self) -> HookLocationMeta {
869        self.batch_location
870    }
871}
872
873impl<T: Clone> TickInputHook for OptionalInitNoneHook<T> {
874    fn can_trigger_tick(&self) -> bool {
875        // TODO(mingwei): Singletons/Optionals will soon not trigger a tick, so then this will always return false.
876        // Only advancing to a new value is nontrivial; releasing null (or re-releasing the
877        // latest value) does not, by itself, drive a tick.
878        !self.input.borrow().is_empty()
879    }
880
881    fn autonomous_decision<'a>(
882        &mut self,
883        driver: &mut Borrowed<'a>,
884        force_nontrivial: bool,
885    ) -> bool {
886        let mut current_input = self.input.borrow_mut();
887        if current_input.is_empty() {
888            // Case 1 (trivial): No input.
889            if force_nontrivial {
890                panic!("Cannot make nontrivial decision when there is no input");
891            }
892
893            if let Some(last) = &self.last_released {
894                // Presence is monotone: once non-null, re-release the latest value.
895                self.to_release = Some((Some(last.clone()), false));
896            } else {
897                // Still in the initial-null prefix.
898                self.to_release = Some((None, false));
899            }
900            false
901        } else if !force_nontrivial && produce().generate(driver).unwrap() {
902            // Case 2 (trivial): Keep latest value (may be `Some` or `None`)
903            if let Some(last) = &self.last_released {
904                // Already non-null; re-release the latest value (models snapshot lag).
905                self.to_release = Some((Some(last.clone()), false));
906            } else {
907                // Still in the initial-null prefix even though a value is buffered: models a
908                // snapshot that does not yet include the first value.
909                self.to_release = Some((None, false));
910            }
911            false
912        } else {
913            // Case 3 (non-trivial): Advance to new value.
914            let idx_to_release = (0..current_input.len()).generate(driver).unwrap();
915            self.skipped_states = current_input.drain(0..idx_to_release).collect(); // Drop earlier items
916            let item = current_input.pop_front().unwrap();
917            self.to_release = Some((Some(item), true));
918            true
919        }
920    }
921}
922
923/// A scripted decision for a snapshot hook: which buffered version of the state the next
924/// tick execution observes.
925#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
926pub enum SnapshotDecision<T> {
927    /// Reveal the first buffered version equal to this value, skipping over earlier
928    /// versions.
929    Reveal(T),
930    /// Advance to the next buffered version.
931    RevealNext,
932    /// Reveal the newest version that has arrived by the time the tick fires.
933    RevealLatest,
934    /// Observe the previously revealed version again.
935    Keep,
936}
937
938impl<T> ScriptDecision for SnapshotDecision<T>
939where
940    T: serde::Serialize + serde::de::DeserializeOwned,
941{
942    fn describe(&self) -> String {
943        match self {
944            SnapshotDecision::Reveal(_) => "reveal(..)".to_owned(),
945            SnapshotDecision::RevealNext => "reveal_next()".to_owned(),
946            SnapshotDecision::RevealLatest => "reveal_latest()".to_owned(),
947            SnapshotDecision::Keep => "keep()".to_owned(),
948        }
949    }
950}
951
952/// The pending-input view a snapshot hook reports to its test-side handle (see
953/// [`ScriptableHook::status`]), used by `pause_until_*` predicates.
954#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
955pub struct SnapshotStatus {
956    /// The number of buffered versions newer than the last revealed one.
957    pub newer_versions: usize,
958}
959
960impl<T: Clone + PartialEq + serde::Serialize + serde::de::DeserializeOwned> ScriptableHook
961    for SingletonHook<T>
962{
963    type Decision = SnapshotDecision<T>;
964    type Status = SnapshotStatus;
965
966    fn is_honorable(&self, decision: &SnapshotDecision<T>) -> Result<bool, String> {
967        let input = self.input.borrow();
968        Ok(match decision {
969            SnapshotDecision::Reveal(target) => input.iter().any(|version| version == target),
970            SnapshotDecision::RevealNext => !input.is_empty(),
971            SnapshotDecision::RevealLatest => !input.is_empty() || self.last_released.is_some(),
972            SnapshotDecision::Keep => self.last_released.is_some(),
973        })
974    }
975
976    fn apply(&mut self, decision: SnapshotDecision<T>) {
977        match decision {
978            SnapshotDecision::Reveal(target) => {
979                let mut input = self.input.borrow_mut();
980                let idx = input.iter().position(|version| *version == target).unwrap();
981                self.skipped_states = input.drain(0..idx).collect();
982                let item = input.pop_front().unwrap();
983                self.to_release = Some((item, true));
984            }
985            SnapshotDecision::RevealNext => {
986                let mut input = self.input.borrow_mut();
987                self.skipped_states = vec![];
988                let item = input.pop_front().unwrap();
989                self.to_release = Some((item, true));
990            }
991            SnapshotDecision::RevealLatest => {
992                let mut input = self.input.borrow_mut();
993                if input.is_empty() {
994                    self.skipped_states = vec![];
995                    self.to_release = Some((self.last_released.clone().unwrap(), false));
996                } else {
997                    let skip_count = input.len() - 1;
998                    self.skipped_states = input.drain(0..skip_count).collect();
999                    let item = input.pop_front().unwrap();
1000                    self.to_release = Some((item, true));
1001                }
1002            }
1003            SnapshotDecision::Keep => {
1004                self.skipped_states = vec![];
1005                self.to_release = Some((self.last_released.clone().unwrap(), false));
1006            }
1007        }
1008    }
1009
1010    fn implicit(&mut self) {
1011        if let Some(last) = &self.last_released {
1012            self.skipped_states = vec![];
1013            self.to_release = Some((last.clone(), false));
1014        } else {
1015            // `is_ready()` prevents the tick from running before the singleton has a
1016            // value, so this is unreachable.
1017            abort!("scripted snapshot hook asked for implicit behavior with no revealed value");
1018        }
1019    }
1020
1021    fn status(&self) -> SnapshotStatus {
1022        SnapshotStatus {
1023            newer_versions: self.input.borrow().len(),
1024        }
1025    }
1026
1027    fn describe_pending(&self) -> Option<String> {
1028        let input = self.input.borrow();
1029        (!input.is_empty()).then(|| {
1030            format!(
1031                "{} buffered version(s): {:?}",
1032                input.len(),
1033                TruncatedVecDebug(RefCell::new(Some(input.iter())), 8, self.format_item_debug)
1034            )
1035        })
1036    }
1037}
1038
1039impl<T: Clone + PartialEq + serde::Serialize + serde::de::DeserializeOwned> ScriptableTickInputHook
1040    for SingletonHook<T>
1041{
1042    fn decision_triggers_tick(&self, decision: &SnapshotDecision<T>) -> bool {
1043        match decision {
1044            SnapshotDecision::Reveal(_) | SnapshotDecision::RevealNext => true,
1045            SnapshotDecision::RevealLatest => !self.input.borrow().is_empty(),
1046            SnapshotDecision::Keep => false,
1047        }
1048    }
1049}
1050/// A passthrough singleton hook for fold outputs that are already controlled by a
1051/// `TopLevelFoldHook`. Always releases the latest value without any non-deterministic
1052/// decisions, since the fold hook already made the only meaningful choice (which subset
1053/// of inputs to process).
1054pub struct PassthroughSingletonHook<T> {
1055    input: Rc<RefCell<VecDeque<T>>>,
1056    to_release: Option<T>,
1057    output: Sender<T>,
1058    batch_location: HookLocationMeta,
1059    format_item_debug: fn(&T) -> Option<String>,
1060}
1061
1062impl<T> PassthroughSingletonHook<T> {
1063    pub fn new(
1064        input: Rc<RefCell<VecDeque<T>>>,
1065        output: Sender<T>,
1066        batch_location: HookLocationMeta,
1067        format_item_debug: fn(&T) -> Option<String>,
1068    ) -> Self {
1069        Self {
1070            input,
1071            to_release: None,
1072            output,
1073            batch_location,
1074            format_item_debug,
1075        }
1076    }
1077}
1078
1079impl<T> RuntimeHook for PassthroughSingletonHook<T> {
1080    fn has_pending_input(&self) -> bool {
1081        !self.input.borrow().is_empty()
1082    }
1083
1084    fn only_one_possible_decision(&self) -> bool {
1085        // Releasing the latest value is the only behavior this hook ever has: the
1086        // controlling `TopLevelFoldHook` already made every meaningful choice, so even
1087        // with buffered input there is nothing non-deterministic left to decide. (This
1088        // hook is why the choice question is separate from `can_trigger_tick`.)
1089        true
1090    }
1091
1092    fn release_decision(&mut self, log_writer: Option<&mut dyn std::fmt::Write>) {
1093        if let Some(to_release) = self.to_release.take() {
1094            if let Some(log_writer) = log_writer {
1095                let HookLocationMeta {
1096                    location: batch_location,
1097                    line,
1098                    caret_indent,
1099                } = self.batch_location;
1100                let note_str = format!(
1101                    "^ releasing snapshot: {:?}",
1102                    ManualDebug(&to_release, self.format_item_debug)
1103                );
1104
1105                log_release(
1106                    log_writer,
1107                    batch_location,
1108                    line,
1109                    caret_indent,
1110                    &note_str,
1111                    colored::Color::Green,
1112                );
1113            }
1114
1115            self.output.try_send(to_release).unwrap();
1116        } else {
1117            panic!("No decision to release");
1118        }
1119    }
1120
1121    fn location_meta(&self) -> HookLocationMeta {
1122        self.batch_location
1123    }
1124}
1125
1126impl<T> TickInputHook for PassthroughSingletonHook<T> {
1127    fn can_trigger_tick(&self) -> bool {
1128        !self.input.borrow().is_empty()
1129    }
1130
1131    fn autonomous_decision<'a>(
1132        &mut self,
1133        _driver: &mut Borrowed<'a>,
1134        _force_trigger: bool,
1135    ) -> bool {
1136        let mut current_input = self.input.borrow_mut();
1137        // Always take the last (most recent) value, discard intermediates.
1138        if let Some(item) = current_input.pop_back() {
1139            current_input.clear();
1140            self.to_release = Some(item);
1141            true
1142        } else {
1143            false
1144        }
1145    }
1146}
1147
1148pub struct KeyedSingletonHook<K: Hash + Eq + Clone, V: Clone> {
1149    input: Rc<RefCell<FxHashMap<K, VecDeque<V>>>>, // FxHasher is deterministic
1150    to_release: Option<Vec<(K, V, bool)>>,         // (key, data, is new)
1151    last_released: FxHashMap<K, V>,
1152    skipped_states: FxHashMap<K, Vec<V>>,
1153    output: Sender<(K, V)>,
1154    batch_location: HookLocationMeta,
1155    format_key_debug: fn(&K) -> Option<String>,
1156    format_value_debug: fn(&V) -> Option<String>,
1157}
1158
1159impl<K: Hash + Eq + Clone, V: Clone> KeyedSingletonHook<K, V> {
1160    pub fn new(
1161        input: Rc<RefCell<FxHashMap<K, VecDeque<V>>>>,
1162        output: Sender<(K, V)>,
1163        batch_location: HookLocationMeta,
1164        format_key_debug: fn(&K) -> Option<String>,
1165        format_value_debug: fn(&V) -> Option<String>,
1166    ) -> Self {
1167        Self {
1168            input,
1169            to_release: None,
1170            last_released: FxHashMap::default(),
1171            skipped_states: FxHashMap::default(),
1172            output,
1173            batch_location,
1174            format_key_debug,
1175            format_value_debug,
1176        }
1177    }
1178}
1179
1180impl<K: Hash + Eq + Clone, V: Clone> RuntimeHook for KeyedSingletonHook<K, V> {
1181    fn has_pending_input(&self) -> bool {
1182        #[expect(clippy::disallowed_methods, reason = "FxHasher is deterministic")]
1183        !self.input.borrow().values().all(|q| q.is_empty())
1184    }
1185
1186    fn only_one_possible_decision(&self) -> bool {
1187        // Even a sole buffered version for a key admits two resolutions (a key not yet
1188        // in the snapshot may stay withheld; a key with a previous value may keep it).
1189        #[expect(clippy::disallowed_methods, reason = "FxHasher is deterministic")]
1190        self.input.borrow().values().all(|q| q.is_empty())
1191    }
1192
1193    fn release_decision(&mut self, log_writer: Option<&mut dyn std::fmt::Write>) {
1194        if let Some(to_release) = self.to_release.take() {
1195            if let Some(log_writer) = log_writer {
1196                let HookLocationMeta {
1197                    location: batch_location,
1198                    line,
1199                    caret_indent,
1200                } = self.batch_location;
1201                let note_str = if to_release.is_empty() {
1202                    "^ releasing no items".to_owned()
1203                } else {
1204                    let mut mapping_text = String::new();
1205                    for (key, value, is_new) in &to_release {
1206                        let entry_text = if *is_new {
1207                            format!(
1208                                "{:?}: {:?}",
1209                                ManualDebug(key, self.format_key_debug),
1210                                ManualDebug(value, self.format_value_debug)
1211                            )
1212                        } else {
1213                            format!(
1214                                "{:?}: {:?} (unchanged)",
1215                                ManualDebug(key, self.format_key_debug),
1216                                ManualDebug(value, self.format_value_debug)
1217                            )
1218                        };
1219                        if !mapping_text.is_empty() {
1220                            mapping_text.push_str(", ");
1221                        }
1222                        mapping_text.push_str(&entry_text);
1223                    }
1224                    format!("^ releasing items: {{ {} }}", mapping_text)
1225                };
1226
1227                log_release(
1228                    log_writer,
1229                    batch_location,
1230                    line,
1231                    caret_indent,
1232                    &note_str,
1233                    colored::Color::Green,
1234                );
1235            }
1236
1237            for (key, value, _) in to_release {
1238                self.output.try_send((key, value)).unwrap();
1239            }
1240        } else {
1241            panic!("No decision to release");
1242        }
1243    }
1244
1245    fn location_meta(&self) -> HookLocationMeta {
1246        self.batch_location
1247    }
1248}
1249
1250impl<K: Hash + Eq + Clone, V: Clone> TickInputHook for KeyedSingletonHook<K, V> {
1251    fn can_trigger_tick(&self) -> bool {
1252        #[expect(clippy::disallowed_methods, reason = "FxHasher is deterministic")]
1253        !self.input.borrow().values().all(|q| q.is_empty())
1254    }
1255
1256    fn autonomous_decision<'a>(
1257        &mut self,
1258        driver: &mut Borrowed<'a>,
1259        mut force_trigger: bool,
1260    ) -> bool {
1261        let mut current_input = self.input.borrow_mut();
1262        self.to_release = Some(vec![]);
1263        #[expect(clippy::disallowed_methods, reason = "FxHasher is deterministic")]
1264        let nonempty_key_count = current_input.values().filter(|q| !q.is_empty()).count();
1265
1266        let mut remaining_nonempty_keys = nonempty_key_count;
1267        let mut any_triggered = false;
1268        #[expect(clippy::disallowed_methods, reason = "FxHasher is deterministic")]
1269        for (key, queue) in current_input.iter_mut() {
1270            if queue.is_empty() {
1271                self.to_release.as_mut().unwrap().push((
1272                    key.clone(),
1273                    self.last_released.get(key).unwrap().clone(),
1274                    false,
1275                ));
1276
1277                continue;
1278            }
1279
1280            remaining_nonempty_keys -= 1;
1281
1282            let must_reveal = force_trigger && remaining_nonempty_keys == 0;
1283
1284            if !must_reveal
1285                && self.last_released.contains_key(key)
1286                && produce().generate(driver).unwrap()
1287            {
1288                // Re-release the last item for this key
1289                let last = self.last_released.get(key).unwrap().clone();
1290                self.to_release
1291                    .as_mut()
1292                    .unwrap()
1293                    .push((key.clone(), last, false));
1294            } else {
1295                let allow_null_release = !must_reveal && !self.last_released.contains_key(key);
1296                if allow_null_release && produce().generate(driver).unwrap() {
1297                    // Don't emit anything, this key is not yet added to the snapshot
1298                    continue;
1299                } else {
1300                    // Release a new item for this key
1301                    let idx_to_release = (0..queue.len()).generate(driver).unwrap();
1302                    let skipped: Vec<V> = queue.drain(0..idx_to_release).collect();
1303                    let item = queue.pop_front().unwrap();
1304                    self.skipped_states.insert(key.clone(), skipped);
1305                    self.to_release
1306                        .as_mut()
1307                        .unwrap()
1308                        .push((key.clone(), item.clone(), true));
1309                    self.last_released.insert(key.clone(), item);
1310
1311                    any_triggered |= true;
1312                    force_trigger = false;
1313                }
1314            }
1315        }
1316
1317        any_triggered
1318    }
1319}
hydro_lang/sim/runtime/tick_input.rs

hydro_lang/sim/runtime/
tick_input.rs
