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        if !v2 && (self.sources.is_some() || self.acquisition_pages.is_some()) {
333            return Err(BundleError::Role {
334                location: ItemLocation::Bundle,
335            });
336        }
337        if self.decision_id.is_empty() || self.decision_id.len() > 256 {
338            return Err(BundleError::DecisionId);
339        }
340        let cap_ms = host_cap_ms.map_or(MAX_BUNDLE_LIFETIME_MS, |c| c.min(MAX_BUNDLE_LIFETIME_MS));
341        let life = |location, problem| Err(BundleError::Lifetime { location, problem });
342        if self.expires_at_ms <= self.created_at_ms {
343            return life(ItemLocation::Bundle, LifetimeProblem::NotAfterCreation);
344        }
345        // When creation plus the cap is past the last timestamp, every
346        // representable expiry is within the cap.
347        let beyond = match self.created_at_ms.checked_add(DurationMs(cap_ms)) {
348            Ok(latest) => self.expires_at_ms > latest,
349            Err(_) => false,
350        };
351        if beyond {
352            return life(ItemLocation::Bundle, LifetimeProblem::ExceedsCap { cap_ms });
353        }
354        for (what, count) in [
355            ("descriptors", self.descriptors.len()),
356            ("calibrations", self.calibrations.len()),
357            ("supplies", self.supplies.len()),
358            ("sources", self.sources.as_ref().map_or(0, Vec::len)),
359            (
360                "acquisition_pages",
361                self.acquisition_pages.as_ref().map_or(0, Vec::len),
362            ),
363        ] {
364            if count > MAX_BUNDLE_ITEMS {
365                return Err(BundleError::TooMany { what, count });
366            }
367        }
368        for (location, item) in self.items() {
369            let exp = item.retention.expires_at();
370            if exp <= self.created_at_ms {
371                return life(location, LifetimeProblem::NotAfterCreation);
372            }
373            if exp > self.expires_at_ms {
374                return life(location, LifetimeProblem::OutlivesBundle);
375            }
376            let allowed: &[&str] = match location {
377                ItemLocation::Plan if v2 => &[schema::PLAN_V3],
378                ItemLocation::Snapshot if v2 => &[schema::SNAPSHOT, schema::SNAPSHOT_V2],
379                ItemLocation::Run if v2 => &[schema::RUN_V3],
380                ItemLocation::Source(_) => &[schema::SOURCE],
381                ItemLocation::AcquisitionPage(_) => {
382                    &[schema::ACQUISITION_PAGE, schema::RUNTIME_REASON]
383                }
384                ItemLocation::Plan => &[schema::PLAN, schema::PLAN_V3],
385                ItemLocation::Descriptor(_) => &[schema::BACKEND],
386                ItemLocation::Calibration(_) => &[schema::CALIBRATION],
387                ItemLocation::Snapshot => &[schema::SNAPSHOT],
388                ItemLocation::Run => &[schema::RUN, schema::RUN_V3],
389                ItemLocation::Supply(_) => &[schema::BACKEND_OUTPUT, schema::RUNTIME_REASON],
390                ItemLocation::Bundle => &[],
391            };
392            if !allowed.contains(&item.document_schema.as_str()) {
393                return Err(BundleError::Role { location });
394            }
395        }
396        for (location, item, id) in [
397            (ItemLocation::Plan, &self.plan, self.plan_id.as_str()),
398            (
399                ItemLocation::Snapshot,
400                &self.snapshot,
401                self.snapshot_id.as_str(),
402            ),
403        ] {
404            if item.identity.as_ref().is_some_and(|d| d.as_str() != id) {
405                return Err(BundleError::Identity { location });
406            }
407        }
408        if matches!(
409            self.capture,
410            CaptureStatus::Disabled | CaptureStatus::LimitExceeded
411        ) && (!self.supplies.is_empty()
412            || self
413                .acquisition_pages
414                .as_ref()
415                .is_some_and(|p| !p.is_empty()))
416        {
417            return Err(BundleError::SuppliesWithoutCapture);
418        }
419        if v2
420            && (self
421                .supplies
422                .windows(2)
423                .any(|w| w[0].sequence >= w[1].sequence)
424                || self
425                    .acquisition_pages
426                    .as_ref()
427                    .is_some_and(|pages| pages.windows(2).any(|w| w[0].sequence >= w[1].sequence)))
428        {
429            return Err(BundleError::Sequence { index: 0 });
430        }
431        // One sequence space shared by acquisition pages and supplies (spec
432        // 036 3.15.3): pages enter `sequence` first, so a page colliding with
433        // another page or with a supply is refused here or in the supply loop.
434        let mut sequence = BTreeSet::new();
435        let mut page_keys = BTreeSet::new();
436        if let Some(pages) = &self.acquisition_pages {
437            for (i, p) in pages.iter().enumerate() {
438                if !sequence.insert(p.sequence) {
439                    return Err(BundleError::Sequence { index: i });
440                }
441                if !page_keys.insert((&p.step, p.tier, p.page)) {
442                    return Err(BundleError::DuplicateSupply { index: i });
443                }
444            }
445        }
446        let mut keys = BTreeSet::new();
447        for (i, s) in self.supplies.iter().enumerate() {
448            if s.failure_context.is_some()
449                && (!v2 || s.value.document_schema != schema::RUNTIME_REASON)
450            {
451                return Err(BundleError::Role {
452                    location: ItemLocation::Supply(i),
453                });
454            }
455            // The v2 windows(2) checks above already enforce ascending order
456            // in both supplies and acquisition_pages; only v1 requires i here.
457            if (!v2 && s.sequence != i as u64) || !sequence.insert(s.sequence) {
458                return Err(BundleError::Sequence { index: i });
459            }
460            if !keys.insert((&s.step, &s.instance)) {
461                return Err(BundleError::DuplicateSupply { index: i });
462            }
463        }
464        if !v2 && sequence.iter().copied().ne(0..sequence.len() as u64) {
465            return Err(BundleError::Sequence { index: 0 });
466        }
467        Ok(())
468    }
469}
470
471/// What a runtime captured for one admitted decision (spec 004, 5.3): a
472/// bounded in-memory handoff, not persistent retention and not a callback.
473#[derive(Debug, Clone, PartialEq)]
474pub struct Capture {
475    pub decision_id: String,
476    pub scope: Scope,
477    /// `complete`, `limit_exceeded` or `cancelled`. Entries are empty under
478    /// `limit_exceeded`: a partial capture is never reported as complete.
479    pub status: CaptureStatus,
480    /// In actual supply order.
481    pub supplies: Vec<CapturedSupply>,
482    /// The record-canonical lengths of every captured value document plus
483    /// each `RequestId` string (spec 004, 5.7), counted before any discard.
484    pub bytes: u64,
485    /// Acquisition supplies share the sequence with semantic supplies.
486    pub acquisition_pages: Vec<CapturedAcquisition>,
487}
488
489/// One supplied value, as [`SupplyEntry`] describes it before retention.
490#[derive(Debug, Clone, PartialEq)]
491pub struct CapturedSupply {
492    pub sequence: u64,
493    pub step: String,
494    pub instance: Vec<String>,
495    pub request: RequestId,
496    pub attempt_id: Option<String>,
497    pub target: u32,
498    pub value: CapturedValue,
499    pub failure_context: Option<CapturedFailureContext>,
500}
501
502#[derive(Debug, Clone, PartialEq)]
503pub enum CapturedValue {
504    /// Names the actual producing artifact, fallback included.
505    Output(BackendOutputDoc),
506    Reason(RuntimeReasonDoc),
507}
508
509/// One captured acquisition before the host applies retention.
510#[derive(Debug, Clone, PartialEq)]
511pub struct CapturedAcquisition {
512    pub sequence: u64,
513    pub step: String,
514    pub tier: u32,
515    pub page: u32,
516    pub request: crate::ids::AcquisitionRequestId,
517    pub attempt_id: Option<String>,
518    pub value: CapturedPage,
519}
520
521#[derive(Debug, Clone, PartialEq)]
522pub enum CapturedPage {
523    Page(crate::acquisition::AcquisitionPage),
524    Reason(RuntimeReasonDoc),
525}
526
527// Empty captures are meaningful for a skipped acquisition or a removed last page.
528fn captured_pages<'de, D: serde::Deserializer<'de>>(
529    d: D,
530) -> Result<Option<Vec<AcquisitionSupply>>, D::Error> {
531    let pages = Vec::<AcquisitionSupply>::deserialize(d)?;
532    if pages.len() > MAX_BUNDLE_ITEMS {
533        return Err(serde::de::Error::custom("too many captured pages"));
534    }
535    Ok(Some(pages))
536}
537
538fn failure_context<'de, D: serde::Deserializer<'de>>(
539    d: D,
540) -> Result<Option<CapturedFailureContext>, D::Error> {
541    CapturedFailureContext::deserialize(d).map(Some)
542}
rustev_contract/replay.rs

rustev_contract/
replay.rs
