rill_ml/decision/
mod.rs

1//! Bounded, caller-clocked primitives for decisions with delayed feedback.
2//!
3//! The ledger intentionally knows nothing about product actions or reward
4//! meaning. Callers provide identifiers, timestamps, generations, contexts,
5//! and actions. Pending decisions are never evicted implicitly.
6
7use std::collections::BTreeMap;
8
9use thiserror::Error;
10
11use crate::RillError;
12#[cfg(feature = "serde")]
13use crate::ValidateState;
14
15/// Version of the serialized ledger state introduced with the Preview API.
16pub const DECISION_LEDGER_STATE_VERSION: u32 = 1;
17/// Default maximum serialized-equivalent context size per decision.
18pub const DEFAULT_MAX_CONTEXT_BYTES: usize = 1024 * 1024;
19/// Default maximum serialized-equivalent action size per decision.
20pub const DEFAULT_MAX_ACTION_BYTES: usize = 64 * 1024;
21
22/// Caller-extensible validation hook that makes per-decision memory bounded.
23pub trait BoundedDecisionValue {
24    /// Validate content and reject an estimated representation above `max_bytes`.
25    fn validate_bounded(&self, max_bytes: usize) -> Result<(), DecisionLedgerError>;
26}
27
28impl BoundedDecisionValue for Vec<f64> {
29    fn validate_bounded(&self, max_bytes: usize) -> Result<(), DecisionLedgerError> {
30        if self
31            .len()
32            .checked_mul(std::mem::size_of::<f64>())
33            .is_none_or(|size| size > max_bytes)
34        {
35            return Err(DecisionLedgerError::ValueTooLarge);
36        }
37        if self.iter().any(|value| !value.is_finite()) {
38            return Err(DecisionLedgerError::InvalidValue);
39        }
40        Ok(())
41    }
42}
43
44impl BoundedDecisionValue for Vec<u8> {
45    fn validate_bounded(&self, max_bytes: usize) -> Result<(), DecisionLedgerError> {
46        if self.len() > max_bytes {
47            return Err(DecisionLedgerError::ValueTooLarge);
48        }
49        Ok(())
50    }
51}
52
53impl BoundedDecisionValue for String {
54    fn validate_bounded(&self, max_bytes: usize) -> Result<(), DecisionLedgerError> {
55        if self.len() > max_bytes {
56            return Err(DecisionLedgerError::ValueTooLarge);
57        }
58        Ok(())
59    }
60}
61
62macro_rules! impl_fixed_decision_value {
63    ($($type:ty),+ $(,)?) => {
64        $(impl BoundedDecisionValue for $type {
65            fn validate_bounded(&self, max_bytes: usize) -> Result<(), DecisionLedgerError> {
66                if std::mem::size_of::<$type>() > max_bytes {
67                    return Err(DecisionLedgerError::ValueTooLarge);
68                }
69                Ok(())
70            }
71        })+
72    };
73}
74
75impl_fixed_decision_value!((), bool, usize, u64, u32, u16, u8, i64, i32, i16, i8);
76
77/// Opaque, caller-assigned decision identifier.
78#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
79#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
80pub struct DecisionId(pub u128);
81
82/// Capacity limits for a [`DecisionLedger`].
83#[derive(Debug, Clone, Copy, PartialEq, Eq)]
84#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
85#[cfg_attr(feature = "serde", serde(deny_unknown_fields))]
86#[non_exhaustive]
87pub struct DecisionLedgerConfig {
88    /// Maximum number of incomplete decisions. They are never auto-evicted.
89    pub max_pending: usize,
90    /// Maximum number of completed tombstones. They are never auto-evicted.
91    pub max_completed: usize,
92    /// Maximum bounded-size estimate for one context.
93    pub max_context_bytes: usize,
94    /// Maximum bounded-size estimate for one action.
95    pub max_action_bytes: usize,
96}
97
98impl DecisionLedgerConfig {
99    /// Build explicit pending/completed bounds.
100    pub fn new(max_pending: usize, max_completed: usize) -> Result<Self, DecisionLedgerError> {
101        if max_pending == 0 || max_completed == 0 {
102            return Err(DecisionLedgerError::InvalidCapacity);
103        }
104        Ok(Self {
105            max_pending,
106            max_completed,
107            max_context_bytes: DEFAULT_MAX_CONTEXT_BYTES,
108            max_action_bytes: DEFAULT_MAX_ACTION_BYTES,
109        })
110    }
111
112    /// Override per-entry context/action byte bounds.
113    pub fn with_value_limits(
114        mut self,
115        max_context_bytes: usize,
116        max_action_bytes: usize,
117    ) -> Result<Self, DecisionLedgerError> {
118        if max_context_bytes == 0 || max_action_bytes == 0 {
119            return Err(DecisionLedgerError::InvalidCapacity);
120        }
121        self.max_context_bytes = max_context_bytes;
122        self.max_action_bytes = max_action_bytes;
123        Ok(self)
124    }
125}
126
127impl Default for DecisionLedgerConfig {
128    fn default() -> Self {
129        Self {
130            max_pending: 1_024,
131            max_completed: 4_096,
132            max_context_bytes: DEFAULT_MAX_CONTEXT_BYTES,
133            max_action_bytes: DEFAULT_MAX_ACTION_BYTES,
134        }
135    }
136}
137
138/// Immutable facts recorded at decision time.
139#[derive(Debug, Clone, PartialEq)]
140#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
141#[cfg_attr(feature = "serde", serde(deny_unknown_fields))]
142pub struct PendingDecision<C, A> {
143    /// Unique identifier.
144    pub id: DecisionId,
145    /// Context presented to the decision policy.
146    pub context: C,
147    /// Selected action.
148    pub action: A,
149    /// Caller-supplied decision timestamp.
150    pub created_at: u64,
151    /// Last timestamp at which feedback is accepted, inclusively.
152    pub expires_at: u64,
153    /// Model generation that produced the action.
154    pub model_generation: u64,
155}
156
157impl<C, A> PendingDecision<C, A> {
158    /// Construct immutable decision facts.
159    pub const fn new(
160        id: DecisionId,
161        context: C,
162        action: A,
163        created_at: u64,
164        expires_at: u64,
165        model_generation: u64,
166    ) -> Self {
167        Self {
168            id,
169            context,
170            action,
171            created_at,
172            expires_at,
173            model_generation,
174        }
175    }
176}
177
178/// Delayed outcome submitted by a caller.
179#[derive(Debug, Clone, PartialEq)]
180#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
181#[cfg_attr(feature = "serde", serde(deny_unknown_fields))]
182pub struct DecisionOutcome<A> {
183    /// Decision being completed.
184    pub decision_id: DecisionId,
185    /// Action the outcome belongs to.
186    pub action: A,
187    /// Caller-defined finite reward.
188    pub reward: f64,
189    /// Caller-supplied outcome timestamp.
190    pub observed_at: u64,
191    /// Generation expected by the outcome producer.
192    pub model_generation: u64,
193}
194
195/// Completed decision retained as a bounded replay tombstone.
196#[derive(Debug, Clone, PartialEq)]
197#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
198#[cfg_attr(feature = "serde", serde(deny_unknown_fields))]
199pub struct CompletedDecision<C, A> {
200    /// Original immutable decision facts.
201    pub decision: PendingDecision<C, A>,
202    /// Accepted reward.
203    pub reward: f64,
204    /// Accepted outcome timestamp.
205    pub observed_at: u64,
206}
207
208/// Result of registering a decision.
209#[derive(Debug, Clone, Copy, PartialEq, Eq)]
210pub struct DecisionReceipt {
211    /// Registered or replayed identifier.
212    pub id: DecisionId,
213    /// Whether this call inserted new pending state.
214    pub status: RegistrationStatus,
215}
216
217/// Registration/idempotency status.
218#[derive(Debug, Clone, Copy, PartialEq, Eq)]
219pub enum RegistrationStatus {
220    /// Newly inserted pending decision.
221    Registered,
222    /// Exact replay of an existing pending decision.
223    PendingReplay,
224    /// Exact replay of a decision already represented by a tombstone.
225    CompletedReplay,
226}
227
228/// Result of accepted feedback.
229#[derive(Debug, Clone, Copy, PartialEq, Eq)]
230pub enum FeedbackStatus {
231    /// Feedback was accepted and the decision moved to completed state.
232    Applied,
233}
234
235/// Validation and capacity failures. Every failure leaves the ledger intact.
236#[derive(Debug, Error, Clone, PartialEq)]
237#[non_exhaustive]
238pub enum DecisionLedgerError {
239    /// A configured capacity was zero.
240    #[error("decision ledger capacities must be greater than zero")]
241    InvalidCapacity,
242    /// Decision expiry preceded creation.
243    #[error("decision expiry precedes its creation time")]
244    InvalidTimeRange,
245    /// A reward was NaN or infinite.
246    #[error("decision reward must be finite")]
247    NonFiniteReward,
248    /// Pending storage is full; live decisions are not evicted implicitly.
249    #[error("pending decision capacity is exhausted")]
250    PendingCapacityExceeded,
251    /// Tombstone storage is full; completed entries require explicit cleanup.
252    #[error("completed decision capacity is exhausted")]
253    CompletedCapacityExceeded,
254    /// A context or action exceeds the configured byte estimate.
255    #[error("decision context or action exceeds its configured byte limit")]
256    ValueTooLarge,
257    /// A context or action violates its type-specific semantic invariants.
258    #[error("decision context or action is invalid")]
259    InvalidValue,
260    /// The identifier exists with different immutable facts.
261    #[error("decision id already exists with different facts")]
262    DecisionConflict,
263    /// Feedback references no pending or completed identifier.
264    #[error("feedback references an unknown decision")]
265    UnknownDecision,
266    /// Feedback was already accepted for this identifier.
267    #[error("feedback was already applied")]
268    DuplicateFeedback,
269    /// Outcome timestamp precedes decision creation.
270    #[error("feedback predates the decision")]
271    FeedbackBeforeDecision,
272    /// Outcome timestamp is after the inclusive expiry boundary.
273    #[error("feedback arrived after decision expiry")]
274    FeedbackExpired,
275    /// Outcome generation differs from the recorded generation.
276    #[error("feedback model generation does not match the decision")]
277    GenerationMismatch,
278    /// Outcome action differs from the selected action.
279    #[error("feedback action does not match the decision")]
280    ActionMismatch,
281}
282
283/// Bounded pending decisions plus bounded completed tombstones.
284#[derive(Debug, Clone, PartialEq)]
285#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
286#[cfg_attr(feature = "serde", serde(deny_unknown_fields))]
287pub struct DecisionLedger<C, A> {
288    state_version: u32,
289    config: DecisionLedgerConfig,
290    pending: BTreeMap<DecisionId, PendingDecision<C, A>>,
291    completed: BTreeMap<DecisionId, CompletedDecision<C, A>>,
292}
293
294impl<C, A> DecisionLedger<C, A>
295where
296    C: PartialEq + BoundedDecisionValue,
297    A: PartialEq + BoundedDecisionValue,
298{
299    /// Create an empty ledger with explicit capacity bounds.
300    pub fn new(config: DecisionLedgerConfig) -> Result<Self, DecisionLedgerError> {
301        if config.max_pending == 0
302            || config.max_completed == 0
303            || config.max_context_bytes == 0
304            || config.max_action_bytes == 0
305        {
306            return Err(DecisionLedgerError::InvalidCapacity);
307        }
308        Ok(Self {
309            state_version: DECISION_LEDGER_STATE_VERSION,
310            config,
311            pending: BTreeMap::new(),
312            completed: BTreeMap::new(),
313        })
314    }
315
316    /// Register immutable decision facts, accepting exact idempotent replay.
317    pub fn register(
318        &mut self,
319        decision: PendingDecision<C, A>,
320    ) -> Result<DecisionReceipt, DecisionLedgerError> {
321        if decision.expires_at < decision.created_at {
322            return Err(DecisionLedgerError::InvalidTimeRange);
323        }
324        decision
325            .context
326            .validate_bounded(self.config.max_context_bytes)?;
327        decision
328            .action
329            .validate_bounded(self.config.max_action_bytes)?;
330        if let Some(existing) = self.pending.get(&decision.id) {
331            return if existing == &decision {
332                Ok(DecisionReceipt {
333                    id: decision.id,
334                    status: RegistrationStatus::PendingReplay,
335                })
336            } else {
337                Err(DecisionLedgerError::DecisionConflict)
338            };
339        }
340        if let Some(existing) = self.completed.get(&decision.id) {
341            return if existing.decision == decision {
342                Ok(DecisionReceipt {
343                    id: decision.id,
344                    status: RegistrationStatus::CompletedReplay,
345                })
346            } else {
347                Err(DecisionLedgerError::DecisionConflict)
348            };
349        }
350        if self.pending.len() >= self.config.max_pending {
351            return Err(DecisionLedgerError::PendingCapacityExceeded);
352        }
353        let id = decision.id;
354        self.pending.insert(id, decision);
355        Ok(DecisionReceipt {
356            id,
357            status: RegistrationStatus::Registered,
358        })
359    }
360
361    /// Validate and apply delayed feedback transactionally.
362    pub fn apply_feedback(
363        &mut self,
364        outcome: DecisionOutcome<A>,
365    ) -> Result<FeedbackStatus, DecisionLedgerError> {
366        if !outcome.reward.is_finite() {
367            return Err(DecisionLedgerError::NonFiniteReward);
368        }
369        outcome
370            .action
371            .validate_bounded(self.config.max_action_bytes)?;
372        if self.completed.contains_key(&outcome.decision_id) {
373            return Err(DecisionLedgerError::DuplicateFeedback);
374        }
375        let pending = self
376            .pending
377            .get(&outcome.decision_id)
378            .ok_or(DecisionLedgerError::UnknownDecision)?;
379        if outcome.observed_at < pending.created_at {
380            return Err(DecisionLedgerError::FeedbackBeforeDecision);
381        }
382        if outcome.observed_at > pending.expires_at {
383            return Err(DecisionLedgerError::FeedbackExpired);
384        }
385        if outcome.model_generation != pending.model_generation {
386            return Err(DecisionLedgerError::GenerationMismatch);
387        }
388        if outcome.action != pending.action {
389            return Err(DecisionLedgerError::ActionMismatch);
390        }
391        if self.completed.len() >= self.config.max_completed {
392            return Err(DecisionLedgerError::CompletedCapacityExceeded);
393        }
394
395        // Every fallible check is above this point. Moving the pending entry is
396        // therefore an all-or-nothing state transition.
397        let decision = self
398            .pending
399            .remove(&outcome.decision_id)
400            .ok_or(DecisionLedgerError::UnknownDecision)?;
401        self.completed.insert(
402            outcome.decision_id,
403            CompletedDecision {
404                decision,
405                reward: outcome.reward,
406                observed_at: outcome.observed_at,
407            },
408        );
409        Ok(FeedbackStatus::Applied)
410    }
411
412    /// Return pending decision facts without updating state.
413    pub fn pending(&self, id: DecisionId) -> Option<&PendingDecision<C, A>> {
414        self.pending.get(&id)
415    }
416
417    /// Return a completed tombstone without updating state.
418    pub fn completed(&self, id: DecisionId) -> Option<&CompletedDecision<C, A>> {
419        self.completed.get(&id)
420    }
421
422    /// Number of incomplete decisions.
423    pub fn pending_len(&self) -> usize {
424        self.pending.len()
425    }
426
427    /// Number of retained completed tombstones.
428    pub fn completed_len(&self) -> usize {
429        self.completed.len()
430    }
431
432    /// Validate all feature-independent structural and bounded-value invariants.
433    ///
434    /// This is available even when the optional `serde` feature is disabled;
435    /// [`ValidateState`] delegates to the same checks when serde persistence is
436    /// enabled.
437    pub fn validate(&self) -> Result<(), RillError> {
438        self.validate_structure()
439    }
440
441    /// Explicitly clear decisions whose expiry is strictly before `now`.
442    pub fn clear_expired(&mut self, now: u64) -> Vec<DecisionId> {
443        let ids = self
444            .pending
445            .iter()
446            .filter_map(|(&id, decision)| (decision.expires_at < now).then_some(id))
447            .collect::<Vec<_>>();
448        for id in &ids {
449            self.pending.remove(id);
450        }
451        ids
452    }
453
454    /// Explicitly clear tombstones observed strictly before `before`.
455    pub fn clear_completed_before(&mut self, before: u64) -> Vec<DecisionId> {
456        let ids = self
457            .completed
458            .iter()
459            .filter_map(|(&id, decision)| (decision.observed_at < before).then_some(id))
460            .collect::<Vec<_>>();
461        for id in &ids {
462            self.completed.remove(id);
463        }
464        ids
465    }
466
467    fn validate_structure(&self) -> Result<(), RillError> {
468        if self.state_version != DECISION_LEDGER_STATE_VERSION {
469            return Err(RillError::IncompatibleStateVersion {
470                expected: DECISION_LEDGER_STATE_VERSION,
471                actual: self.state_version,
472            });
473        }
474        if self.config.max_pending == 0
475            || self.config.max_completed == 0
476            || self.config.max_context_bytes == 0
477            || self.config.max_action_bytes == 0
478        {
479            return Err(RillError::InvalidState(
480                "decision ledger capacity must be non-zero".to_owned(),
481            ));
482        }
483        if self.pending.len() > self.config.max_pending
484            || self.completed.len() > self.config.max_completed
485        {
486            return Err(RillError::InvalidState(
487                "decision ledger exceeds its configured capacity".to_owned(),
488            ));
489        }
490        for (id, decision) in &self.pending {
491            if id != &decision.id || decision.expires_at < decision.created_at {
492                return Err(RillError::InvalidState(
493                    "pending decision has inconsistent id or timestamps".to_owned(),
494                ));
495            }
496            if self.completed.contains_key(id) {
497                return Err(RillError::InvalidState(
498                    "decision exists in pending and completed sets".to_owned(),
499                ));
500            }
501            decision
502                .context
503                .validate_bounded(self.config.max_context_bytes)
504                .map_err(|error| RillError::InvalidState(error.to_string()))?;
505            decision
506                .action
507                .validate_bounded(self.config.max_action_bytes)
508                .map_err(|error| RillError::InvalidState(error.to_string()))?;
509        }
510        for (id, completed) in &self.completed {
511            let decision = &completed.decision;
512            if id != &decision.id
513                || decision.expires_at < decision.created_at
514                || completed.observed_at < decision.created_at
515                || completed.observed_at > decision.expires_at
516                || !completed.reward.is_finite()
517            {
518                return Err(RillError::InvalidState(
519                    "completed decision has inconsistent state".to_owned(),
520                ));
521            }
522            decision
523                .context
524                .validate_bounded(self.config.max_context_bytes)
525                .map_err(|error| RillError::InvalidState(error.to_string()))?;
526            decision
527                .action
528                .validate_bounded(self.config.max_action_bytes)
529                .map_err(|error| RillError::InvalidState(error.to_string()))?;
530        }
531        Ok(())
532    }
533
534    /// Validate structural invariants and caller-owned context/action state.
535    pub fn validate_state_with<FC, FA>(
536        &self,
537        mut validate_context: FC,
538        mut validate_action: FA,
539    ) -> Result<(), RillError>
540    where
541        FC: FnMut(&C) -> Result<(), RillError>,
542        FA: FnMut(&A) -> Result<(), RillError>,
543    {
544        self.validate_structure()?;
545        for decision in self.pending.values() {
546            validate_context(&decision.context)?;
547            validate_action(&decision.action)?;
548        }
549        for completed in self.completed.values() {
550            validate_context(&completed.decision.context)?;
551            validate_action(&completed.decision.action)?;
552        }
553        Ok(())
554    }
555}
556
557#[cfg(feature = "serde")]
558impl<C, A> ValidateState for DecisionLedger<C, A>
559where
560    C: PartialEq + BoundedDecisionValue,
561    A: PartialEq + BoundedDecisionValue,
562{
563    fn validate_state(&self) -> Result<(), RillError> {
564        self.validate()
565    }
566}
567
568/// LinUCB-oriented decision facts; the generic ledger remains algorithm-free.
569#[cfg(feature = "bandit")]
570pub type DelayedContextualDecision = PendingDecision<Vec<f64>, usize>;
571
572/// Atomically apply a validated outcome to both a ledger and LinUCB model.
573///
574/// The helper clones both Preview ledger state and the model, so either both
575/// transitions succeed or neither changes. Callers requiring a lower-copy
576/// path can validate with [`DecisionLedger::pending`] and manage their own
577/// transaction boundary.
578#[cfg(feature = "bandit")]
579pub fn apply_contextual_outcome(
580    ledger: &mut DecisionLedger<Vec<f64>, usize>,
581    model: &mut crate::bandit::LinUcb,
582    outcome: DecisionOutcome<usize>,
583) -> Result<FeedbackStatus, RillError> {
584    use crate::bandit::ContextualBandit;
585
586    let pending = ledger
587        .pending(outcome.decision_id)
588        .ok_or_else(|| RillError::InvalidState("unknown delayed decision".to_owned()))?;
589    if pending.context.len() != model.feature_count() {
590        return Err(RillError::DimensionMismatch {
591            expected: model.feature_count(),
592            actual: pending.context.len(),
593        });
594    }
595    for &value in &pending.context {
596        if !value.is_finite() {
597            return Err(RillError::NonFiniteValue {
598                field: "decision context",
599                value,
600            });
601        }
602    }
603    let context = pending.context.clone();
604    let action = outcome.action;
605    let reward = outcome.reward;
606    let mut next_ledger = ledger.clone();
607    let mut next_model = model.clone();
608    let status = next_ledger
609        .apply_feedback(outcome)
610        .map_err(|error| RillError::InvalidState(error.to_string()))?;
611    next_model.update(action, &context, reward)?;
612    *ledger = next_ledger;
613    *model = next_model;
614    Ok(status)
615}
616
617#[cfg(test)]
618mod tests {
619    use super::*;
620    use proptest::prelude::*;
621
622    fn decision(id: u128) -> PendingDecision<Vec<f64>, usize> {
623        PendingDecision::new(DecisionId(id), vec![1.0, 2.0], 1, 10, 20, 7)
624    }
625
626    fn outcome(id: u128) -> DecisionOutcome<usize> {
627        DecisionOutcome {
628            decision_id: DecisionId(id),
629            action: 1,
630            reward: 0.5,
631            observed_at: 15,
632            model_generation: 7,
633        }
634    }
635
636    #[test]
637    fn registration_is_idempotent_but_conflicts_fail() {
638        let mut ledger = DecisionLedger::new(DecisionLedgerConfig::new(2, 2).unwrap()).unwrap();
639        assert_eq!(
640            ledger.register(decision(1)).unwrap().status,
641            RegistrationStatus::Registered
642        );
643        assert_eq!(
644            ledger.register(decision(1)).unwrap().status,
645            RegistrationStatus::PendingReplay
646        );
647        let mut conflict = decision(1);
648        conflict.action = 0;
649        assert_eq!(
650            ledger.register(conflict),
651            Err(DecisionLedgerError::DecisionConflict)
652        );
653        assert_eq!(ledger.pending_len(), 1);
654    }
655
656    #[test]
657    fn feedback_checks_are_atomic_and_tombstoned() {
658        let mut ledger = DecisionLedger::new(DecisionLedgerConfig::new(1, 1).unwrap()).unwrap();
659        ledger.register(decision(1)).unwrap();
660        let before = ledger.clone();
661        let mut bad = outcome(1);
662        bad.action = 0;
663        assert_eq!(
664            ledger.apply_feedback(bad),
665            Err(DecisionLedgerError::ActionMismatch)
666        );
667        assert_eq!(ledger, before);
668        assert_eq!(
669            ledger.apply_feedback(outcome(1)).unwrap(),
670            FeedbackStatus::Applied
671        );
672        assert_eq!(ledger.pending_len(), 0);
673        assert_eq!(ledger.completed_len(), 1);
674        assert_eq!(
675            ledger.apply_feedback(outcome(1)),
676            Err(DecisionLedgerError::DuplicateFeedback)
677        );
678    }
679
680    #[test]
681    fn expiry_boundary_is_inclusive_and_cleanup_is_explicit() {
682        let mut ledger = DecisionLedger::new(DecisionLedgerConfig::new(2, 2).unwrap()).unwrap();
683        ledger.register(decision(1)).unwrap();
684        let mut at_boundary = outcome(1);
685        at_boundary.observed_at = 20;
686        assert!(ledger.apply_feedback(at_boundary).is_ok());
687
688        ledger.register(decision(2)).unwrap();
689        assert!(ledger.clear_expired(20).is_empty());
690        assert_eq!(ledger.clear_expired(21), vec![DecisionId(2)]);
691    }
692
693    #[test]
694    fn capacity_never_evicts_live_or_completed_state() {
695        let mut ledger = DecisionLedger::new(DecisionLedgerConfig::new(1, 1).unwrap()).unwrap();
696        ledger.register(decision(1)).unwrap();
697        assert_eq!(
698            ledger.register(decision(2)),
699            Err(DecisionLedgerError::PendingCapacityExceeded)
700        );
701        ledger.apply_feedback(outcome(1)).unwrap();
702        ledger.register(decision(2)).unwrap();
703        assert_eq!(
704            ledger.apply_feedback(outcome(2)),
705            Err(DecisionLedgerError::CompletedCapacityExceeded)
706        );
707        assert!(ledger.pending(DecisionId(2)).is_some());
708    }
709
710    #[test]
711    fn per_decision_context_and_action_memory_is_bounded() {
712        let config = DecisionLedgerConfig::new(2, 2)
713            .unwrap()
714            .with_value_limits(8, std::mem::size_of::<usize>())
715            .unwrap();
716        let mut ledger = DecisionLedger::new(config).unwrap();
717        assert_eq!(
718            ledger.register(decision(1)),
719            Err(DecisionLedgerError::ValueTooLarge)
720        );
721        assert_eq!(ledger.pending_len(), 0);
722    }
723
724    #[test]
725    fn every_feedback_fact_is_checked_without_mutation() {
726        let mut ledger = DecisionLedger::new(DecisionLedgerConfig::new(2, 2).unwrap()).unwrap();
727        ledger.register(decision(1)).unwrap();
728        let cases = [
729            (
730                DecisionOutcome {
731                    decision_id: DecisionId(9),
732                    ..outcome(1)
733                },
734                DecisionLedgerError::UnknownDecision,
735            ),
736            (
737                DecisionOutcome {
738                    observed_at: 9,
739                    ..outcome(1)
740                },
741                DecisionLedgerError::FeedbackBeforeDecision,
742            ),
743            (
744                DecisionOutcome {
745                    observed_at: 21,
746                    ..outcome(1)
747                },
748                DecisionLedgerError::FeedbackExpired,
749            ),
750            (
751                DecisionOutcome {
752                    model_generation: 8,
753                    ..outcome(1)
754                },
755                DecisionLedgerError::GenerationMismatch,
756            ),
757            (
758                DecisionOutcome {
759                    action: 0,
760                    ..outcome(1)
761                },
762                DecisionLedgerError::ActionMismatch,
763            ),
764        ];
765        for (bad, expected) in cases {
766            let before = ledger.clone();
767            assert_eq!(ledger.apply_feedback(bad), Err(expected));
768            assert_eq!(ledger, before);
769        }
770    }
771
772    #[test]
773    fn completed_replay_and_explicit_tombstone_cleanup() {
774        let mut ledger = DecisionLedger::new(DecisionLedgerConfig::new(2, 2).unwrap()).unwrap();
775        ledger.register(decision(1)).unwrap();
776        ledger.apply_feedback(outcome(1)).unwrap();
777        assert_eq!(
778            ledger.register(decision(1)).unwrap().status,
779            RegistrationStatus::CompletedReplay
780        );
781        assert!(ledger.clear_completed_before(15).is_empty());
782        assert_eq!(ledger.clear_completed_before(16), vec![DecisionId(1)]);
783        assert_eq!(ledger.completed_len(), 0);
784    }
785
786    #[test]
787    fn caller_validator_can_enforce_domain_specific_context_rules() {
788        let mut ledger = DecisionLedger::new(DecisionLedgerConfig::new(2, 2).unwrap()).unwrap();
789        let mut invalid = decision(1);
790        invalid.context[0] = -1.0;
791        ledger.register(invalid).unwrap();
792        assert!(
793            ledger
794                .validate_state_with(
795                    |context| {
796                        for &value in context {
797                            if value < 0.0 {
798                                return Err(RillError::InvalidState(
799                                    "context values must be non-negative".to_owned(),
800                                ));
801                            }
802                        }
803                        Ok(())
804                    },
805                    |_| Ok(()),
806                )
807                .is_err()
808        );
809    }
810
811    #[cfg(feature = "bandit")]
812    #[test]
813    fn contextual_helper_updates_model_and_ledger_together() {
814        let mut ledger = DecisionLedger::new(DecisionLedgerConfig::new(2, 2).unwrap()).unwrap();
815        ledger
816            .register(PendingDecision::new(DecisionId(1), vec![1.0], 0, 10, 20, 7))
817            .unwrap();
818        let config = crate::bandit::LinUcbConfig {
819            arm_count: 1,
820            feature_count: 1,
821            alpha: 1.0,
822        };
823        let mut model = crate::bandit::LinUcb::new(config).unwrap();
824        apply_contextual_outcome(
825            &mut ledger,
826            &mut model,
827            DecisionOutcome {
828                decision_id: DecisionId(1),
829                action: 0,
830                reward: 1.0,
831                observed_at: 12,
832                model_generation: 7,
833            },
834        )
835        .unwrap();
836        assert_eq!(ledger.pending_len(), 0);
837        assert_eq!(ledger.completed_len(), 1);
838        assert_eq!(crate::bandit::ContextualBandit::samples_seen(&model), 1);
839    }
840
841    #[cfg(feature = "serde")]
842    #[test]
843    fn serde_roundtrip_and_corruption_validation() {
844        let mut ledger = DecisionLedger::new(DecisionLedgerConfig::new(2, 2).unwrap()).unwrap();
845        ledger.register(decision(1)).unwrap();
846        let json = serde_json::to_string(&ledger).unwrap();
847        let restored: DecisionLedger<Vec<f64>, usize> = serde_json::from_str(&json).unwrap();
848        assert_eq!(restored, ledger);
849        restored.validate_state().unwrap();
850
851        let corrupt = json.replace("\"state_version\":1", "\"state_version\":9");
852        let restored: DecisionLedger<Vec<f64>, usize> = serde_json::from_str(&corrupt).unwrap();
853        assert!(restored.validate_state().is_err());
854
855        let corrupt = json.replace("\"max_pending\":2", "\"max_pending\":0");
856        let restored: DecisionLedger<Vec<f64>, usize> = serde_json::from_str(&corrupt).unwrap();
857        assert!(restored.validate_state().is_err());
858    }
859
860    proptest! {
861        #[test]
862        fn arbitrary_invalid_feedback_never_removes_pending(
863            reward in prop_oneof![Just(f64::NAN), Just(f64::INFINITY)],
864            observed_at in any::<u64>(),
865        ) {
866            let mut ledger = DecisionLedger::new(DecisionLedgerConfig::new(1, 1).unwrap()).unwrap();
867            ledger.register(decision(1)).unwrap();
868            let before = ledger.clone();
869            let result = ledger.apply_feedback(DecisionOutcome {
870                decision_id: DecisionId(1), action: 1, reward, observed_at, model_generation: 7,
871            });
872            prop_assert!(result.is_err());
873            prop_assert_eq!(ledger, before);
874        }
875    }
876}
rill_ml/decision/mod.rs

rill_ml/decision/
mod.rs
