rustev_contract/
replay.rs

1//! Replay bundles, `rustev.replay/1` (spec 004, 3.2), and the in-memory
2//! capture a runtime hands back (spec 004, 5.3).
3//!
4//! A bundle describes one admitted decision: its scope, lifetimes, the
5//! retained plan with every descriptor and calibration needed to recompile
6//! it, the snapshot, the delivered run record, the capture status, and one
7//! entry per value supplied to the core, in actual supply order. It is
8//! evidence of what the host retained, never an attestation of a producer
9//! or an authorization token.
10
11use std::collections::BTreeSet;
12use std::fmt;
13
14use serde::{Deserialize, Serialize};
15
16use crate::ids::{PlanId, RequestId, SnapshotId};
17use crate::limits::REPLAY_V1;
18use crate::output::BackendOutputDoc;
19use crate::reason::RuntimeReasonDoc;
20use crate::retention::{MAX_BUNDLE_LIFETIME_MS, RetentionItem};
21use crate::schema;
22use crate::scope::Scope;
23use crate::time::{DurationMs, Timestamp};
24
25/// The most descriptors, calibrations or supplies one bundle carries.
26pub const MAX_BUNDLE_ITEMS: usize = 4096;
27/// The most resolved bytes of one case, external items included.
28pub const MAX_RESOLVED_BYTES: usize = 16 * 1024 * 1024;
29
30#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
31#[serde(deny_unknown_fields)]
32pub struct ReplayBundle {
33    pub schema: String,
34    pub decision_id: String,
35    pub scope: Scope,
36    /// Host wall-clock times; copying a bundle resets neither.
37    pub created_at_ms: Timestamp,
38    pub expires_at_ms: Timestamp,
39    /// The domain time the decision was evaluated at.
40    pub evaluation_time_ms: Timestamp,
41    pub plan_id: PlanId,
42    pub plan: RetentionItem,
43    /// Every descriptor the plan's bindings name.
44    pub descriptors: Vec<RetentionItem>,
45    /// Every calibration artifact the plan binds.
46    pub calibrations: Vec<RetentionItem>,
47    pub snapshot_id: SnapshotId,
48    pub snapshot: RetentionItem,
49    /// The delivered `rustev.run/1` record; the expected judgment is in it.
50    pub run: RetentionItem,
51    pub capture: CaptureStatus,
52    /// In actual supply order.
53    pub supplies: Vec<SupplyEntry>,
54    /// Replay 2: the source descriptors bound by the retained plan.
55    #[serde(
56        default,
57        skip_serializing_if = "Option::is_none",
58        deserialize_with = "crate::acquisition_record::list"
59    )]
60    #[schemars(with = "Vec<RetentionItem>", length(min = 1, max = 4096))]
61    pub sources: Option<Vec<RetentionItem>>,
62    /// Replay 2: captured pages share the semantic supply sequence.
63    #[serde(
64        default,
65        skip_serializing_if = "Option::is_none",
66        deserialize_with = "captured_pages"
67    )]
68    #[schemars(with = "Vec<AcquisitionSupply>", length(max = 4096))]
69    pub acquisition_pages: Option<Vec<AcquisitionSupply>>,
70}
71
72crate::document::versioned_document!(
73    ReplayBundle,
74    [schema::REPLAY, schema::REPLAY_V2],
75    "rustev.replay/1 or rustev.replay/2",
76    REPLAY_V1,
77    check = ReplayBundle::parsed_members
78);
79
80#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
81#[serde(deny_unknown_fields)]
82pub struct AcquisitionSupply {
83    pub sequence: u64,
84    pub step: String,
85    pub tier: u32,
86    pub page: u32,
87    pub request: crate::ids::AcquisitionRequestId,
88    pub attempt_id: Option<String>,
89    /// An acquisition page or runtime reason, subject to retention.
90    pub value: RetentionItem,
91}
92
93/// Whether every supplied value was captured. Only `complete` reproduces.
94#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
95#[serde(rename_all = "snake_case")]
96pub enum CaptureStatus {
97    Complete,
98    Disabled,
99    /// The byte or request cap was hit; captured payloads were discarded.
100    LimitExceeded,
101    Cancelled,
102    Abandoned,
103}
104
105/// One value successfully supplied to the core.
106#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
107#[serde(deny_unknown_fields)]
108pub struct SupplyEntry {
109    pub step: String,
110    pub instance: Vec<String>,
111    /// The identity of the request for the producing target (spec 004, 3.2);
112    /// the document is recomputed from the retained plan and snapshot.
113    pub request: RequestId,
114    /// Actual `supply` order, contiguous from 0.
115    pub sequence: u64,
116    /// The attempt that produced the value (spec 004, 3.2); none when
117    /// nothing was dispatched or a retry or fallback followed the last one.
118    pub attempt_id: Option<String>,
119    /// The target the value came from (0 is the primary, then declared
120    /// fallbacks): the producing target of an output; for a failure, the `to`
121    /// of the last fallback transition, else 0.
122    pub target: u32,
123    /// A `rustev.backend-output/1` or `rustev.runtime-reason/1` document.
124    pub value: RetentionItem,
125    /// Replay 2: runtime facts needed to recompute an actionability cause.
126    #[serde(
127        default,
128        skip_serializing_if = "Option::is_none",
129        deserialize_with = "failure_context"
130    )]
131    pub failure_context: Option<CapturedFailureContext>,
132}
133
134#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
135#[serde(deny_unknown_fields)]
136pub struct CapturedFailureContext {
137    pub origin: crate::actionability::Origin,
138    pub determinism: crate::definition::Determinism,
139}
140
141/// Where in a bundle a problem is.
142#[derive(
143    Debug,
144    Clone,
145    Copy,
146    PartialEq,
147    Eq,
148    PartialOrd,
149    Ord,
150    Hash,
151    Serialize,
152    Deserialize,
153    schemars::JsonSchema,
154)]
155#[serde(rename_all = "snake_case")]
156pub enum ItemLocation {
157    Bundle,
158    Plan,
159    Descriptor(usize),
160    Calibration(usize),
161    Snapshot,
162    Run,
163    Supply(usize),
164    Source(usize),
165    AcquisitionPage(usize),
166}
167
168impl fmt::Display for ItemLocation {
169    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
170        match self {
171            ItemLocation::Source(i) => write!(f, "sources[{i}]"),
172            ItemLocation::AcquisitionPage(i) => write!(f, "acquisition_pages[{i}]"),
173            ItemLocation::Bundle => f.write_str("bundle"),
174            ItemLocation::Plan => f.write_str("plan"),
175            ItemLocation::Descriptor(i) => write!(f, "descriptors[{i}]"),
176            ItemLocation::Calibration(i) => write!(f, "calibrations[{i}]"),
177            ItemLocation::Snapshot => f.write_str("snapshot"),
178            ItemLocation::Run => f.write_str("run"),
179            ItemLocation::Supply(i) => write!(f, "supplies[{i}]"),
180        }
181    }
182}
183
184#[derive(Debug, Clone, Copy, PartialEq, Eq)]
185pub enum LifetimeProblem {
186    /// An expiry at or before creation.
187    NotAfterCreation,
188    /// A lifetime beyond the hard or host cap.
189    ExceedsCap { cap_ms: u64 },
190    /// An item that outlives its bundle.
191    OutlivesBundle,
192}
193
194/// Why a bundle is structurally invalid, before anything is resolved.
195#[derive(Debug, Clone, PartialEq, Eq)]
196pub enum BundleError {
197    Schema(String),
198    /// A decision id is 1 to 256 bytes.
199    DecisionId,
200    Lifetime {
201        location: ItemLocation,
202        problem: LifetimeProblem,
203    },
204    TooMany {
205        what: &'static str,
206        count: usize,
207    },
208    /// Supply sequence numbers are not contiguous from 0 in entry order.
209    Sequence {
210        index: usize,
211    },
212    /// Two entries share a step and instance.
213    DuplicateSupply {
214        index: usize,
215    },
216    /// An item declares a document schema its role does not allow.
217    Role {
218        location: ItemLocation,
219    },
220    /// The plan or snapshot item declares an identity other than the
221    /// bundle's `plan_id` or `snapshot_id`.
222    Identity {
223        location: ItemLocation,
224    },
225    /// A disabled or limit-exceeded capture carries supplies.
226    SuppliesWithoutCapture,
227}
228
229impl fmt::Display for BundleError {
230    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
231        match self {
232            BundleError::Schema(s) => write!(f, "bundle schema {s:?}"),
233            BundleError::DecisionId => f.write_str("a decision id is 1 to 256 bytes"),
234            BundleError::Lifetime { location, problem } => {
235                write!(f, "lifetime of {location}: {problem:?}")
236            }
237            BundleError::TooMany { what, count } => {
238                write!(f, "{count} {what}, more than {MAX_BUNDLE_ITEMS}")
239            }
240            BundleError::Sequence { index } => {
241                write!(f, "supplies[{index}] breaks the contiguous supply sequence")
242            }
243            BundleError::DuplicateSupply { index } => {
244                write!(f, "supplies[{index}] repeats a step and instance")
245            }
246            BundleError::Role { location } => {
247                write!(f, "{location} declares a schema its role does not allow")
248            }
249            BundleError::Identity { location } => {
250                write!(
251                    f,
252                    "{location} declares an identity the bundle does not name"
253                )
254            }
255            BundleError::SuppliesWithoutCapture => {
256                f.write_str("a disabled or limit-exceeded capture has no supplies")
257            }
258        }
259    }
260}
261
262impl std::error::Error for BundleError {}
263
264impl ReplayBundle {
265    fn parsed_members(&self) -> Result<(), String> {
266        // Preserve replay 1 parsing; its existing structural check remains explicit.
267        if self.schema == schema::REPLAY {
268            if self.sources.is_some()
269                || self.acquisition_pages.is_some()
270                || self.supplies.iter().any(|s| s.failure_context.is_some())
271            {
272                return Err("acquisition members require rustev.replay/2".into());
273            }
274            return Ok(());
275        }
276        self.check(None).map_err(|e| e.to_string())
277    }
278
279    /// Every retention item with its location, in document order.
280    pub fn items(&self) -> Vec<(ItemLocation, &RetentionItem)> {
281        let mut v = vec![(ItemLocation::Plan, &self.plan)];
282        v.extend(
283            self.descriptors
284                .iter()
285                .enumerate()
286                .map(|(i, d)| (ItemLocation::Descriptor(i), d)),
287        );
288        v.extend(
289            self.calibrations
290                .iter()
291                .enumerate()
292                .map(|(i, c)| (ItemLocation::Calibration(i), c)),
293        );
294        v.push((ItemLocation::Snapshot, &self.snapshot));
295        v.push((ItemLocation::Run, &self.run));
296        v.extend(
297            self.supplies
298                .iter()
299                .enumerate()
300                .map(|(i, s)| (ItemLocation::Supply(i), &s.value)),
301        );
302        if let Some(sources) = &self.sources {
303            v.extend(
304                sources
305                    .iter()
306                    .enumerate()
307                    .map(|(i, s)| (ItemLocation::Source(i), s)),
308            );
309        }
310        if let Some(pages) = &self.acquisition_pages {
311            v.extend(
312                pages
313                    .iter()
314                    .enumerate()
315                    .map(|(i, p)| (ItemLocation::AcquisitionPage(i), &p.value)),
316            );
317        }
318        v
319    }
320
321    /// Structural checks that need no resolved bytes: schema, decision id,
322    /// lifetimes under the hard cap and an optional tighter host cap, item
323    /// counts (at most 4096 each; the plan's own request bound is checked at
324    /// replay, once the plan is resolved), item roles, declared plan and
325    /// snapshot identities, and the supply sequence. Applies to a typed
326    /// bundle as well as a parsed one.
327    pub fn check(&self, host_cap_ms: Option<u64>) -> Result<(), BundleError> {
328        if self.schema != schema::REPLAY && self.schema != schema::REPLAY_V2 {
329            return Err(BundleError::Schema(self.schema.chars().take(80).collect()));
330        }
331        let v2 = self.schema == schema::REPLAY_V2;
332        // Replay 2 permits absent sources and pages, including source-free plans.
333        if !v2 && (self.sources.is_some() || self.acquisition_pages.is_some()) {
334            return Err(BundleError::Role {
335                location: ItemLocation::Bundle,
336            });
337        }
338        if self.decision_id.is_empty() || self.decision_id.len() > 256 {
339            return Err(BundleError::DecisionId);
340        }
341        let cap_ms = host_cap_ms.map_or(MAX_BUNDLE_LIFETIME_MS, |c| c.min(MAX_BUNDLE_LIFETIME_MS));
342        let life = |location, problem| Err(BundleError::Lifetime { location, problem });
343        if self.expires_at_ms <= self.created_at_ms {
344            return life(ItemLocation::Bundle, LifetimeProblem::NotAfterCreation);
345        }
346        // When creation plus the cap is past the last timestamp, every
347        // representable expiry is within the cap.
348        let beyond = match self.created_at_ms.checked_add(DurationMs(cap_ms)) {
349            Ok(latest) => self.expires_at_ms > latest,
350            Err(_) => false,
351        };
352        if beyond {
353            return life(ItemLocation::Bundle, LifetimeProblem::ExceedsCap { cap_ms });
354        }
355        for (what, count) in [
356            ("descriptors", self.descriptors.len()),
357            ("calibrations", self.calibrations.len()),
358            ("supplies", self.supplies.len()),
359            ("sources", self.sources.as_ref().map_or(0, Vec::len)),
360            (
361                "acquisition_pages",
362                self.acquisition_pages.as_ref().map_or(0, Vec::len),
363            ),
364        ] {
365            if count > MAX_BUNDLE_ITEMS {
366                return Err(BundleError::TooMany { what, count });
367            }
368        }
369        for (location, item) in self.items() {
370            let exp = item.retention.expires_at();
371            if exp <= self.created_at_ms {
372                return life(location, LifetimeProblem::NotAfterCreation);
373            }
374            if exp > self.expires_at_ms {
375                return life(location, LifetimeProblem::OutlivesBundle);
376            }
377            let allowed: &[&str] = match location {
378                ItemLocation::Plan if v2 => &[schema::PLAN_V3],
379                ItemLocation::Snapshot if v2 => &[schema::SNAPSHOT, schema::SNAPSHOT_V2],
380                ItemLocation::Run if v2 => &[schema::RUN_V3],
381                ItemLocation::Source(_) => &[schema::SOURCE],
382                ItemLocation::AcquisitionPage(_) => {
383                    &[schema::ACQUISITION_PAGE, schema::RUNTIME_REASON]
384                }
385                ItemLocation::Plan => &[schema::PLAN, schema::PLAN_V3],
386                ItemLocation::Descriptor(_) => &[schema::BACKEND],
387                ItemLocation::Calibration(_) => &[schema::CALIBRATION],
388                ItemLocation::Snapshot => &[schema::SNAPSHOT],
389                ItemLocation::Run => &[schema::RUN, schema::RUN_V3],
390                ItemLocation::Supply(_) => &[schema::BACKEND_OUTPUT, schema::RUNTIME_REASON],
391                ItemLocation::Bundle => &[],
392            };
393            if !allowed.contains(&item.document_schema.as_str()) {
394                return Err(BundleError::Role { location });
395            }
396        }
397        for (location, item, id) in [
398            (ItemLocation::Plan, &self.plan, self.plan_id.as_str()),
399            (
400                ItemLocation::Snapshot,
401                &self.snapshot,
402                self.snapshot_id.as_str(),
403            ),
404        ] {
405            if item.identity.as_ref().is_some_and(|d| d.as_str() != id) {
406                return Err(BundleError::Identity { location });
407            }
408        }
409        if matches!(
410            self.capture,
411            CaptureStatus::Disabled | CaptureStatus::LimitExceeded
412        ) && (!self.supplies.is_empty()
413            || self
414                .acquisition_pages
415                .as_ref()
416                .is_some_and(|p| !p.is_empty()))
417        {
418            return Err(BundleError::SuppliesWithoutCapture);
419        }
420        if v2
421            && (self
422                .supplies
423                .windows(2)
424                .any(|w| w[0].sequence >= w[1].sequence)
425                || self
426                    .acquisition_pages
427                    .as_ref()
428                    .is_some_and(|pages| pages.windows(2).any(|w| w[0].sequence >= w[1].sequence)))
429        {
430            return Err(BundleError::Sequence { index: 0 });
431        }
432        // One sequence space shared by acquisition pages and supplies (spec
433        // 036 3.15.3): pages enter `sequence` first, so a page colliding with
434        // another page or with a supply is refused here or in the supply loop.
435        let mut sequence = BTreeSet::new();
436        let mut page_keys = BTreeSet::new();
437        if let Some(pages) = &self.acquisition_pages {
438            for (i, p) in pages.iter().enumerate() {
439                if !sequence.insert(p.sequence) {
440                    return Err(BundleError::Sequence { index: i });
441                }
442                if !page_keys.insert((&p.step, p.tier, p.page)) {
443                    return Err(BundleError::DuplicateSupply { index: i });
444                }
445            }
446        }
447        let mut keys = BTreeSet::new();
448        for (i, s) in self.supplies.iter().enumerate() {
449            if s.failure_context.is_some()
450                && (!v2 || s.value.document_schema != schema::RUNTIME_REASON)
451            {
452                return Err(BundleError::Role {
453                    location: ItemLocation::Supply(i),
454                });
455            }
456            // The v2 windows(2) checks above already enforce ascending order
457            // in both supplies and acquisition_pages; only v1 requires i here.
458            if (!v2 && s.sequence != i as u64) || !sequence.insert(s.sequence) {
459                return Err(BundleError::Sequence { index: i });
460            }
461            if !keys.insert((&s.step, &s.instance)) {
462                return Err(BundleError::DuplicateSupply { index: i });
463            }
464        }
465        if !v2 && sequence.iter().copied().ne(0..sequence.len() as u64) {
466            return Err(BundleError::Sequence { index: 0 });
467        }
468        Ok(())
469    }
470}
471
472/// What a runtime captured for one admitted decision (spec 004, 5.3): a
473/// bounded in-memory handoff, not persistent retention and not a callback.
474#[derive(Debug, Clone, PartialEq)]
475pub struct Capture {
476    pub decision_id: String,
477    pub scope: Scope,
478    /// `complete`, `limit_exceeded` or `cancelled`. Entries are empty under
479    /// `limit_exceeded`: a partial capture is never reported as complete.
480    pub status: CaptureStatus,
481    /// In actual supply order.
482    pub supplies: Vec<CapturedSupply>,
483    /// The record-canonical lengths of every captured value document plus
484    /// each `RequestId` string (spec 004, 5.7), counted before any discard.
485    pub bytes: u64,
486    /// Acquisition supplies share the sequence with semantic supplies.
487    pub acquisition_pages: Vec<CapturedAcquisition>,
488}
489
490/// One supplied value, as [`SupplyEntry`] describes it before retention.
491#[derive(Debug, Clone, PartialEq)]
492pub struct CapturedSupply {
493    pub sequence: u64,
494    pub step: String,
495    pub instance: Vec<String>,
496    pub request: RequestId,
497    pub attempt_id: Option<String>,
498    pub target: u32,
499    pub value: CapturedValue,
500    pub failure_context: Option<CapturedFailureContext>,
501}
502
503#[derive(Debug, Clone, PartialEq)]
504pub enum CapturedValue {
505    /// Names the actual producing artifact, fallback included.
506    Output(BackendOutputDoc),
507    Reason(RuntimeReasonDoc),
508}
509
510/// One captured acquisition before the host applies retention.
511#[derive(Debug, Clone, PartialEq)]
512pub struct CapturedAcquisition {
513    pub sequence: u64,
514    pub step: String,
515    pub tier: u32,
516    pub page: u32,
517    pub request: crate::ids::AcquisitionRequestId,
518    pub attempt_id: Option<String>,
519    pub value: CapturedPage,
520}
521
522#[derive(Debug, Clone, PartialEq)]
523pub enum CapturedPage {
524    Page(crate::acquisition::AcquisitionPage),
525    Reason(RuntimeReasonDoc),
526}
527
528// Empty captures are meaningful for a skipped acquisition or a removed last page.
529fn captured_pages<'de, D: serde::Deserializer<'de>>(
530    d: D,
531) -> Result<Option<Vec<AcquisitionSupply>>, D::Error> {
532    let pages = Vec::<AcquisitionSupply>::deserialize(d)?;
533    if pages.len() > MAX_BUNDLE_ITEMS {
534        return Err(serde::de::Error::custom("too many captured pages"));
535    }
536    Ok(Some(pages))
537}
538
539fn failure_context<'de, D: serde::Deserializer<'de>>(
540    d: D,
541) -> Result<Option<CapturedFailureContext>, D::Error> {
542    CapturedFailureContext::deserialize(d).map(Some)
543}
rustev_contract/replay.rs

rustev_contract/
replay.rs
