rustev_eval/
assemble.rs

1//! Assembling a replay bundle from what the host retained (spec 004, 5.5).
2//!
3//! Assembly takes the caller-retained plan, its descriptors and calibrations,
4//! the snapshot, the delivered run record and the runtime's capture,
5//! validates every binding between them, and applies the host's explicit
6//! retention choice. Content defaults to digest-only and metadata to seven
7//! days; keeping bytes, embedded or external, is an explicit choice, and no
8//! item outlives the bundle. A capture is not a promise that a bundle fits:
9//! the escaped size is checked and refused with a typed error.
10
11use std::fmt;
12
13use rustev_contract::calibration::CalibrationArtifact;
14use rustev_contract::canonical::tagged_digest;
15use rustev_contract::descriptor::BackendDescriptor;
16use rustev_contract::ids::ContentDigest;
17use rustev_contract::limits::REPLAY_V1;
18use rustev_contract::plan::Plan;
19use rustev_contract::replay::{
20    AcquisitionSupply, BundleError, Capture, CaptureStatus, CapturedPage, CapturedValue,
21    ItemLocation, MAX_BUNDLE_ITEMS, MAX_RESOLVED_BYTES, ReplayBundle, SupplyEntry,
22};
23use rustev_contract::retention::{
24    DEFAULT_METADATA_LIFETIME_MS, MAX_BUNDLE_LIFETIME_MS, Retention, RetentionItem,
25};
26use rustev_contract::run::{CoreEvidence, RunRecord, Termination};
27use rustev_contract::scope::Scope;
28use rustev_contract::snapshot::Snapshot;
29use rustev_contract::time::{DurationMs, Timestamp};
30use rustev_contract::{Document, Identified, schema};
31use rustev_core::{Compiled, LoadError};
32
33/// How the host keeps document bytes.
34pub enum Content<'a> {
35    /// Only digests: nothing is replayable. The default.
36    DigestOnly,
37    /// Embedded in the bundle.
38    Embedded,
39    /// Stored by the host, which returns an opaque reference for the bytes.
40    External(&'a dyn Fn(ItemLocation, &[u8]) -> String),
41}
42
43/// The host's explicit retention choice for one bundle.
44pub struct RetentionChoice<'a> {
45    /// From creation; every item expires with the bundle.
46    pub lifetime_ms: u64,
47    pub content: Content<'a>,
48    /// A cap tighter than the hard 30 days.
49    pub host_cap_ms: Option<u64>,
50}
51
52impl Default for RetentionChoice<'_> {
53    fn default() -> Self {
54        RetentionChoice {
55            lifetime_ms: DEFAULT_METADATA_LIFETIME_MS,
56            content: Content::DigestOnly,
57            host_cap_ms: None,
58        }
59    }
60}
61
62/// What the host retained for one decision.
63pub struct AssemblyInput<'a> {
64    /// `None` when capture was off: the bundle states `disabled`.
65    pub capture: Option<&'a Capture>,
66    /// The host's isolation scope; a capture's must be the same.
67    pub scope: &'a Scope,
68    pub plan: &'a Plan,
69    pub descriptors: &'a [BackendDescriptor],
70    pub calibrations: &'a [CalibrationArtifact],
71    pub snapshot: &'a Snapshot,
72    pub run: &'a RunRecord,
73    /// The domain time the decision was evaluated at.
74    pub evaluation_time: Timestamp,
75    /// Host wall-clock creation time.
76    pub created_at_ms: Timestamp,
77}
78
79/// Why no bundle was assembled.
80#[derive(Debug, Clone, PartialEq, Eq)]
81pub enum AssemblyError {
82    /// The lifetime is zero, beyond the cap, or past the last timestamp.
83    Lifetime {
84        lifetime_ms: u64,
85    },
86    /// Capture, run record, plan and snapshot do not describe one decision.
87    Binding {
88        what: &'static str,
89    },
90    /// The plan does not load with the given dependencies.
91    Plan(String),
92    /// A document has no record canonical form.
93    NotCanonical {
94        location: ItemLocation,
95    },
96    /// The escaped bundle or its embedded bytes exceed the replay bounds.
97    TooLarge {
98        bytes: usize,
99        limit: usize,
100    },
101    Bundle(BundleError),
102}
103
104impl fmt::Display for AssemblyError {
105    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
106        write!(f, "{self:?}")
107    }
108}
109
110impl std::error::Error for AssemblyError {}
111
112fn bytes_of<T: Document + PartialEq>(
113    doc: &T,
114    location: ItemLocation,
115) -> Result<Vec<u8>, AssemblyError> {
116    doc.record_canonical()
117        .map_err(|_| AssemblyError::NotCanonical { location })
118}
119
120/// Assemble and check a bundle.
121pub fn assemble(
122    input: &AssemblyInput<'_>,
123    choice: &RetentionChoice<'_>,
124) -> Result<ReplayBundle, AssemblyError> {
125    assemble_with_sources(input, choice, &[])
126}
127
128/// Assemble a plan 3 bundle with its retained source descriptors.
129pub fn assemble_with_sources(
130    input: &AssemblyInput<'_>,
131    choice: &RetentionChoice<'_>,
132    sources: &[rustev_contract::source::SourceDescriptor],
133) -> Result<ReplayBundle, AssemblyError> {
134    let cap = choice
135        .host_cap_ms
136        .map_or(MAX_BUNDLE_LIFETIME_MS, |c| c.min(MAX_BUNDLE_LIFETIME_MS));
137    let lifetime = choice.lifetime_ms;
138    if lifetime == 0 || lifetime > cap {
139        return Err(AssemblyError::Lifetime {
140            lifetime_ms: lifetime,
141        });
142    }
143    let expires = input
144        .created_at_ms
145        .checked_add(DurationMs(lifetime))
146        .map_err(|_| AssemblyError::Lifetime {
147            lifetime_ms: lifetime,
148        })?;
149    let binding = |what| Err(AssemblyError::Binding { what });
150    // A `rustev.run/2` record enters the bundle as its deterministic
151    // version 1 projection (spec 020, 3.2). Plan 3 keeps its version 3 record.
152    let projected;
153    let run = if input.run.schema == schema::RUN_V2 {
154        projected = input.run.project_v1().map_err(|_| AssemblyError::Binding {
155            what: "the run record's version 1 projection",
156        })?;
157        &projected
158    } else {
159        input.run
160    };
161    let plan_id = input.plan.id().map_err(|_| AssemblyError::NotCanonical {
162        location: ItemLocation::Plan,
163    })?;
164    let snapshot_id = input
165        .snapshot
166        .id()
167        .map_err(|_| AssemblyError::NotCanonical {
168            location: ItemLocation::Snapshot,
169        })?;
170    if run.plan_id != plan_id {
171        return binding("run record plan id");
172    }
173    if let CoreEvidence::Recorded(e) = &run.core {
174        if e.plan_id != plan_id || e.judgment.plan_id != plan_id {
175            return binding("evidence plan id");
176        }
177        if e.snapshot_id != snapshot_id {
178            return binding("snapshot id");
179        }
180        if e.evaluation_time_ms != input.evaluation_time {
181            return binding("evaluation time");
182        }
183        if e.decision_id != run.decision_id {
184            return binding("evidence decision id");
185        }
186    }
187    if let Some(c) = input.capture {
188        if c.decision_id != run.decision_id {
189            return binding("capture decision id");
190        }
191        if &c.scope != input.scope {
192            return binding("capture scope");
193        }
194        let max = input
195            .plan
196            .definition
197            .limits
198            .max_semantic_requests
199            .min(MAX_BUNDLE_ITEMS as u64);
200        if c.supplies.len() as u64 > max {
201            return binding("capture entries beyond the plan's request bound");
202        }
203        if c.status == CaptureStatus::Complete && run.termination != Termination::Judged {
204            return binding("a complete capture of a run without a judgment");
205        }
206    }
207    Compiled::load_checked_with_sources(input.plan, input.descriptors, input.calibrations, sources)
208        .map_err(|e: LoadError| AssemblyError::Plan(e.to_string()))?;
209
210    // Every byte replay would resolve, embedded or external, counts.
211    let mut resolved = 0usize;
212    let mut not_utf8 = None;
213    let mut item = |location: ItemLocation,
214                    document_schema: &str,
215                    bytes: Vec<u8>,
216                    identified: bool|
217     -> RetentionItem {
218        let digest = ContentDigest::parse(&tagged_digest(document_schema, &bytes))
219            .expect("a tagged digest is an identity");
220        let retention = match &choice.content {
221            Content::DigestOnly => Retention::DigestOnly {
222                digest: digest.clone(),
223                expires_at_ms: expires,
224            },
225            Content::Embedded => {
226                resolved += bytes.len();
227                Retention::Retained {
228                    bytes: String::from_utf8(bytes).unwrap_or_else(|_| {
229                        not_utf8 = Some(location);
230                        String::new()
231                    }),
232                    digest: digest.clone(),
233                    expires_at_ms: expires,
234                }
235            }
236            Content::External(store) => {
237                resolved += bytes.len();
238                Retention::External {
239                    reference: store(location, &bytes),
240                    digest: digest.clone(),
241                    expires_at_ms: expires,
242                }
243            }
244        };
245        RetentionItem {
246            document_schema: document_schema.into(),
247            identity: identified.then_some(digest),
248            retention,
249        }
250    };
251
252    let plan = item(
253        ItemLocation::Plan,
254        input.plan.schema(),
255        bytes_of(input.plan, ItemLocation::Plan)?,
256        true,
257    );
258    let mut descriptors = vec![];
259    for (i, d) in input.descriptors.iter().enumerate() {
260        let at = ItemLocation::Descriptor(i);
261        descriptors.push(item(at, schema::BACKEND, bytes_of(d, at)?, true));
262    }
263    let mut calibrations = vec![];
264    for (i, c) in input.calibrations.iter().enumerate() {
265        let at = ItemLocation::Calibration(i);
266        calibrations.push(item(at, schema::CALIBRATION, bytes_of(c, at)?, true));
267    }
268    let snapshot = item(
269        ItemLocation::Snapshot,
270        input.snapshot.schema(),
271        bytes_of(input.snapshot, ItemLocation::Snapshot)?,
272        true,
273    );
274    let run_item = item(
275        ItemLocation::Run,
276        run.schema(),
277        bytes_of(run, ItemLocation::Run)?,
278        false,
279    );
280    let (capture, mut supplies) = match input.capture {
281        None => (CaptureStatus::Disabled, vec![]),
282        Some(c) => (c.status, vec![]),
283    };
284    for (i, s) in input
285        .capture
286        .map(|c| c.supplies.as_slice())
287        .unwrap_or(&[])
288        .iter()
289        .enumerate()
290    {
291        let at = ItemLocation::Supply(i);
292        let value = match &s.value {
293            CapturedValue::Output(o) => item(at, schema::BACKEND_OUTPUT, bytes_of(o, at)?, false),
294            CapturedValue::Reason(r) => item(at, schema::RUNTIME_REASON, bytes_of(r, at)?, false),
295        };
296        supplies.push(SupplyEntry {
297            failure_context: s.failure_context,
298            step: s.step.clone(),
299            instance: s.instance.clone(),
300            request: s.request.clone(),
301            sequence: s.sequence,
302            attempt_id: s.attempt_id.clone(),
303            target: s.target,
304            value,
305        });
306    }
307    let mut retained_sources = vec![];
308    for (i, source) in sources.iter().enumerate() {
309        let at = ItemLocation::Source(i);
310        retained_sources.push(item(at, schema::SOURCE, bytes_of(source, at)?, true));
311    }
312    let mut pages = vec![];
313    for (i, p) in input
314        .capture
315        .into_iter()
316        .flat_map(|c| &c.acquisition_pages)
317        .enumerate()
318    {
319        let at = ItemLocation::AcquisitionPage(i);
320        let value = match &p.value {
321            CapturedPage::Page(p) => item(at, schema::ACQUISITION_PAGE, bytes_of(p, at)?, false),
322            CapturedPage::Reason(r) => item(at, schema::RUNTIME_REASON, bytes_of(r, at)?, false),
323        };
324        pages.push(AcquisitionSupply {
325            sequence: p.sequence,
326            step: p.step.clone(),
327            tier: p.tier,
328            page: p.page,
329            request: p.request.clone(),
330            attempt_id: p.attempt_id.clone(),
331            value,
332        });
333    }
334    let max_pages = input
335        .plan
336        .definition
337        .limits
338        .acquisition
339        .as_ref()
340        .map_or(0, |l| l.max_calls);
341    if pages.len() as u64 > max_pages {
342        return binding("capture pages beyond the plan's call bound");
343    }
344    if let Some(location) = not_utf8 {
345        return Err(AssemblyError::NotCanonical { location });
346    }
347    if resolved > MAX_RESOLVED_BYTES {
348        return Err(AssemblyError::TooLarge {
349            bytes: resolved,
350            limit: MAX_RESOLVED_BYTES,
351        });
352    }
353    let scope = input.scope.clone();
354    let bundle = ReplayBundle {
355        sources: (!retained_sources.is_empty()).then_some(retained_sources),
356        acquisition_pages: input.plan.acquisitions.is_some().then_some(pages),
357        schema: if input.plan.schema == schema::PLAN_V3 {
358            schema::REPLAY_V2
359        } else {
360            schema::REPLAY
361        }
362        .into(),
363        decision_id: run.decision_id.clone(),
364        scope,
365        created_at_ms: input.created_at_ms,
366        expires_at_ms: expires,
367        evaluation_time_ms: input.evaluation_time,
368        plan_id,
369        plan,
370        descriptors,
371        calibrations,
372        snapshot_id,
373        snapshot,
374        run: run_item,
375        capture,
376        supplies,
377    };
378    let size = bundle
379        .record_canonical()
380        .map_err(|_| AssemblyError::NotCanonical {
381            location: ItemLocation::Bundle,
382        })?
383        .len();
384    if size > REPLAY_V1.max_bytes {
385        return Err(AssemblyError::TooLarge {
386            bytes: size,
387            limit: REPLAY_V1.max_bytes,
388        });
389    }
390    bundle
391        .check(choice.host_cap_ms)
392        .map_err(AssemblyError::Bundle)?;
393    Ok(bundle)
394}
rustev_eval/assemble.rs

rustev_eval/
assemble.rs
