Skip to main content

rustev_runtime/
optimization.rs

1//! Bounded reusable-work state for spec 017.
2//!
3//! State is kept per cache namespace, which is per plan: one plan's bounds
4//! never evict another plan's entries, and a per-scope bound evicts only the
5//! inserting scope's own oldest entries (I-2).
6
7use std::collections::{BTreeMap, VecDeque};
8use std::sync::{Arc, Mutex};
9
10use rustev_contract::canonical::{record_canonical_bytes, tagged_digest};
11use rustev_contract::ids::{ArtifactId, DescriptorId, PlanId};
12use rustev_contract::optimization::{MemoryCachePolicy, OptimizationPolicy};
13use rustev_contract::output::RawOutput;
14
15use crate::clock::Instant;
16
17const KEY_TAG: &str = "rustev.optimization-key/1";
18const ENTRY_TAG: &str = "rustev.optimization-entry/1";
19
20#[derive(Debug, Clone, PartialEq, Eq)]
21pub(crate) struct RequestKey {
22    pub id: String,
23    pub material: Vec<u8>,
24    pub scope_id: String,
25    pub namespace: String,
26}
27
28impl RequestKey {
29    /// The request key over `material`, as a decision entry of the
30    /// persistent tier retains it (spec 044, 3.6.4).
31    pub(crate) fn from_material(material: Vec<u8>, scope_id: String, namespace: String) -> Self {
32        RequestKey {
33            id: tagged_digest(KEY_TAG, &material),
34            material,
35            scope_id,
36            namespace,
37        }
38    }
39}
40
41/// Every output-affecting member available at the runtime seam.
42///
43/// The plan id transitively identifies operation, task, question, ordered
44/// options or levels, ordered candidates, preprocessing, tokenization,
45/// truncation, precision, backend configuration, requested privacy,
46/// served-identity requirements, step semantics, calibration, fallback,
47/// protocol, adapter, artifact binding and the optimization policy. The
48/// canonical projection identifies the concrete projected values, instance,
49/// canonical inputs and provenance-relevant revisions. Descriptor and
50/// artifact ids bind the selected installed target. `principal` is the
51/// host-supplied complete authorized-scope handle. `profile` is the plan's
52/// profile identity or `none` (spec 044, 3.5.2), held apart from the plan
53/// id so no profile can collide with none. Runtime and key-schema versions
54/// prevent reuse across implementation or key interpretation.
55pub(crate) struct KeyParts<'a> {
56    pub plan_id: &'a PlanId,
57    pub profile: &'a str,
58    pub policy: &'a OptimizationPolicy,
59    pub principal: &'a [u8],
60    pub backend_artifact: &'a ArtifactId,
61    pub backend_descriptor: &'a DescriptorId,
62    pub step: &'a str,
63    pub instance: &'a [String],
64    pub projection: &'a [u8],
65}
66
67fn part(out: &mut Vec<u8>, bytes: &[u8]) {
68    out.extend_from_slice(&(bytes.len() as u64).to_be_bytes());
69    out.extend_from_slice(bytes);
70}
71
72pub(crate) fn request_key(p: KeyParts<'_>) -> RequestKey {
73    let mut material = Vec::new();
74    part(&mut material, p.plan_id.as_str().as_bytes());
75    part(&mut material, p.profile.as_bytes());
76    part(&mut material, &p.policy.key_schema_version.to_be_bytes());
77    part(
78        &mut material,
79        &p.policy.cache_namespace_version.to_be_bytes(),
80    );
81    part(&mut material, env!("CARGO_PKG_VERSION").as_bytes());
82    part(&mut material, p.principal);
83    part(&mut material, p.backend_artifact.as_str().as_bytes());
84    part(&mut material, p.backend_descriptor.as_str().as_bytes());
85    part(&mut material, p.step.as_bytes());
86    part(&mut material, &(p.instance.len() as u64).to_be_bytes());
87    for member in p.instance {
88        part(&mut material, member.as_bytes());
89    }
90    part(&mut material, p.projection);
91    let scope_id = scope_id(p.principal);
92    let namespace = namespace(p.plan_id, p.policy);
93    RequestKey {
94        id: tagged_digest(KEY_TAG, &material),
95        material,
96        scope_id,
97        namespace,
98    }
99}
100
101/// The plan's profile identity for request keys (spec 044, 3.5.2): the
102/// JSON array of the profile ids its instantiation applied, in chain
103/// order, or `none`. An array delimits every id, and never reads `none`.
104/// The match is exhaustive, so a new chain variant must be given its own
105/// encoding here and can never share another's key.
106pub(crate) fn profile_of(plan: &rustev_contract::plan::Plan) -> String {
107    use rustev_contract::plan::ProfileChain;
108    match plan.instantiation.as_ref().map(|i| &i.profile) {
109        None | Some(ProfileChain::None) => "none".into(),
110        // An empty chain is `[]`: still an array, never `none`.
111        Some(ProfileChain::Chain(entries)) => {
112            profile_ids(&entries.iter().map(|e| e.id.as_str()).collect::<Vec<_>>())
113        }
114    }
115}
116
117fn profile_ids(ids: &[&str]) -> String {
118    serde_json::to_string(ids).expect("a list of strings serializes")
119}
120
121pub(crate) fn scope_id(principal: &[u8]) -> String {
122    tagged_digest("rustev.optimization-scope/1", principal)
123}
124
125pub(crate) fn namespace(plan_id: &PlanId, policy: &OptimizationPolicy) -> String {
126    format!(
127        "{}/{}/{}",
128        plan_id, policy.key_schema_version, policy.cache_namespace_version
129    )
130}
131
132#[derive(Debug, Clone, PartialEq)]
133pub(crate) struct CachedOutput {
134    pub output: RawOutput,
135    pub source_attempt_id: String,
136    pub target: u32,
137    pub entry_id: String,
138    pub age_ms: u64,
139    pub generation: u64,
140}
141
142pub(crate) struct CacheWrite {
143    pub output: RawOutput,
144    pub source_attempt_id: String,
145    pub target: u32,
146    pub captured_generation: u64,
147    pub now: Instant,
148}
149
150#[derive(Debug, Clone, PartialEq, Eq)]
151pub(crate) struct CacheAdmission {
152    pub entry_id: String,
153    pub evicted_entries: u64,
154    pub evicted_bytes: u64,
155}
156
157/// Why a lookup did not hit.
158#[derive(Debug, Clone, Copy, PartialEq, Eq)]
159pub(crate) enum CacheMiss {
160    Absent,
161    Expired,
162    /// Equal digests over different key material: an integrity event.
163    Collision,
164}
165
166/// Why a validated output was not admitted. Nothing is touched before the
167/// admissibility preconditions hold.
168#[derive(Debug, Clone, Copy, PartialEq, Eq)]
169pub(crate) enum CacheRefusal {
170    /// The output has no canonical form.
171    Uncanonical,
172    /// The entry exceeds the per-entry bound.
173    EntryTooLarge,
174    /// The namespace generation advanced after the request captured it.
175    StaleGeneration,
176    /// A retained entry has the same digest and different key material.
177    Collision,
178    /// The policy's memory tier holds no entry: it has no `max_entries`
179    /// (no memory tier, spec 044 3.1) or a bound of 0, which the compiler
180    /// refuses. Nothing is admitted, never an unbounded or default tier.
181    NoCapacity,
182}
183
184#[derive(Debug, Clone)]
185enum CachedValue {
186    Semantic(RawOutput),
187    Page(Vec<u8>),
188}
189
190struct StoredWrite {
191    output: CachedValue,
192    source_attempt_id: String,
193    target: u32,
194    captured_generation: u64,
195    now: Instant,
196}
197
198pub(crate) struct PageWrite {
199    pub bytes: Vec<u8>,
200    pub attempt: String,
201    pub generation: u64,
202    pub now: Instant,
203}
204
205#[derive(Debug, Clone)]
206struct Entry {
207    key: RequestKey,
208    output: CachedValue,
209    source_attempt_id: String,
210    target: u32,
211    entry_id: String,
212    created: Instant,
213    expires: Instant,
214    generation: u64,
215    bytes: u64,
216    /// Admission sequence, which makes lazily deleted queue slots
217    /// recognizable.
218    seq: u64,
219}
220
221type Slot = (String, u64);
222
223#[derive(Debug, Default)]
224struct NamespaceState {
225    generation: u64,
226    entries: BTreeMap<String, Entry>,
227    fifo: VecDeque<Slot>,
228    scope_fifo: BTreeMap<String, VecDeque<Slot>>,
229    scope_bytes: BTreeMap<String, u64>,
230    total_bytes: u64,
231    next_seq: u64,
232    /// The decision-result cache's memory tier (spec 044, 3.6), under the
233    /// same generation and with its own entry bound.
234    decisions: BTreeMap<String, decision::DecisionSlot>,
235    decision_fifo: VecDeque<Slot>,
236}
237
238impl NamespaceState {
239    fn live(&self, slot: &Slot) -> bool {
240        self.entries.get(&slot.0).is_some_and(|e| e.seq == slot.1)
241    }
242
243    fn remove(&mut self, id: &str) -> Option<Entry> {
244        let entry = self.entries.remove(id)?;
245        self.total_bytes = self.total_bytes.saturating_sub(entry.bytes);
246        if let Some(bytes) = self.scope_bytes.get_mut(&entry.key.scope_id) {
247            *bytes = bytes.saturating_sub(entry.bytes);
248            if *bytes == 0 {
249                self.scope_bytes.remove(&entry.key.scope_id);
250            }
251        }
252        // Queue slots are deleted lazily; compact when stale slots dominate
253        // so the queues stay proportional to the live entries.
254        if self.fifo.len() > 2 * self.entries.len() + 16 {
255            let entries = &self.entries;
256            self.fifo
257                .retain(|s| entries.get(&s.0).is_some_and(|e| e.seq == s.1));
258        }
259        if let Some(queue) = self.scope_fifo.get_mut(&entry.key.scope_id) {
260            let live = self.scope_bytes.contains_key(&entry.key.scope_id);
261            if !live {
262                self.scope_fifo.remove(&entry.key.scope_id);
263            } else if queue.len() > 16 {
264                let entries = &self.entries;
265                queue.retain(|s| entries.get(&s.0).is_some_and(|e| e.seq == s.1));
266            }
267        }
268        Some(entry)
269    }
270
271    /// Evict the oldest live entry of `scope`, or of the namespace.
272    fn evict_oldest(&mut self, scope: Option<&str>) -> Option<Entry> {
273        loop {
274            let slot = match scope {
275                Some(scope) => self.scope_fifo.get_mut(scope)?.pop_front()?,
276                None => self.fifo.pop_front()?,
277            };
278            if self.live(&slot) {
279                return self.remove(&slot.0);
280            }
281        }
282    }
283}
284
285/// Which retained entries an invalidation selects.
286#[derive(Debug, Clone, PartialEq, Eq)]
287pub(crate) enum InvalidationSelector {
288    Namespace { namespace: String },
289    Scope { namespace: String, scope_id: String },
290}
291
292#[derive(Debug, Clone, PartialEq, Eq)]
293pub struct InvalidationResult {
294    /// The memory tier's namespace generation after the invalidation.
295    /// Results captured under an earlier generation are never stored.
296    pub generation: u64,
297    pub removed_entries: u64,
298    pub removed_bytes: u64,
299    /// The persistent tier's result, when the plan has one (spec 044,
300    /// 3.10.3).
301    pub persistent: Option<PersistentInvalidation>,
302    /// Whether every tier of the selection is invalidated. Until it is, the
303    /// selection's persistent entries are neither read nor written.
304    pub complete: bool,
305}
306
307/// The persistent tier's part of an invalidation (spec 044, 3.10.3).
308#[derive(Debug, Clone, PartialEq, Eq)]
309pub enum PersistentInvalidation {
310    /// Marked unavailable; `Runtime::invalidate` has not completed it.
311    Pending,
312    /// The store advanced the selected generation.
313    Done { generation: u64, removed: u64 },
314    /// The store failed; the selection stays unavailable until a later
315    /// `Runtime::invalidate` reconciles it (spec 017, 3.7).
316    Failed { detail: String },
317}
318
319#[derive(Debug, Default)]
320pub(crate) struct Optimizer {
321    namespaces: Mutex<BTreeMap<String, NamespaceState>>,
322    /// Shared calls in flight, per plan namespace (spec 044, 3.2): a key
323    /// holds the plan, so no attempt is shared across plans, and each plan's
324    /// `max_shared_calls` bounds its own slots.
325    shared: Mutex<BTreeMap<String, crate::driver::SharedCalls>>,
326}
327
328impl Optimizer {
329    pub(crate) fn shared_calls(
330        &self,
331        namespace: &str,
332        max_slots: usize,
333    ) -> crate::driver::SharedCalls {
334        self.shared
335            .lock()
336            .unwrap_or_else(|e| e.into_inner())
337            .entry(namespace.to_string())
338            .or_insert_with(|| Arc::new(crate::shared::Registry::new(max_slots)))
339            .clone()
340    }
341
342    fn lock(&self) -> std::sync::MutexGuard<'_, BTreeMap<String, NamespaceState>> {
343        self.namespaces.lock().unwrap_or_else(|e| e.into_inner())
344    }
345
346    pub(crate) fn generation(&self, namespace: &str) -> u64 {
347        self.lock().get(namespace).map_or(0, |ns| ns.generation)
348    }
349
350    fn lookup_entry(&self, key: &RequestKey, now: Instant) -> Result<Entry, CacheMiss> {
351        let mut namespaces = self.lock();
352        let Some(ns) = namespaces.get_mut(&key.namespace) else {
353            return Err(CacheMiss::Absent);
354        };
355        let Some(entry) = ns.entries.get(&key.id) else {
356            return Err(CacheMiss::Absent);
357        };
358        if entry.key.material != key.material {
359            return Err(CacheMiss::Collision);
360        }
361        if entry.expires <= now || entry.generation != ns.generation {
362            ns.remove(&key.id);
363            return Err(CacheMiss::Expired);
364        }
365        Ok(entry.clone())
366    }
367
368    pub(crate) fn lookup(&self, key: &RequestKey, now: Instant) -> Result<CachedOutput, CacheMiss> {
369        let e = self.lookup_entry(key, now)?;
370        let CachedValue::Semantic(output) = e.output else {
371            return Err(CacheMiss::Collision);
372        };
373        Ok(CachedOutput {
374            output,
375            source_attempt_id: e.source_attempt_id,
376            target: e.target,
377            entry_id: e.entry_id,
378            age_ms: now.saturating_sub(e.created),
379            generation: e.generation,
380        })
381    }
382
383    pub(crate) fn lookup_page(
384        &self,
385        key: &RequestKey,
386        now: Instant,
387    ) -> Result<(Vec<u8>, u64), CacheMiss> {
388        let e = self.lookup_entry(key, now)?;
389        let CachedValue::Page(bytes) = e.output else {
390            return Err(CacheMiss::Collision);
391        };
392        Ok((bytes, now.saturating_sub(e.created)))
393    }
394
395    pub(crate) fn insert_page(
396        &self,
397        policy: &MemoryCachePolicy,
398        key: RequestKey,
399        page: PageWrite,
400    ) -> Result<CacheAdmission, CacheRefusal> {
401        self.insert_value(
402            policy,
403            key,
404            StoredWrite {
405                source_attempt_id: page.attempt,
406                target: 0,
407                captured_generation: page.generation,
408                now: page.now,
409                output: CachedValue::Page(page.bytes),
410            },
411        )
412    }
413
414    pub(crate) fn insert(
415        &self,
416        policy: &MemoryCachePolicy,
417        key: RequestKey,
418        write: CacheWrite,
419    ) -> Result<CacheAdmission, CacheRefusal> {
420        self.insert_value(
421            policy,
422            key,
423            StoredWrite {
424                output: CachedValue::Semantic(write.output),
425                source_attempt_id: write.source_attempt_id,
426                target: write.target,
427                captured_generation: write.captured_generation,
428                now: write.now,
429            },
430        )
431    }
432
433    fn insert_value(
434        &self,
435        policy: &MemoryCachePolicy,
436        key: RequestKey,
437        write: StoredWrite,
438    ) -> Result<CacheAdmission, CacheRefusal> {
439        let output_bytes = match &write.output {
440            CachedValue::Semantic(output) => {
441                record_canonical_bytes(output).map_err(|_| CacheRefusal::Uncanonical)?
442            }
443            CachedValue::Page(bytes) => bytes.clone(),
444        };
445        let bytes = key
446            .material
447            .len()
448            .saturating_add(output_bytes.len())
449            .saturating_add(write.source_attempt_id.len()) as u64;
450        // Core validation guarantees max_entry_bytes <= max_scope_bytes <=
451        // max_bytes and max_entries >= 1, so an entry within its own bound
452        // is always admissible after eviction.
453        if bytes > policy.max_entry_bytes {
454            return Err(CacheRefusal::EntryTooLarge);
455        }
456        let mut namespaces = self.lock();
457        let ns = namespaces.entry(key.namespace.clone()).or_default();
458        if ns.generation != write.captured_generation {
459            return Err(CacheRefusal::StaleGeneration);
460        }
461        if ns
462            .entries
463            .get(&key.id)
464            .is_some_and(|entry| entry.key.material != key.material)
465        {
466            return Err(CacheRefusal::Collision);
467        }
468        // Replacing an equivalent entry is not an eviction.
469        ns.remove(&key.id);
470        let mut evicted_entries = 0_u64;
471        let mut evicted_bytes = 0_u64;
472        let mut evicted = |entry: Entry| {
473            evicted_entries += 1;
474            evicted_bytes = evicted_bytes.saturating_add(entry.bytes);
475        };
476        while ns
477            .scope_bytes
478            .get(&key.scope_id)
479            .copied()
480            .unwrap_or(0)
481            .saturating_add(bytes)
482            > policy.max_scope_bytes
483        {
484            let Some(entry) = ns.evict_oldest(Some(&key.scope_id)) else {
485                break;
486            };
487            evicted(entry);
488        }
489        while ns.entries.len() as u64 >= policy.max_entries
490            || ns.total_bytes.saturating_add(bytes) > policy.max_bytes
491        {
492            let Some(entry) = ns.evict_oldest(None) else {
493                break;
494            };
495            evicted(entry);
496        }
497        let mut entry_material = key.material.clone();
498        part(&mut entry_material, &output_bytes);
499        part(&mut entry_material, write.source_attempt_id.as_bytes());
500        let entry_id = tagged_digest(ENTRY_TAG, &entry_material);
501        let seq = ns.next_seq;
502        ns.next_seq += 1;
503        *ns.scope_bytes.entry(key.scope_id.clone()).or_default() += bytes;
504        ns.total_bytes += bytes;
505        ns.fifo.push_back((key.id.clone(), seq));
506        ns.scope_fifo
507            .entry(key.scope_id.clone())
508            .or_default()
509            .push_back((key.id.clone(), seq));
510        let generation = ns.generation;
511        ns.entries.insert(
512            key.id.clone(),
513            Entry {
514                key,
515                output: write.output,
516                source_attempt_id: write.source_attempt_id,
517                target: write.target,
518                entry_id: entry_id.clone(),
519                created: write.now,
520                expires: write.now.saturating_add(policy.ttl_ms),
521                generation,
522                bytes,
523                seq,
524            },
525        );
526        Ok(CacheAdmission {
527            entry_id,
528            evicted_entries,
529            evicted_bytes,
530        })
531    }
532
533    /// Advance the namespace generation, then remove the selected entries.
534    /// Any result captured under the old generation is refused at insert.
535    pub(crate) fn invalidate(&self, selector: &InvalidationSelector) -> InvalidationResult {
536        let namespace = match selector {
537            InvalidationSelector::Namespace { namespace }
538            | InvalidationSelector::Scope { namespace, .. } => namespace,
539        };
540        let mut namespaces = self.lock();
541        let ns = namespaces.entry(namespace.clone()).or_default();
542        ns.generation = ns.generation.saturating_add(1);
543        let ids: Vec<String> = ns
544            .entries
545            .iter()
546            .filter(|(_, entry)| match selector {
547                InvalidationSelector::Namespace { .. } => true,
548                InvalidationSelector::Scope { scope_id, .. } => &entry.key.scope_id == scope_id,
549            })
550            .map(|(id, _)| id.clone())
551            .collect();
552        let mut removed_bytes = 0;
553        for id in &ids {
554            if let Some(entry) = ns.remove(id) {
555                removed_bytes += entry.bytes;
556            }
557        }
558        let decisions: Vec<String> = ns
559            .decisions
560            .iter()
561            .filter(|(_, slot)| match selector {
562                InvalidationSelector::Namespace { .. } => true,
563                InvalidationSelector::Scope { scope_id, .. } => &slot.key.scope_id == scope_id,
564            })
565            .map(|(id, _)| id.clone())
566            .collect();
567        for id in &decisions {
568            if let Some(slot) = ns.decisions.remove(id) {
569                removed_bytes += slot.bytes;
570            }
571        }
572        InvalidationResult {
573            generation: ns.generation,
574            removed_entries: (ids.len() + decisions.len()) as u64,
575            removed_bytes,
576            persistent: None,
577            complete: true,
578        }
579    }
580
581    #[cfg(test)]
582    fn force_collision(&self, key: &RequestKey, other_material: Vec<u8>) {
583        let mut namespaces = self.lock();
584        if let Some(entry) = namespaces
585            .get_mut(&key.namespace)
586            .and_then(|ns| ns.entries.get_mut(&key.id))
587        {
588            entry.key.material = other_material;
589        }
590    }
591
592    #[cfg(test)]
593    fn total_bytes(&self, namespace: &str) -> u64 {
594        self.lock().get(namespace).map_or(0, |ns| ns.total_bytes)
595    }
596
597    #[cfg(test)]
598    fn queue_lengths(&self, namespace: &str) -> (usize, usize) {
599        let namespaces = self.lock();
600        let ns = &namespaces[namespace];
601        (
602            ns.fifo.len(),
603            ns.scope_fifo.values().map(VecDeque::len).sum(),
604        )
605    }
606}
607
608/// Acquisition identity and complete authorized scope, in the versioned cache namespace.
609pub(crate) fn acquisition_key(
610    plan: &PlanId,
611    policy: &OptimizationPolicy,
612    principal: &[u8],
613    request: &rustev_core::evaluate::PendingAcquisition,
614) -> RequestKey {
615    let mut material = Vec::new();
616    part(&mut material, b"acquisition-page");
617    part(&mut material, &policy.key_schema_version.to_be_bytes());
618    // Like `request_key`, bind the runtime version to prevent reuse across
619    // implementations (spec 017, Implementation record: complete memory-cache
620    // key). Spec 036 3.13.7 uses that versioned key scheme.
621    part(&mut material, env!("CARGO_PKG_VERSION").as_bytes());
622    // `AcquisitionRequestId` is the identity of the whole
623    // `rustev.acquisition-request/1` document, which names its `operation`
624    // (spec 036 3.3.1), so two operations never share a key.
625    part(&mut material, request.request_id.as_str().as_bytes());
626    part(&mut material, request.request.source.as_str().as_bytes());
627    part(&mut material, principal);
628    RequestKey {
629        id: tagged_digest(KEY_TAG, &material),
630        material,
631        scope_id: scope_id(principal),
632        namespace: namespace(plan, policy),
633    }
634}
635
636// ---------------------------------------------------------------------------
637// The decision-result cache (spec 044, 3.5 and 3.6)
638// ---------------------------------------------------------------------------
639
640pub(crate) mod decision {
641    //! The decision-result cache's key, memory tier and per-decision run
642    //! (spec 044, 3.5 and 3.6).
643
644    use std::collections::{BTreeMap, BTreeSet};
645    use std::sync::{Arc, Mutex, MutexGuard};
646
647    use rustev_contract::acquisition_record::AcquisitionRecord;
648    use rustev_contract::cache_entry::EntryGenerations;
649    use rustev_contract::canonical::{record_canonical_bytes, tagged_digest};
650    use rustev_contract::ids::{ContentDigest, PlanId};
651    use rustev_contract::optimization::{
652        DecisionCachePolicy, DecisionCacheTier, OptimizationPolicy, RefreshFailurePolicy,
653        SubstitutedInput,
654    };
655    use rustev_contract::output::RawOutput;
656    use rustev_contract::scope::Handle;
657    use rustev_contract::snapshot::{Entry, Snapshot};
658    use rustev_contract::time::Timestamp;
659
660    use super::{CacheRefusal, Optimizer, RequestKey, namespace, part, profile_of, scope_id};
661
662    const DECISION_KEY_TAG: &str = "rustev.decision-key/1";
663    const DECISION_ENTRY_TAG: &str = "rustev.decision-entry/1";
664    const DECISION_INPUT_TAG: &str = "rustev.decision-input/1";
665
666    /// What a decision key is derived from: the plan's compiled
667    /// `decision_key_schema`, and the decision's scope, dimensions and
668    /// snapshot (spec 044, 3.5 and 3.6.1).
669    pub(crate) struct DecisionKeyParts<'a> {
670        pub schema: &'a [String],
671        pub plan_id: &'a PlanId,
672        pub profile: &'a str,
673        pub policy: &'a OptimizationPolicy,
674        pub principal: &'a [u8],
675        pub dimensions: &'a BTreeMap<String, Handle>,
676        pub snapshot: &'a Snapshot,
677    }
678
679    /// Why a decision has no key, so is neither served from nor admitted to
680    /// the decision cache (spec 044, 3.5 items 3 and 4).
681    #[derive(Debug, Clone, PartialEq, Eq)]
682    pub(crate) enum Unkeyed {
683        /// A declared dimension the host did not supply.
684        DimensionMissing(String),
685        /// A dimension the plan does not declare, a member this runtime cannot
686        /// derive, or a snapshot entry without canonical form.
687        Ineligible(String),
688    }
689
690    impl Unkeyed {
691        pub(crate) fn diagnostic(&self) -> String {
692            match self {
693                Unkeyed::DimensionMissing(d) => format!("key_dimension_missing: {d:?}"),
694                Unkeyed::Ineligible(why) => format!("key_ineligible: {why}"),
695            }
696        }
697    }
698
699    /// The decision-level key (spec 044, 3.6.1): every member of the compiled
700    /// schema in order, each named in the material beside its value. An input
701    /// member holds the digest of each of the input's snapshot entries as
702    /// supplied, in snapshot order; a volatile input is not a member.
703    pub(crate) fn decision_key(p: DecisionKeyParts<'_>) -> Result<RequestKey, Unkeyed> {
704        if let Some(extra) = p
705            .dimensions
706            .keys()
707            .find(|d| !p.schema.contains(&format!("dimension:{d}")))
708        {
709            return Err(Unkeyed::Ineligible(format!(
710                "dimension {extra:?} is not declared by the plan"
711            )));
712        }
713        let mut material = Vec::new();
714        for member in p.schema {
715            part(&mut material, member.as_bytes());
716            match member.as_str() {
717                "plan_id" => part(&mut material, p.plan_id.as_str().as_bytes()),
718                "key_schema_version" => {
719                    part(&mut material, &p.policy.key_schema_version.to_be_bytes())
720                }
721                "cache_namespace_version" => part(
722                    &mut material,
723                    &p.policy.cache_namespace_version.to_be_bytes(),
724                ),
725                "runtime_version" => part(&mut material, env!("CARGO_PKG_VERSION").as_bytes()),
726                "scope" => part(&mut material, p.principal),
727                "profile" => part(&mut material, p.profile.as_bytes()),
728                m => {
729                    if let Some(name) = m.strip_prefix("dimension:") {
730                        let handle = p
731                            .dimensions
732                            .get(name)
733                            .ok_or_else(|| Unkeyed::DimensionMissing(name.into()))?;
734                        part(&mut material, handle.as_str().as_bytes());
735                    } else if let Some(name) = m.strip_prefix("input:") {
736                        let entries: Vec<_> = p
737                            .snapshot
738                            .entries
739                            .iter()
740                            .filter(|e| e.field == name)
741                            .collect();
742                        part(&mut material, &(entries.len() as u64).to_be_bytes());
743                        for e in entries {
744                            let bytes = record_canonical_bytes(e).map_err(|_| {
745                                Unkeyed::Ineligible(format!("input {name:?} has no canonical form"))
746                            })?;
747                            part(
748                                &mut material,
749                                tagged_digest(DECISION_INPUT_TAG, &bytes).as_bytes(),
750                            );
751                        }
752                    } else {
753                        return Err(Unkeyed::Ineligible(format!("unknown key member {m:?}")));
754                    }
755                }
756            }
757        }
758        Ok(RequestKey {
759            id: tagged_digest(DECISION_KEY_TAG, &material),
760            material,
761            scope_id: scope_id(p.principal),
762            namespace: namespace(p.plan_id, p.policy),
763        })
764    }
765
766    /// A request outside the refresh closure as a decision entry holds it
767    /// (spec 044, 3.6.4): its 017 key, the validated output of target 0 and the
768    /// attempt that produced it. Its derivation follows from the plan the key
769    /// names, and its lineage is that attempt and the entry's source record.
770    #[derive(Debug, Clone)]
771    pub(crate) struct ReusedRequest {
772        pub key: RequestKey,
773        pub output: RawOutput,
774        pub source_attempt_id: String,
775    }
776
777    /// By step and instance.
778    pub(crate) type ReusedRequests = BTreeMap<(String, Vec<String>), ReusedRequest>;
779
780    /// A decision entry of the memory tier (spec 044, 3.6.4). It holds the
781    /// reusable work and its lineage, never a judgment to serve.
782    #[derive(Debug, Clone)]
783    pub(crate) struct DecisionEntry {
784        pub requests: ReusedRequests,
785        /// The source decision's snapshot entries of its volatile inputs, as
786        /// supplied, for `serve_cached_within_freshness` (3.6.5).
787        pub volatile: Vec<Entry>,
788        pub source_run_record: ContentDigest,
789        /// The source decision's evaluation time: decision entries expire by
790        /// evaluation time (spec 044, decision 6).
791        pub created: Timestamp,
792    }
793
794    /// A retained decision entry, in its namespace's state.
795    #[derive(Debug, Clone)]
796    pub(super) struct DecisionSlot {
797        pub(super) key: RequestKey,
798        entry: Arc<DecisionEntry>,
799        entry_id: String,
800        generation: u64,
801        pub(super) bytes: u64,
802        seq: u64,
803    }
804
805    /// A decision entry that may be served.
806    #[derive(Debug, Clone)]
807    pub(crate) struct DecisionHit {
808        pub entry: Arc<DecisionEntry>,
809        pub entry_id: String,
810        pub age_ms: u64,
811        /// The store's generations, for an entry of the persistent tier.
812        pub generations: Option<EntryGenerations>,
813    }
814
815    /// An entry's id: a digest of its key material and its source run
816    /// record's digest, the same in either tier.
817    pub(crate) fn decision_entry_id(key: &RequestKey, source: &ContentDigest) -> String {
818        let mut material = key.material.clone();
819        part(&mut material, source.as_str().as_bytes());
820        tagged_digest(DECISION_ENTRY_TAG, &material)
821    }
822
823    /// Why a decision lookup did not hit; each is a diagnostic (spec 044, 3.9).
824    #[derive(Debug, Clone, Copy, PartialEq, Eq)]
825    pub(crate) enum DecisionMiss {
826        Absent,
827        Collision,
828        GenerationAdvanced,
829        FromFuture,
830        Expired,
831    }
832
833    impl DecisionMiss {
834        pub(crate) fn diagnostic(self) -> &'static str {
835            match self {
836                DecisionMiss::Absent => "decision cache miss",
837                DecisionMiss::Collision => "entry_collision",
838                DecisionMiss::GenerationAdvanced => "generation_advanced",
839                DecisionMiss::FromFuture => "entry_from_future",
840                DecisionMiss::Expired => "entry_expired",
841            }
842        }
843    }
844
845    /// The bytes a decision entry counts against `max_entry_bytes`: its key
846    /// material, each reused request's key material, canonical output and
847    /// source attempt id, and each stored volatile entry's canonical bytes.
848    pub(crate) fn decision_entry_bytes(
849        key: &RequestKey,
850        requests: &ReusedRequests,
851        volatile: &[Entry],
852    ) -> Result<u64, CacheRefusal> {
853        let mut bytes = key.material.len() as u64;
854        for e in volatile {
855            let entry = record_canonical_bytes(e).map_err(|_| CacheRefusal::Uncanonical)?;
856            bytes = bytes.saturating_add(entry.len() as u64);
857        }
858        for r in requests.values() {
859            let output =
860                record_canonical_bytes(&r.output).map_err(|_| CacheRefusal::Uncanonical)?;
861            bytes = bytes
862                .saturating_add(r.key.material.len() as u64)
863                .saturating_add(output.len() as u64)
864                .saturating_add(r.source_attempt_id.len() as u64);
865        }
866        Ok(bytes)
867    }
868
869    impl Optimizer {
870        /// Read the decision key's entry (spec 044, 3.6.3 item 1). An entry is
871        /// served only with the request's own key material, the namespace's
872        /// current generation, and an age by evaluation time within `ttl_ms`.
873        pub(crate) fn lookup_decision(
874            &self,
875            key: &RequestKey,
876            now: Timestamp,
877            ttl_ms: u64,
878        ) -> Result<DecisionHit, DecisionMiss> {
879            let mut namespaces = self.lock();
880            let Some(ns) = namespaces.get_mut(&key.namespace) else {
881                return Err(DecisionMiss::Absent);
882            };
883            let Some(slot) = ns.decisions.get(&key.id) else {
884                return Err(DecisionMiss::Absent);
885            };
886            if slot.key.material != key.material {
887                return Err(DecisionMiss::Collision);
888            }
889            if slot.generation != ns.generation {
890                ns.decisions.remove(&key.id);
891                return Err(DecisionMiss::GenerationAdvanced);
892            }
893            let (now, created) = (now.as_ms(), slot.entry.created.as_ms());
894            if now < created {
895                return Err(DecisionMiss::FromFuture);
896            }
897            let age_ms = u64::try_from(now - created).unwrap_or(u64::MAX);
898            if age_ms > ttl_ms {
899                ns.decisions.remove(&key.id);
900                return Err(DecisionMiss::Expired);
901            }
902            Ok(DecisionHit {
903                entry: slot.entry.clone(),
904                entry_id: slot.entry_id.clone(),
905                age_ms,
906                generations: None,
907            })
908        }
909
910        /// Admit a decision entry under the generation captured when the
911        /// decision started (spec 044, 3.6.4), evicting the namespace's oldest
912        /// decision entries beyond `max_entries`.
913        pub(crate) fn insert_decision(
914            &self,
915            policy: &DecisionCachePolicy,
916            key: RequestKey,
917            entry: DecisionEntry,
918            captured_generation: u64,
919        ) -> Result<String, CacheRefusal> {
920            let bytes = decision_entry_bytes(&key, &entry.requests, &entry.volatile)?;
921            if bytes > policy.max_entry_bytes {
922                return Err(CacheRefusal::EntryTooLarge);
923            }
924            // Present and positive exactly when the policy has a memory tier
925            // (spec 044, 3.1); otherwise refused, never bounded by a default.
926            let Some(max_entries) = policy.max_entries.filter(|&n| n > 0) else {
927                return Err(CacheRefusal::NoCapacity);
928            };
929            let mut namespaces = self.lock();
930            let ns = namespaces.entry(key.namespace.clone()).or_default();
931            if ns.generation != captured_generation {
932                return Err(CacheRefusal::StaleGeneration);
933            }
934            if ns
935                .decisions
936                .get(&key.id)
937                .is_some_and(|slot| slot.key.material != key.material)
938            {
939                return Err(CacheRefusal::Collision);
940            }
941            ns.decisions.remove(&key.id);
942            while ns.decisions.len() as u64 >= max_entries {
943                let Some((id, seq)) = ns.decision_fifo.pop_front() else {
944                    break;
945                };
946                if ns.decisions.get(&id).is_some_and(|slot| slot.seq == seq) {
947                    ns.decisions.remove(&id);
948                }
949            }
950            if ns.decision_fifo.len() > 2 * ns.decisions.len() + 16 {
951                let decisions = &ns.decisions;
952                ns.decision_fifo
953                    .retain(|(id, seq)| decisions.get(id).is_some_and(|slot| slot.seq == *seq));
954            }
955            let entry_id = decision_entry_id(&key, &entry.source_run_record);
956            let seq = ns.next_seq;
957            ns.next_seq += 1;
958            ns.decision_fifo.push_back((key.id.clone(), seq));
959            let generation = ns.generation;
960            ns.decisions.insert(
961                key.id.clone(),
962                DecisionSlot {
963                    key,
964                    entry: Arc::new(entry),
965                    entry_id: entry_id.clone(),
966                    generation,
967                    bytes,
968                    seq,
969                },
970            );
971            Ok(entry_id)
972        }
973    }
974
975    type At = (String, Vec<String>);
976
977    /// What one decision did with the decision cache, filled while it runs.
978    #[derive(Debug, Default)]
979    pub(crate) struct DecisionLog {
980        /// The 017 key of each request outside the closure, on a miss.
981        pub keys: BTreeMap<At, RequestKey>,
982        /// Each output supplied to the core, on a miss.
983        pub outputs: BTreeMap<At, RawOutput>,
984        pub reused: Vec<At>,
985        pub refreshed: Vec<At>,
986        pub mismatched: Vec<At>,
987        pub diagnostics: Vec<String>,
988    }
989
990    /// One decision's part in the decision-result cache (spec 044, 3.6): its
991    /// key, the entry it is served from on a hit, and its log.
992    #[derive(Debug)]
993    pub(crate) struct DecisionRun {
994        /// Steps whose requests are dispatched again on a hit (3.6.2).
995        pub closure: BTreeSet<String>,
996        /// Absent when the decision is unkeyed (3.5 item 3) or live.
997        pub key: Option<RequestKey>,
998        /// The namespace generation captured before the lookup; admission
999        /// uses it (3.6.4).
1000        pub generation: u64,
1001        /// The store's generations, captured by the persistent read before
1002        /// the work began; a write needs them (3.4.3).
1003        pub captured: Option<EntryGenerations>,
1004        pub hit: Option<DecisionHit>,
1005        /// Under `admission: refuse`, an unkeyed decision's requests are
1006        /// refused with this diagnostic and never dispatched.
1007        pub refused: Option<String>,
1008        outcome: rustev_contract::optimization::DecisionCacheOutcome,
1009        log: Mutex<DecisionLog>,
1010        on_refresh_failure: RefreshFailurePolicy,
1011        /// Volatile inputs the core did not accept from the new snapshot.
1012        failed_inputs: Vec<String>,
1013        /// Stored volatile entries supplied under
1014        /// `serve_cached_within_freshness` (3.6.5).
1015        substituted: Vec<SubstitutedInput>,
1016        /// The served run abandoned under `recompute` (3.6.5), omitted when
1017        /// it dispatched nothing; `recomputed` says a rerun followed.
1018        abandoned: Option<rustev_contract::optimization::AbandonedRun>,
1019        recomputed: bool,
1020    }
1021
1022    /// The digest the run record names a substituted volatile entry by
1023    /// (spec 044, 3.6.6): the one the decision key takes of a snapshot entry.
1024    fn entry_digest(e: &Entry) -> Option<ContentDigest> {
1025        let bytes = record_canonical_bytes(e).ok()?;
1026        ContentDigest::parse(&tagged_digest(DECISION_INPUT_TAG, &bytes)).ok()
1027    }
1028
1029    /// What failed in a served decision's refresh (spec 044, 3.6.5): each
1030    /// volatile input the core did not accept from the new snapshot, and
1031    /// each refreshed request that failed or was not supplied.
1032    fn failures(
1033        failed_inputs: &[String],
1034        log: &DecisionLog,
1035        requests: &[rustev_contract::run::RequestRecord],
1036        acquisitions: &[AcquisitionRecord],
1037    ) -> Vec<String> {
1038        use rustev_contract::acquisition_record::AcquisitionResult;
1039        use rustev_contract::run::RequestResult;
1040        let mut failed: Vec<String> = failed_inputs
1041            .iter()
1042            .map(|v| format!("input {v:?}"))
1043            .collect();
1044        failed.extend(log.refreshed.iter().filter_map(|(step, instance)| {
1045            requests
1046                .iter()
1047                .find(|r| &r.step == step && &r.instance == instance)
1048                .filter(|r| {
1049                    matches!(
1050                        r.result,
1051                        RequestResult::Failed(_) | RequestResult::NotSupplied
1052                    )
1053                })
1054                .map(|r| format!("step {:?}", r.step))
1055        }));
1056        // Every acquisition of a served decision is refreshed (3.6.2), and
1057        // one that fails is named once per step.
1058        let mut steps = BTreeSet::new();
1059        failed.extend(
1060            acquisitions
1061                .iter()
1062                .filter(|a| !matches!(a.result, AcquisitionResult::Page))
1063                .filter(|a| steps.insert(&a.step))
1064                .map(|a| format!("acquisition step {:?}", a.step)),
1065        );
1066        failed
1067    }
1068
1069    impl DecisionRun {
1070        /// Derive the key and read the memory tier after admission and before
1071        /// the core starts (spec 044, 3.6.3 item 1); [`DecisionRun::read_store`]
1072        /// reads the persistent tier next. `None` for a plan without the
1073        /// decision cache.
1074        pub(crate) fn start(
1075            optimizer: &Optimizer,
1076            plan: &crate::PreparedPlan,
1077            req: &crate::DecisionRequest,
1078            dimensions: &BTreeMap<String, rustev_contract::scope::Handle>,
1079            live: bool,
1080        ) -> Option<DecisionRun> {
1081            use rustev_contract::optimization::{DecisionCacheOutcome, OptimizationAdmission};
1082            use rustev_contract::plan::PlanExecution;
1083            let policy = plan.optimization.as_ref()?;
1084            let cache = policy.decision_cache.as_ref()?;
1085            let PlanExecution::Declared(d) = &plan.compiled.plan.execution else {
1086                return None;
1087            };
1088            let compiled = d.decision_cache.as_ref()?;
1089            let mut closure: BTreeSet<String> = compiled.refresh_closure.iter().cloned().collect();
1090            // A generate step's output screening requests are its own, so
1091            // they follow it into the closure (spec 044, 3.6.2, increment 5
1092            // part e).
1093            for s in &plan.compiled.plan.steps {
1094                if let rustev_contract::plan::PlanStepDetail::Generate(g) = &s.detail
1095                    && closure.contains(&s.id)
1096                {
1097                    closure.extend(
1098                        g.output.fields.iter().map(|f| {
1099                            rustev_contract::generation::output_screen_step(&s.id, &f.name)
1100                        }),
1101                    );
1102                }
1103            }
1104            let mut run = DecisionRun {
1105                closure,
1106                key: None,
1107                generation: 0,
1108                captured: None,
1109                hit: None,
1110                refused: None,
1111                outcome: DecisionCacheOutcome::Miss,
1112                log: Mutex::default(),
1113                on_refresh_failure: cache.on_refresh_failure,
1114                failed_inputs: vec![],
1115                substituted: vec![],
1116                abandoned: None,
1117                recomputed: false,
1118            };
1119            let log = run.log.get_mut().unwrap_or_else(|e| e.into_inner());
1120            // A live re-execution observes fresh backend work (spec 026, 3.4).
1121            if live {
1122                run.outcome = DecisionCacheOutcome::Bypassed;
1123                log.diagnostics
1124                    .push("live re-execution observes fresh backend work".into());
1125                return Some(run);
1126            }
1127            let key = decision_key(DecisionKeyParts {
1128                schema: &compiled.decision_key_schema,
1129                plan_id: &plan.compiled.id,
1130                profile: &profile_of(&plan.compiled.plan),
1131                policy,
1132                principal: &req.principal_handle,
1133                dimensions,
1134                snapshot: &req.snapshot,
1135            });
1136            match key {
1137                Err(unkeyed) => {
1138                    let diagnostic = unkeyed.diagnostic();
1139                    log.diagnostics.push(diagnostic.clone());
1140                    if policy.admission == OptimizationAdmission::Refuse {
1141                        run.outcome = DecisionCacheOutcome::Refused;
1142                        run.refused = Some(diagnostic);
1143                    } else {
1144                        run.outcome = DecisionCacheOutcome::Bypassed;
1145                    }
1146                }
1147                Ok(key) => {
1148                    run.generation = optimizer.generation(&key.namespace);
1149                    if cache.tiers.contains(&DecisionCacheTier::Memory) {
1150                        match optimizer.lookup_decision(&key, req.evaluation_time, cache.ttl_ms) {
1151                            Ok(hit) => run.hit = Some(hit),
1152                            Err(miss) => log.diagnostics.push(miss.diagnostic().into()),
1153                        }
1154                    }
1155                    run.key = Some(key);
1156                }
1157            }
1158            Some(run)
1159        }
1160
1161        /// After a memory miss, the persistent tier (spec 044, 3.3). A served
1162        /// entry is admitted to the memory tier under its bounds. A store
1163        /// failure under `fail_closed` leaves the decision neither served
1164        /// nor admitted, `refused` or `bypassed` under `admission`; under
1165        /// `continue_without_cache` it is a miss (3.4.2 and 3.6.6).
1166        pub(crate) async fn read_store(
1167            &mut self,
1168            optimizer: &Optimizer,
1169            policy: &OptimizationPolicy,
1170            cache: &DecisionCachePolicy,
1171            calls: &crate::store::StoreCtx<'_>,
1172            now: Timestamp,
1173        ) {
1174            use crate::store::{DecisionRead, read_decision};
1175            use rustev_contract::optimization::{
1176                DecisionCacheOutcome, OptimizationAdmission, StoreFailurePolicy,
1177            };
1178            if !cache.tiers.contains(&DecisionCacheTier::Persistent)
1179                || self.hit.is_some()
1180                || self.refused.is_some()
1181            {
1182                return;
1183            }
1184            let Some(key) = self.key.clone() else {
1185                return;
1186            };
1187            match read_decision(calls, &key, now).await {
1188                DecisionRead::Hit {
1189                    entry,
1190                    age_ms,
1191                    generations,
1192                } => {
1193                    let entry_id = decision_entry_id(&key, &entry.source_run_record);
1194                    let entry = Arc::new(entry);
1195                    if cache.tiers.contains(&DecisionCacheTier::Memory)
1196                        && let Err(refusal) =
1197                            optimizer.insert_decision(cache, key, (*entry).clone(), self.generation)
1198                    {
1199                        self.diagnose(crate::driver::cache_refusal_detail(refusal).into());
1200                    }
1201                    self.hit = Some(DecisionHit {
1202                        entry,
1203                        entry_id,
1204                        age_ms,
1205                        generations: Some(generations),
1206                    });
1207                }
1208                DecisionRead::Miss {
1209                    captured,
1210                    diagnostic,
1211                } => {
1212                    self.captured = captured;
1213                    self.diagnose(diagnostic);
1214                }
1215                DecisionRead::Failed(detail) => {
1216                    self.diagnose(detail.clone());
1217                    if calls.persistent.on_store_failure == StoreFailurePolicy::FailClosed {
1218                        self.key = None;
1219                        if policy.admission == OptimizationAdmission::Refuse {
1220                            self.outcome = DecisionCacheOutcome::Refused;
1221                            self.refused = Some(detail);
1222                        } else {
1223                            self.outcome = DecisionCacheOutcome::Bypassed;
1224                        }
1225                    }
1226                }
1227            }
1228        }
1229
1230        pub(crate) fn log(&self) -> MutexGuard<'_, DecisionLog> {
1231            self.log.lock().unwrap_or_else(|e| e.into_inner())
1232        }
1233
1234        /// An output supplied to the core or the generation stage, which a
1235        /// judged miss may store (spec 044, 3.6.4).
1236        pub(crate) fn supplied(&self, step: &str, instance: &[String], output: &RawOutput) {
1237            if self.hit.is_none() {
1238                self.log()
1239                    .outputs
1240                    .insert((step.to_string(), instance.to_vec()), output.clone());
1241            }
1242        }
1243
1244        /// The volatile inputs of a served decision, once the core has
1245        /// validated the new snapshot (spec 044, 3.6.5): each one it did not
1246        /// accept is a failed refresh. Under `serve_cached_within_freshness`
1247        /// an input with no accepted entry is given the entry's stored
1248        /// snapshot entries with their original as-of times, and the snapshot
1249        /// to restart the core over is returned, so the core's freshness check
1250        /// decides. Any other failure is supplied as it is. A decision under
1251        /// capture is never substituted: its capture could not carry the
1252        /// stored entries a replay would need.
1253        pub(crate) fn refresh_inputs(
1254            &mut self,
1255            volatile: &[String],
1256            ev: &rustev_core::evaluate::Evaluation<'_>,
1257            snapshot: &Snapshot,
1258            capturing: bool,
1259        ) -> Option<Snapshot> {
1260            use rustev_contract::judgment::Unresolved;
1261            let entry = self.hit.as_ref()?.entry.clone();
1262            let mut missing = vec![];
1263            for v in volatile.iter().filter(|v| !v.starts_with("step:")) {
1264                if let Err(u) = ev.status_of(&format!("input:{v}")) {
1265                    if matches!(u, Unresolved::MissingEvidence { .. }) {
1266                        missing.push(v);
1267                    }
1268                    self.failed_inputs.push(v.clone());
1269                }
1270            }
1271            if self.failed_inputs.is_empty() {
1272                return None;
1273            }
1274            match self.on_refresh_failure {
1275                // Under `recompute` the decision runs again once the served
1276                // run has ended (3.6.5).
1277                RefreshFailurePolicy::Unresolved | RefreshFailurePolicy::Recompute => None,
1278                RefreshFailurePolicy::ServeCachedWithinFreshness if capturing => {
1279                    self.diagnose(
1280                        "stored volatile entries are not substituted under capture".into(),
1281                    );
1282                    None
1283                }
1284                RefreshFailurePolicy::ServeCachedWithinFreshness => {
1285                    let mut substituted = snapshot.clone();
1286                    for input in missing {
1287                        let stored: Vec<&Entry> = entry
1288                            .volatile
1289                            .iter()
1290                            .filter(|e| &e.field == input)
1291                            .collect();
1292                        let digests: Option<Vec<_>> =
1293                            stored.iter().map(|e| entry_digest(e)).collect();
1294                        // An input the entry holds nothing for, or nothing
1295                        // canonical, stays as the new snapshot supplies it.
1296                        let Some(digests) = digests.filter(|d| !d.is_empty()) else {
1297                            continue;
1298                        };
1299                        self.substituted
1300                            .extend(digests.into_iter().map(|entry_digest| SubstitutedInput {
1301                                input: input.clone(),
1302                                entry_digest,
1303                            }));
1304                        substituted.entries.extend(stored.into_iter().cloned());
1305                    }
1306                    (!self.substituted.is_empty()).then_some(substituted)
1307                }
1308            }
1309        }
1310
1311        /// Whether a served run's refresh failed under `recompute`, so the
1312        /// decision runs again (spec 044, 3.6.5). A rerun never recomputes.
1313        pub(crate) fn recomputes(
1314            &self,
1315            requests: &[rustev_contract::run::RequestRecord],
1316            acquisitions: &[AcquisitionRecord],
1317        ) -> bool {
1318            self.on_refresh_failure == RefreshFailurePolicy::Recompute
1319                && self.hit.is_some()
1320                && !failures(&self.failed_inputs, &self.log(), requests, acquisitions).is_empty()
1321        }
1322
1323        /// Abandon the served run once every request it dispatched has ended
1324        /// (spec 044, 3.6.5): its refreshed and mismatched requests and its
1325        /// acquisitions are kept with their attempts, and the rerun is a
1326        /// miss with nothing supplied from the entry, which it may then
1327        /// admit (3.6.4).
1328        pub(crate) fn abandon(
1329            &mut self,
1330            requests: Vec<rustev_contract::run::RequestRecord>,
1331            acquisitions: Vec<AcquisitionRecord>,
1332        ) {
1333            let log = std::mem::take(&mut *self.log());
1334            let failed = failures(&self.failed_inputs, &log, &requests, &acquisitions);
1335            let dispatched: BTreeSet<&At> = log.refreshed.iter().chain(&log.mismatched).collect();
1336            let requests: Vec<_> = requests
1337                .into_iter()
1338                .filter(|r| dispatched.contains(&(r.step.clone(), r.instance.clone())))
1339                .collect();
1340            self.abandoned = (!requests.is_empty() || !acquisitions.is_empty()).then_some(
1341                rustev_contract::optimization::AbandonedRun {
1342                    requests,
1343                    acquisitions,
1344                },
1345            );
1346            self.hit = None;
1347            self.recomputed = true;
1348            self.outcome = rustev_contract::optimization::DecisionCacheOutcome::Miss;
1349            self.failed_inputs.clear();
1350            self.substituted.clear();
1351            self.diagnose(format!("refresh_failed: {}", failed.join(", ")));
1352        }
1353
1354        /// The core refused the substituted snapshot; the failure is supplied
1355        /// as it is (spec 044, 3.6.5).
1356        pub(crate) fn unsubstitute(&mut self, detail: String) {
1357            self.substituted.clear();
1358            self.diagnose(format!("stored volatile entries not substituted: {detail}"));
1359        }
1360
1361        /// The run record member (spec 044, 3.6.6). A served decision's failed
1362        /// refresh names the policy applied, which is the plan's (3.6.5). An
1363        /// acquisition is always inside the closure (the compiler refuses one
1364        /// outside it), so a served decision that acquired is `refreshed`; its
1365        /// pages are listed in the record's `acquisitions`, not in the request
1366        /// indices.
1367        pub(crate) fn record(
1368            &self,
1369            requests: &[rustev_contract::run::RequestRecord],
1370            acquisitions: &[AcquisitionRecord],
1371        ) -> rustev_contract::optimization::DecisionCacheRecord {
1372            use rustev_contract::optimization::{
1373                DecisionCacheOutcome, DecisionCacheRecord, MAX_OPTIMIZATION_DIAGNOSTICS,
1374            };
1375            let log = self.log();
1376            let index = |list: &[At]| {
1377                let mut found: Vec<u32> = list
1378                    .iter()
1379                    .filter_map(|(step, instance)| {
1380                        requests
1381                            .iter()
1382                            .position(|r| &r.step == step && &r.instance == instance)
1383                    })
1384                    .map(|n| n as u32)
1385                    .collect();
1386                found.sort_unstable();
1387                found
1388            };
1389            let mut record = DecisionCacheRecord {
1390                outcome: self.outcome,
1391                entry_id: None,
1392                source_run_record: None,
1393                age_ms: None,
1394                generations: None,
1395                reused: index(&log.reused),
1396                refreshed: index(&log.refreshed),
1397                mismatched: index(&log.mismatched),
1398                substituted: vec![],
1399                on_refresh_failure: None,
1400                abandoned: self.abandoned.clone(),
1401                diagnostics: vec![],
1402            };
1403            if self.recomputed {
1404                record.on_refresh_failure = Some(RefreshFailurePolicy::Recompute);
1405            }
1406            let mut diagnostics = log.diagnostics.clone();
1407            if let Some(hit) = &self.hit {
1408                record.outcome = if record.refreshed.is_empty()
1409                    && record.mismatched.is_empty()
1410                    && acquisitions.is_empty()
1411                {
1412                    DecisionCacheOutcome::Hit
1413                } else {
1414                    DecisionCacheOutcome::Refreshed
1415                };
1416                record.entry_id = Some(hit.entry_id.clone());
1417                record.source_run_record = Some(hit.entry.source_run_record.clone());
1418                record.age_ms = Some(hit.age_ms);
1419                record.generations = hit.generations;
1420                record.substituted = self.substituted.clone();
1421                let failed = failures(&self.failed_inputs, &log, requests, acquisitions);
1422                if !failed.is_empty() {
1423                    record.on_refresh_failure = Some(self.on_refresh_failure);
1424                    diagnostics.push(format!("refresh_failed: {}", failed.join(", ")));
1425                }
1426            }
1427            for d in diagnostics {
1428                let d = crate::wait::cut(&d, crate::driver::DETAIL_BYTES);
1429                match record.diagnostics.len() {
1430                    n if n + 1 < MAX_OPTIMIZATION_DIAGNOSTICS => record.diagnostics.push(d),
1431                    n if n + 1 == MAX_OPTIMIZATION_DIAGNOSTICS => {
1432                        record
1433                            .diagnostics
1434                            .push("further diagnostics dropped".into());
1435                    }
1436                    _ => {}
1437                }
1438            }
1439            record
1440        }
1441
1442        /// The entry a judged miss may admit (spec 044, 3.6.4): every request
1443        /// outside the closure was supplied a validated output of target 0
1444        /// without a fallback transition, and the entry fits
1445        /// `max_entry_bytes`. A served, unkeyed or live decision admits
1446        /// nothing. `Err` holds the diagnostic of an admission not made.
1447        pub(crate) fn admissible(
1448            &self,
1449            requests: &[rustev_contract::run::RequestRecord],
1450            volatile: &[Entry],
1451            policy: &rustev_contract::optimization::DecisionCachePolicy,
1452        ) -> Result<Option<(RequestKey, ReusedRequests)>, String> {
1453            use rustev_contract::run::{RequestResult, Transition};
1454            let (Some(key), None) = (&self.key, &self.hit) else {
1455                return Ok(None);
1456            };
1457            let log = self.log();
1458            let mut reused = ReusedRequests::new();
1459            for r in requests.iter().filter(|r| !self.closure.contains(&r.step)) {
1460                let at = (r.step.clone(), r.instance.clone());
1461                let fallback = r
1462                    .transitions
1463                    .iter()
1464                    .any(|t| matches!(t, Transition::Fallback { .. }));
1465                let source = r
1466                    .supplied_from()
1467                    .filter(|f| f.target == 0 && matches!(r.result, RequestResult::Output { .. }))
1468                    .and_then(|f| {
1469                        f.attempt
1470                            .map(|a| a.attempt_id.clone())
1471                            .or(f.reused_from.map(String::from))
1472                    });
1473                match (fallback, source, log.keys.get(&at), log.outputs.get(&at)) {
1474                    (false, Some(source_attempt_id), Some(k), Some(output)) => {
1475                        reused.insert(
1476                            at,
1477                            ReusedRequest {
1478                                key: k.clone(),
1479                                output: output.clone(),
1480                                source_attempt_id,
1481                            },
1482                        );
1483                    }
1484                    _ => {
1485                        return Err(format!(
1486                            "decision cache not admitted: step {:?} has no validated target 0 output",
1487                            r.step
1488                        ));
1489                    }
1490                }
1491            }
1492            match decision_entry_bytes(key, &reused, volatile) {
1493                Ok(bytes) if bytes <= policy.max_entry_bytes => Ok(Some((key.clone(), reused))),
1494                Ok(_) | Err(CacheRefusal::EntryTooLarge) => {
1495                    Err("decision cache not admitted: entry exceeds max_entry_bytes".into())
1496                }
1497                Err(_) => {
1498                    Err("decision cache not admitted: an output has no canonical form".into())
1499                }
1500            }
1501        }
1502
1503        /// Record that this decision's evidence carries an admission not made.
1504        pub(crate) fn diagnose(&self, diagnostic: String) {
1505            self.log().diagnostics.push(diagnostic);
1506        }
1507    }
1508
1509    #[cfg(test)]
1510    mod tests {
1511        use rustev_contract::definition::ProvenanceClass;
1512        use rustev_contract::optimization::{
1513            DecisionCacheTier, ExpiryBasis, OptimizationAdmission, RefreshFailurePolicy,
1514        };
1515        use rustev_contract::snapshot::Entry;
1516
1517        use super::super::InvalidationSelector;
1518        use super::*;
1519
1520        fn plan() -> (PlanId, OptimizationPolicy) {
1521            (
1522                PlanId::parse(&format!("sha256:{}", "1".repeat(64))).unwrap(),
1523                OptimizationPolicy {
1524                    key_schema_version: 1,
1525                    cache_namespace_version: 1,
1526                    admission: OptimizationAdmission::Refuse,
1527                    expiry: ExpiryBasis::InjectedRuntimeTime,
1528                    batch: None,
1529                    memory_cache: None,
1530                    shared_calls: None,
1531                    persistent_cache: None,
1532                    decision_cache: None,
1533                },
1534            )
1535        }
1536
1537        fn cache(max_entries: u64, max_entry_bytes: u64) -> DecisionCachePolicy {
1538            DecisionCachePolicy {
1539                tiers: vec![DecisionCacheTier::Memory],
1540                max_entry_bytes,
1541                max_entries: Some(max_entries),
1542                ttl_ms: 100,
1543                volatile: vec!["live".into()],
1544                key_dimensions: vec!["revision".into()],
1545                on_refresh_failure: RefreshFailurePolicy::Unresolved,
1546            }
1547        }
1548
1549        /// The schema the compiler writes for `cache()` over the inputs
1550        /// `stable` and `live` (spec 044, 3.6.1).
1551        fn schema() -> Vec<String> {
1552            [
1553                "plan_id",
1554                "key_schema_version",
1555                "cache_namespace_version",
1556                "runtime_version",
1557                "scope",
1558                "profile",
1559                "dimension:revision",
1560                "input:stable",
1561            ]
1562            .map(String::from)
1563            .to_vec()
1564        }
1565
1566        fn snapshot(stable: i64, live: i64) -> Snapshot {
1567            let entry = |field: &str, value: i64| Entry {
1568                field: field.into(),
1569                value: value.into(),
1570                provenance: ProvenanceClass::SystemOfRecord,
1571                as_of_ms: Timestamp::from_ms(1).unwrap(),
1572                source: "SYNTHETIC".into(),
1573                grade: None,
1574                producer: None,
1575            };
1576            Snapshot {
1577                schema: rustev_contract::schema::SNAPSHOT.into(),
1578                entries: vec![entry("stable", stable), entry("live", live)],
1579            }
1580        }
1581
1582        fn dims(revision: &str) -> BTreeMap<String, Handle> {
1583            BTreeMap::from([("revision".to_string(), Handle::new(revision).unwrap())])
1584        }
1585
1586        fn key_with(
1587            schema: &[String],
1588            profile: &str,
1589            principal: &[u8],
1590            dimensions: &BTreeMap<String, Handle>,
1591            snapshot: &Snapshot,
1592        ) -> Result<RequestKey, Unkeyed> {
1593            let (plan_id, policy) = plan();
1594            decision_key(DecisionKeyParts {
1595                schema,
1596                plan_id: &plan_id,
1597                profile,
1598                policy: &policy,
1599                principal,
1600                dimensions,
1601                snapshot,
1602            })
1603        }
1604
1605        fn key(principal: &[u8]) -> RequestKey {
1606            key_with(&schema(), "none", principal, &dims("r1"), &snapshot(1, 1)).unwrap()
1607        }
1608
1609        fn at(ms: i64) -> Timestamp {
1610            Timestamp::from_ms(ms).unwrap()
1611        }
1612
1613        /// An entry reusing one request, created at `created`.
1614        fn entry(created: i64, output_bytes: usize) -> DecisionEntry {
1615            let reused = ReusedRequest {
1616                key: key(b"request"),
1617                output: RawOutput::Label("x".repeat(output_bytes)),
1618                source_attempt_id: "d1/topic/0/1".into(),
1619            };
1620            DecisionEntry {
1621                requests: BTreeMap::from([(("topic".to_string(), vec![]), reused)]),
1622                volatile: vec![],
1623                source_run_record: ContentDigest::parse(&format!("sha256:{}", "2".repeat(64)))
1624                    .unwrap(),
1625                created: at(created),
1626            }
1627        }
1628
1629        /// Spec 044, 3.5 and 3.6.1: the decision key follows the compiled
1630        /// schema. The profile, scope, each dimension and each listed input's
1631        /// snapshot entries are members; an input the schema leaves out (a
1632        /// volatile one) is not. A missing dimension, an undeclared one and
1633        /// an unknown member leave the decision unkeyed.
1634        #[test]
1635        fn the_decision_key_follows_its_schema() {
1636            let (plan_id, policy) = plan();
1637            let s = schema();
1638            let base = key(b"tenant-a");
1639            assert_eq!(base.namespace, namespace(&plan_id, &policy));
1640            assert_eq!(base.scope_id, scope_id(b"tenant-a"));
1641            // A volatile input is not a member: a changed live value keeps
1642            // the key.
1643            assert_eq!(
1644                key_with(&s, "none", b"tenant-a", &dims("r1"), &snapshot(1, 2)).unwrap(),
1645                base
1646            );
1647            for other in [
1648                key_with(&s, r#"["p"]"#, b"tenant-a", &dims("r1"), &snapshot(1, 1)),
1649                key_with(&s, "none", b"tenant-b", &dims("r1"), &snapshot(1, 1)),
1650                key_with(&s, "none", b"tenant-a", &dims("r2"), &snapshot(1, 1)),
1651                key_with(&s, "none", b"tenant-a", &dims("r1"), &snapshot(2, 1)),
1652            ] {
1653                assert_ne!(other.unwrap().id, base.id);
1654            }
1655            let missing = key_with(&s, "none", b"tenant-a", &BTreeMap::new(), &snapshot(1, 1));
1656            assert_eq!(
1657                missing.unwrap_err().diagnostic(),
1658                "key_dimension_missing: \"revision\""
1659            );
1660            let mut extra = dims("r1");
1661            extra.insert("region".into(), Handle::new("eu").unwrap());
1662            assert_eq!(
1663                key_with(&s, "none", b"tenant-a", &extra, &snapshot(1, 1))
1664                    .unwrap_err()
1665                    .diagnostic(),
1666                "key_ineligible: dimension \"region\" is not declared by the plan"
1667            );
1668            let mut unknown = s.clone();
1669            unknown.push("clock".into());
1670            assert_eq!(
1671                key_with(&unknown, "none", b"tenant-a", &dims("r1"), &snapshot(1, 1))
1672                    .unwrap_err()
1673                    .diagnostic(),
1674                "key_ineligible: unknown key member \"clock\""
1675            );
1676        }
1677
1678        /// Spec 044, decision 6: a decision entry is served by evaluation
1679        /// time up to `ttl_ms`; one created after the request's evaluation
1680        /// time is a miss and kept; an expired one is a miss and removed.
1681        #[test]
1682        fn a_decision_entry_expires_by_evaluation_time() {
1683            let o = Optimizer::default();
1684            let k = key(b"tenant-a");
1685            let id = o
1686                .insert_decision(&cache(4, 4_096), k.clone(), entry(1_000, 1), 0)
1687                .unwrap();
1688            let hit = o.lookup_decision(&k, at(1_005), 100).unwrap();
1689            assert_eq!((hit.entry_id.as_str(), hit.age_ms), (id.as_str(), 5));
1690            assert_eq!(hit.entry.requests.len(), 1);
1691            assert_eq!(o.lookup_decision(&k, at(1_100), 100).unwrap().age_ms, 100);
1692            assert_eq!(
1693                o.lookup_decision(&k, at(999), 100).unwrap_err(),
1694                DecisionMiss::FromFuture
1695            );
1696            assert!(o.lookup_decision(&k, at(1_000), 100).is_ok(), "kept");
1697            assert_eq!(
1698                o.lookup_decision(&k, at(1_101), 100).unwrap_err(),
1699                DecisionMiss::Expired
1700            );
1701            assert_eq!(
1702                o.lookup_decision(&k, at(1_000), 100).unwrap_err(),
1703                DecisionMiss::Absent,
1704                "removed"
1705            );
1706            // The entry id binds the key and the source record, and each
1707            // miss names its diagnostic (spec 044, 3.9).
1708            assert_ne!(id, k.id);
1709            let codes = [
1710                DecisionMiss::Absent,
1711                DecisionMiss::Collision,
1712                DecisionMiss::GenerationAdvanced,
1713                DecisionMiss::FromFuture,
1714                DecisionMiss::Expired,
1715            ]
1716            .map(DecisionMiss::diagnostic);
1717            assert_eq!(
1718                codes,
1719                [
1720                    "decision cache miss",
1721                    "entry_collision",
1722                    "generation_advanced",
1723                    "entry_from_future",
1724                    "entry_expired"
1725                ]
1726            );
1727        }
1728
1729        /// Spec 044, 3.6.4 and 017 3.2: admission refuses an entry over
1730        /// `max_entry_bytes`, one captured under an earlier generation and
1731        /// one whose key id holds other material; a lookup with other
1732        /// material is a collision; the oldest entry goes beyond
1733        /// `max_entries`.
1734        #[test]
1735        fn admission_is_bounded_and_checked() {
1736            let o = Optimizer::default();
1737            let a = key(b"tenant-a");
1738            let small = decision_entry_bytes(&a, &entry(0, 1).requests, &[]).unwrap();
1739            let policy = cache(2, small);
1740            assert_eq!(
1741                o.insert_decision(&policy, a.clone(), entry(0, 2), 0),
1742                Err(CacheRefusal::EntryTooLarge)
1743            );
1744            // Stored volatile snapshot entries count too (spec 044, 3.6.5).
1745            let mut with_volatile = entry(0, 1);
1746            with_volatile.volatile = snapshot(1, 2).entries;
1747            assert_eq!(
1748                o.insert_decision(&policy, a.clone(), with_volatile, 0),
1749                Err(CacheRefusal::EntryTooLarge)
1750            );
1751            for max_entries in [None, Some(0)] {
1752                let empty = DecisionCachePolicy {
1753                    max_entries,
1754                    ..policy.clone()
1755                };
1756                assert_eq!(
1757                    o.insert_decision(&empty, a.clone(), entry(0, 1), 0),
1758                    Err(CacheRefusal::NoCapacity)
1759                );
1760            }
1761            o.insert_decision(&policy, a.clone(), entry(0, 1), 0)
1762                .unwrap();
1763            assert_eq!(
1764                o.insert_decision(&policy, a.clone(), entry(0, 1), 1),
1765                Err(CacheRefusal::StaleGeneration)
1766            );
1767            let mut other = a.clone();
1768            *other.material.last_mut().unwrap() ^= 1;
1769            assert_eq!(
1770                o.insert_decision(&policy, other.clone(), entry(0, 1), 0),
1771                Err(CacheRefusal::Collision)
1772            );
1773            assert_eq!(
1774                o.lookup_decision(&other, at(0), 100).unwrap_err(),
1775                DecisionMiss::Collision
1776            );
1777            // Replacing an entry under the same key is not an eviction.
1778            o.insert_decision(&policy, a.clone(), entry(0, 1), 0)
1779                .unwrap();
1780            let (b, c) = (key(b"tenant-b"), key(b"tenant-c"));
1781            o.insert_decision(&policy, b.clone(), entry(0, 1), 0)
1782                .unwrap();
1783            o.insert_decision(&policy, c.clone(), entry(0, 1), 0)
1784                .unwrap();
1785            assert_eq!(
1786                o.lookup_decision(&a, at(0), 100).unwrap_err(),
1787                DecisionMiss::Absent
1788            );
1789            assert!(o.lookup_decision(&b, at(0), 100).is_ok());
1790            assert!(o.lookup_decision(&c, at(0), 100).is_ok());
1791        }
1792
1793        /// Spec 017, 3.7: invalidating a scope removes its decision entries
1794        /// and counts them; the generation advances for the namespace, so
1795        /// no earlier entry is served and no earlier capture is admitted.
1796        #[test]
1797        fn invalidation_reaches_decision_entries() {
1798            let o = Optimizer::default();
1799            let (a, b) = (key(b"tenant-a"), key(b"tenant-b"));
1800            let policy = cache(4, 4_096);
1801            o.insert_decision(&policy, a.clone(), entry(0, 1), 0)
1802                .unwrap();
1803            o.insert_decision(&policy, b.clone(), entry(0, 1), 0)
1804                .unwrap();
1805            let bytes = decision_entry_bytes(&a, &entry(0, 1).requests, &[]).unwrap();
1806            let result = o.invalidate(&InvalidationSelector::Scope {
1807                namespace: a.namespace.clone(),
1808                scope_id: a.scope_id.clone(),
1809            });
1810            assert_eq!((result.removed_entries, result.removed_bytes), (1, bytes));
1811            assert_eq!(
1812                o.lookup_decision(&a, at(0), 100).unwrap_err(),
1813                DecisionMiss::Absent
1814            );
1815            assert_eq!(
1816                o.lookup_decision(&b, at(0), 100).unwrap_err(),
1817                DecisionMiss::GenerationAdvanced
1818            );
1819            assert_eq!(
1820                o.insert_decision(&policy, a.clone(), entry(0, 1), 0),
1821                Err(CacheRefusal::StaleGeneration)
1822            );
1823            o.insert_decision(&policy, a.clone(), entry(0, 1), result.generation)
1824                .unwrap();
1825            let all = o.invalidate(&InvalidationSelector::Namespace {
1826                namespace: a.namespace.clone(),
1827            });
1828            assert_eq!(all.removed_entries, 1);
1829        }
1830    }
1831}
1832
1833#[cfg(test)]
1834mod tests {
1835    use std::collections::BTreeMap;
1836
1837    use rustev_contract::ids::{ArtifactId, DescriptorId, PlanId};
1838    use rustev_contract::optimization::{EvictionOrder, ExpiryBasis, OptimizationAdmission};
1839
1840    use super::*;
1841
1842    fn digest(byte: char) -> String {
1843        format!("sha256:{}", byte.to_string().repeat(64))
1844    }
1845
1846    fn policy(plan: char) -> (PlanId, OptimizationPolicy) {
1847        (
1848            PlanId::parse(&digest(plan)).unwrap(),
1849            OptimizationPolicy {
1850                key_schema_version: 1,
1851                cache_namespace_version: 1,
1852                admission: OptimizationAdmission::Refuse,
1853                expiry: ExpiryBasis::InjectedRuntimeTime,
1854                batch: None,
1855                memory_cache: Some(memory(4096, 2, 2048, 2048)),
1856                shared_calls: None,
1857                persistent_cache: None,
1858                decision_cache: None,
1859            },
1860        )
1861    }
1862
1863    fn memory(max_bytes: u64, max_entries: u64, entry: u64, scope: u64) -> MemoryCachePolicy {
1864        MemoryCachePolicy {
1865            acquisition_pages: false,
1866            max_bytes,
1867            max_entries,
1868            max_entry_bytes: entry,
1869            max_scope_bytes: scope,
1870            ttl_ms: 10,
1871            eviction: EvictionOrder::Fifo,
1872        }
1873    }
1874
1875    fn key_in(plan: char, principal: &[u8], step: &str) -> RequestKey {
1876        let (plan_id, policy) = policy(plan);
1877        request_key(KeyParts {
1878            plan_id: &plan_id,
1879            profile: "none",
1880            policy: &policy,
1881            principal,
1882            backend_artifact: &ArtifactId::parse(&digest('2')).unwrap(),
1883            backend_descriptor: &DescriptorId::parse(&digest('3')).unwrap(),
1884            step,
1885            instance: &["one".into()],
1886            projection: b"projection",
1887        })
1888    }
1889
1890    fn key(principal: &[u8]) -> RequestKey {
1891        key_in('1', principal, "classify")
1892    }
1893
1894    fn output(label: &str) -> RawOutput {
1895        RawOutput::Scores(BTreeMap::from([(label.into(), 1.0)]))
1896    }
1897
1898    fn write(attempt: &str, generation: u64) -> CacheWrite {
1899        CacheWrite {
1900            output: output(&"a".repeat(700)),
1901            source_attempt_id: attempt.into(),
1902            target: 0,
1903            captured_generation: generation,
1904            now: 0,
1905        }
1906    }
1907
1908    /// The accounted size of one `write("x", ..)` entry under `key`.
1909    fn size(key: &RequestKey) -> u64 {
1910        let optimizer = Optimizer::default();
1911        let roomy = memory(1 << 20, 64, 1 << 20, 1 << 20);
1912        optimizer
1913            .insert(&roomy, key.clone(), write("x", 0))
1914            .unwrap();
1915        optimizer.total_bytes(&key.namespace)
1916    }
1917
1918    #[test]
1919    fn scope_and_instance_arity_are_part_of_the_complete_key() {
1920        let a = key(b"tenant-a");
1921        let b = key(b"tenant-b");
1922        assert_ne!(a.id, b.id);
1923        assert_ne!(a.scope_id, b.scope_id);
1924        assert_ne!(a.material, b.material);
1925        // ["ab"] and ["a", "b"] are different instances.
1926        let (plan_id, policy) = policy('1');
1927        let with = |instance: &[String]| {
1928            request_key(KeyParts {
1929                plan_id: &plan_id,
1930                profile: "none",
1931                policy: &policy,
1932                principal: b"t",
1933                backend_artifact: &ArtifactId::parse(&digest('2')).unwrap(),
1934                backend_descriptor: &DescriptorId::parse(&digest('3')).unwrap(),
1935                step: "s",
1936                instance,
1937                projection: b"p",
1938            })
1939            .id
1940        };
1941        assert_ne!(with(&["ab".into()]), with(&["a".into(), "b".into()]));
1942    }
1943
1944    #[test]
1945    fn memory_cache_checks_expiry_collision_and_invalidation_generation() {
1946        let optimizer = Optimizer::default();
1947        let memory = policy('1').1.memory_cache.unwrap();
1948        let key = key(b"tenant");
1949        optimizer
1950            .insert(
1951                &memory,
1952                key.clone(),
1953                CacheWrite {
1954                    output: output("a"),
1955                    source_attempt_id: "attempt-1".into(),
1956                    target: 0,
1957                    captured_generation: 0,
1958                    now: 5,
1959                },
1960            )
1961            .unwrap();
1962        assert_eq!(optimizer.lookup(&key, 14).unwrap().age_ms, 9);
1963        assert_eq!(optimizer.lookup(&key, 15), Err(CacheMiss::Expired));
1964        assert_eq!(optimizer.lookup(&key, 15), Err(CacheMiss::Absent));
1965
1966        optimizer
1967            .insert(&memory, key.clone(), write("a-2", 0))
1968            .unwrap();
1969        optimizer.force_collision(&key, b"other material".to_vec());
1970        assert_eq!(optimizer.lookup(&key, 1), Err(CacheMiss::Collision));
1971        assert_eq!(
1972            optimizer.insert(&memory, key.clone(), write("a-3", 0)),
1973            Err(CacheRefusal::Collision)
1974        );
1975
1976        let result = optimizer.invalidate(&InvalidationSelector::Namespace {
1977            namespace: key.namespace.clone(),
1978        });
1979        assert_eq!(result.generation, 1);
1980        assert_eq!(result.removed_entries, 1);
1981        // A result captured before the invalidation is never stored and so
1982        // never served to a later request (I-3).
1983        assert_eq!(
1984            optimizer.insert(&memory, key.clone(), write("late", 0)),
1985            Err(CacheRefusal::StaleGeneration)
1986        );
1987        assert_eq!(optimizer.lookup(&key, 1), Err(CacheMiss::Absent));
1988        optimizer
1989            .insert(&memory, key.clone(), write("fresh", 1))
1990            .unwrap();
1991        assert_eq!(
1992            optimizer.lookup(&key, 1).unwrap().source_attempt_id,
1993            "fresh"
1994        );
1995    }
1996
1997    #[test]
1998    fn an_oversized_entry_is_refused_before_anything_is_evicted() {
1999        let optimizer = Optimizer::default();
2000        let a = key_in('1', b"tenant", "a");
2001        let n = size(&a);
2002        let tight = memory(4 * n, 2, n, 2 * n);
2003        optimizer.insert(&tight, a.clone(), write("a", 0)).unwrap();
2004        let big = CacheWrite {
2005            output: output(&"b".repeat(800)),
2006            ..write("b", 0)
2007        };
2008        assert_eq!(
2009            optimizer.insert(&tight, key_in('1', b"tenant", "b"), big),
2010            Err(CacheRefusal::EntryTooLarge)
2011        );
2012        assert!(optimizer.lookup(&a, 0).is_ok(), "nothing was evicted");
2013    }
2014
2015    #[test]
2016    fn total_and_count_bounds_evict_the_namespace_fifo_head() {
2017        let optimizer = Optimizer::default();
2018        let a = key(b"tenant-a");
2019        let n = size(&a);
2020        let total = memory(2 * n + n / 2, 8, n, 2 * n + n / 2);
2021        let b = key(b"tenant-b");
2022        let c = key(b"tenant-c");
2023        optimizer.insert(&total, a.clone(), write("a", 0)).unwrap();
2024        optimizer.insert(&total, b.clone(), write("b", 0)).unwrap();
2025        let admission = optimizer.insert(&total, c.clone(), write("c", 0)).unwrap();
2026        assert_eq!(admission.evicted_entries, 1);
2027        assert_eq!(optimizer.lookup(&a, 0), Err(CacheMiss::Absent));
2028        assert!(optimizer.lookup(&b, 0).is_ok());
2029        assert!(optimizer.lookup(&c, 0).is_ok());
2030
2031        let optimizer = Optimizer::default();
2032        let count = memory(100 * n, 2, n, 100 * n);
2033        optimizer.insert(&count, a.clone(), write("a", 0)).unwrap();
2034        optimizer.insert(&count, b.clone(), write("b", 0)).unwrap();
2035        let admission = optimizer.insert(&count, c.clone(), write("c", 0)).unwrap();
2036        assert_eq!(admission.evicted_entries, 1);
2037        assert_eq!(optimizer.lookup(&a, 0), Err(CacheMiss::Absent));
2038    }
2039
2040    #[test]
2041    fn a_scope_bound_evicts_only_that_scopes_entries() {
2042        let optimizer = Optimizer::default();
2043        // Tenant B's entry is the global FIFO head.
2044        let b = key_in('1', b"tenant-b", "b");
2045        let n = size(&b);
2046        // Step names and attempt ids differ by a byte or two.
2047        let scoped = memory(100 * n, 64, n + 8, 2 * n + n / 2);
2048        optimizer.insert(&scoped, b.clone(), write("b", 0)).unwrap();
2049        let a1 = key_in('1', b"tenant-a", "a1");
2050        let a2 = key_in('1', b"tenant-a", "a2");
2051        let a3 = key_in('1', b"tenant-a", "a3");
2052        optimizer
2053            .insert(&scoped, a1.clone(), write("a1", 0))
2054            .unwrap();
2055        optimizer
2056            .insert(&scoped, a2.clone(), write("a2", 0))
2057            .unwrap();
2058        let admission = optimizer
2059            .insert(&scoped, a3.clone(), write("a3", 0))
2060            .unwrap();
2061        assert_eq!(admission.evicted_entries, 1);
2062        assert_eq!(optimizer.lookup(&a1, 0), Err(CacheMiss::Absent));
2063        assert!(optimizer.lookup(&a2, 0).is_ok());
2064        assert!(optimizer.lookup(&a3, 0).is_ok());
2065        assert!(
2066            optimizer.lookup(&b, 0).is_ok(),
2067            "tenant A's traffic never evicts tenant B"
2068        );
2069    }
2070
2071    #[test]
2072    fn one_plans_bounds_never_evict_another_plans_entries() {
2073        let optimizer = Optimizer::default();
2074        let roomy = memory(100_000, 64, 2048, 100_000);
2075        let tiny = memory(2048, 1, 2048, 2048);
2076        let other = key_in('7', b"tenant", "x");
2077        optimizer
2078            .insert(&roomy, other.clone(), write("x", 0))
2079            .unwrap();
2080        for step in ["a", "b", "c"] {
2081            optimizer
2082                .insert(&tiny, key_in('1', b"tenant", step), write(step, 0))
2083                .unwrap();
2084        }
2085        assert!(optimizer.lookup(&other, 0).is_ok());
2086        // Invalidating one namespace leaves the other's generation alone.
2087        optimizer.invalidate(&InvalidationSelector::Namespace {
2088            namespace: key_in('1', b"tenant", "a").namespace,
2089        });
2090        assert_eq!(optimizer.generation(&other.namespace), 0);
2091        assert!(optimizer.lookup(&other, 0).is_ok());
2092    }
2093
2094    #[test]
2095    fn scope_invalidation_is_exact() {
2096        let optimizer = Optimizer::default();
2097        let roomy = memory(100_000, 64, 2048, 100_000);
2098        let a = key(b"tenant-a");
2099        let b = key(b"tenant-b");
2100        optimizer.insert(&roomy, a.clone(), write("a", 0)).unwrap();
2101        optimizer.insert(&roomy, b.clone(), write("b", 0)).unwrap();
2102        let result = optimizer.invalidate(&InvalidationSelector::Scope {
2103            namespace: a.namespace.clone(),
2104            scope_id: a.scope_id.clone(),
2105        });
2106        assert_eq!(result.removed_entries, 1);
2107        assert_eq!(optimizer.lookup(&a, 0), Err(CacheMiss::Absent));
2108        // The generation is namespace-wide, which is conservative: the other
2109        // scope's entry is no longer served.
2110        assert_eq!(optimizer.lookup(&b, 0), Err(CacheMiss::Expired));
2111    }
2112
2113    #[test]
2114    fn lazily_deleted_queue_slots_stay_bounded() {
2115        let optimizer = Optimizer::default();
2116        let roomy = memory(100_000, 64, 2048, 100_000);
2117        let key = key(b"tenant");
2118        for n in 0..500 {
2119            optimizer
2120                .insert(&roomy, key.clone(), write(&format!("a{n}"), 0))
2121                .unwrap();
2122        }
2123        let (fifo, scoped) = optimizer.queue_lengths(&key.namespace);
2124        assert!(fifo <= 20 && scoped <= 20, "{fifo} {scoped}");
2125    }
2126
2127    /// Spec 044, 3.5.2: the profile identity is its own key member, so a
2128    /// profile never shares a key with `none` or with another profile.
2129    #[test]
2130    fn the_profile_is_a_key_member() {
2131        let (plan_id, policy) = policy('1');
2132        let with = |profile: &str| {
2133            request_key(KeyParts {
2134                plan_id: &plan_id,
2135                profile,
2136                policy: &policy,
2137                principal: b"tenant-a",
2138                backend_artifact: &ArtifactId::parse(&digest('2')).unwrap(),
2139                backend_descriptor: &DescriptorId::parse(&digest('3')).unwrap(),
2140                step: "classify",
2141                instance: &["one".into()],
2142                projection: b"projection",
2143            })
2144        };
2145        let none = with("none");
2146        let a = with(&digest('a'));
2147        let b = with(&digest('b'));
2148        assert_eq!(none, key_in('1', b"tenant-a", "classify"));
2149        assert_ne!(none.id, a.id);
2150        assert_ne!(a.id, b.id);
2151        // A chain's ids are delimited: one id holding a comma is not two.
2152        assert_eq!(profile_ids(&["a", "b"]), r#"["a","b"]"#);
2153        assert_ne!(profile_ids(&["a,b"]), profile_ids(&["a", "b"]));
2154        assert_ne!(
2155            with(&profile_ids(&["a,b"])).id,
2156            with(&profile_ids(&["a", "b"])).id
2157        );
2158        // The namespace and scope are the plan's and the principal's alone.
2159        assert_eq!(
2160            (&a.namespace, &a.scope_id),
2161            (&none.namespace, &none.scope_id)
2162        );
2163    }
2164}