rustev_eval/
live.rs

1//! Comparison of a finished live re-execution with its historical case
2//! (spec 026, 3.7 and 3.8).
3//!
4//! This is offline evaluation over documents the host supplies: the
5//! historical bundle (resolved through the host's synchronous [`Resolver`]
6//! at a caller-supplied time), the live run's `rustev.live-reexecution/1`
7//! document, its new run record and its plan. Nothing here dispatches a
8//! backend, reads a clock or reaches the runtime: the live run already
9//! happened, through the runtime's separately named entry point.
10//!
11//! The historical case is first verified by spec 004 reproduction. Then an
12//! explicitly named proposal payload or outcome category, chosen by a
13//! versioned task adapter, is compared. Full judgment bytes are not the
14//! agreement metric, because decision, plan, snapshot and time may
15//! legitimately differ. Changed output is `live-diverged`, never a failed
16//! historical reproduction; unchanged output is agreement on the named
17//! fields only. An expired, erased or otherwise unavailable historical item
18//! makes the comparison unavailable with that reason; nothing is restored
19//! from a digest, and no source document is modified.
20
21use std::collections::BTreeSet;
22
23use rustev_contract::Identified;
24use rustev_contract::ids::{ArtifactId, CalibrationId};
25use rustev_contract::judgment::Judgment;
26use rustev_contract::live::{
27    CapturePosture, Change, ComparisonDefinition, ComparisonOutcome, ComparisonSection,
28    CostComparison, DeliveryObservation, Historical, HistoricalLink, HistoricalRetention,
29    IdentityChanges, ItemObservation, LiveComparison, LiveReexecution, ObservationChanges,
30    OtherChanges, ResultChange, RuntimeObservations,
31};
32use rustev_contract::plan::{BoundCalibration, Plan, PlanExecution, PlanStepDetail};
33use rustev_contract::replay::{CaptureStatus, ReplayBundle};
34use rustev_contract::run::{
35    Cancellation, CoreEvidence, RemoteState, RequestResult, RunRecord, Termination, Transition,
36};
37
38use crate::config::Agreement;
39use crate::dataset::AdapterRef;
40use crate::metrics::CaseOutcome;
41use crate::replay::{Incomparable, ReplayConfig, ReplayOutcome, reproduce};
42use crate::resolve::Resolver;
43
44/// What every live comparison states it does not establish.
45pub const LIMITATIONS: [&str; 5] = [
46    "agreement holds for the named adapter and fields only; it is not determinism, calibration, \
47     quality, correctness or authorization",
48    "a served identity is not recorded by rustev.run/1, so served-identity change is unknown",
49    "historical latency, attempts and cost are copied observations, never remeasured",
50    "a digest proves integrity relative to expected bytes, never truth or producer identity",
51    "cost units are logical units of each backend, not money",
52];
53
54/// What every live comparison leaves out.
55pub const EXCLUSIONS: [&str; 2] = [
56    "full judgment bytes: decision, plan, snapshot and time may legitimately differ",
57    "proposal payload fields the adapter does not name",
58];
59
60/// One finished live re-execution and the historical case it links to.
61pub struct LiveCase<'a> {
62    pub bundle: &'a ReplayBundle,
63    pub resolver: &'a dyn Resolver,
64    /// The caller's trusted scope and the time availability is judged at.
65    pub config: &'a ReplayConfig,
66    pub live: &'a LiveReexecution,
67    /// The new run record the live document binds.
68    pub live_run: &'a RunRecord,
69    /// The plan the live run executed; its identity must be the document's.
70    pub live_plan: &'a Plan,
71    /// The versioned task adapter whose named fields are compared.
72    pub adapter: &'a AdapterRef,
73    pub agreement: &'a Agreement,
74}
75
76fn incomparable(reason: &str) -> ComparisonOutcome {
77    ComparisonOutcome::Incomparable {
78        reason: reason.into(),
79    }
80}
81
82fn observations(run: &RunRecord) -> RuntimeObservations {
83    let attempts = run.requests.iter().flat_map(|r| &r.attempts);
84    let transitions = run.requests.iter().flat_map(|r| &r.transitions);
85    RuntimeObservations {
86        attempts: attempts.clone().count() as u64,
87        retries: transitions
88            .clone()
89            .filter(|t| matches!(t, Transition::Retry { .. }))
90            .count() as u64,
91        fallbacks: transitions
92            .filter(|t| matches!(t, Transition::Fallback { .. }))
93            .count() as u64,
94        cancellations_requested: attempts
95            .clone()
96            .filter(|a| matches!(a.cancellation, Cancellation::Requested { .. }))
97            .count() as u64,
98        possibly_continuing: attempts
99            .filter(|a| a.remote == RemoteState::PossiblyContinuing)
100            .count() as u64,
101        deadline_expired: run.timing.deadline_expired,
102    }
103}
104
105/// The artifact of `step`'s target `target` in `plan`.
106fn bound_artifact(plan: &Plan, step: &str, target: u32) -> Option<ArtifactId> {
107    let Some(PlanStepDetail::Semantic(b)) =
108        plan.steps.iter().find(|s| s.id == step).map(|s| &s.detail)
109    else {
110        return None;
111    };
112    if target == 0 {
113        return Some(b.artifact.clone());
114    }
115    match &plan.execution {
116        PlanExecution::Declared(d) => d
117            .fallbacks
118            .iter()
119            .find(|f| f.step == step && f.target == target)
120            .map(|f| f.artifact.clone()),
121        PlanExecution::None => None,
122    }
123}
124
125/// The artifacts whose outputs were supplied.
126fn output_artifacts(plan: &Plan, run: &RunRecord) -> Option<BTreeSet<ArtifactId>> {
127    run.requests
128        .iter()
129        .filter_map(|r| match r.result {
130            RequestResult::Output { target } => Some(bound_artifact(plan, &r.step, target)),
131            _ => None,
132        })
133        .collect()
134}
135
136fn backends(run: &RunRecord) -> BTreeSet<&str> {
137    run.requests
138        .iter()
139        .flat_map(|r| &r.attempts)
140        .map(|a| a.backend_id.as_str())
141        .collect()
142}
143
144fn calibrations(plan: &Plan) -> BTreeSet<CalibrationId> {
145    plan.steps
146        .iter()
147        .filter_map(|s| match &s.detail {
148            PlanStepDetail::Semantic(b) => match &b.calibration {
149                BoundCalibration::Bound { id, .. } => Some(id.clone()),
150                BoundCalibration::None => None,
151            },
152            _ => None,
153        })
154        .collect()
155}
156
157fn judgment(run: &RunRecord) -> Option<&Judgment> {
158    match (&run.termination, &run.core) {
159        (Termination::Judged, CoreEvidence::Recorded(e)) => Some(&e.judgment),
160        _ => None,
161    }
162}
163
164fn result_change(historical: &CaseOutcome, live: &CaseOutcome) -> ResultChange {
165    match (historical, live) {
166        _ if historical.name() == live.name() => ResultChange::Unchanged,
167        (CaseOutcome::Abstention { .. }, CaseOutcome::Abstention { .. }) => {
168            ResultChange::UnresolvedCategoryChanged
169        }
170        _ => ResultChange::ProposalChanged,
171    }
172}
173
174fn delivery_change(live: &DeliveryObservation) -> Change {
175    // The historical record is the delivered one the bundle retains.
176    match live {
177        DeliveryObservation::Acknowledged { .. } => Change::Unchanged,
178        DeliveryObservation::Queued => Change::Unknown,
179        DeliveryObservation::Dropped { .. } | DeliveryObservation::NotDelivered { .. } => {
180            Change::Changed
181        }
182    }
183}
184
185fn retention_change(live: &CapturePosture) -> Change {
186    // A reproduced historical case had a complete capture.
187    match live {
188        CapturePosture::On {
189            status: CaptureStatus::Complete,
190            ..
191        } => Change::Unchanged,
192        _ => Change::Changed,
193    }
194}
195
196/// Compare a finished live re-execution with its historical case. Returns
197/// the comparison and what was observed of the historical case's retention.
198pub fn compare_live(case: &LiveCase<'_>) -> (LiveComparison, HistoricalRetention) {
199    let live = case.live;
200    let live_obs = observations(case.live_run);
201    let observed_retention = |unavailable: Vec<ItemObservation>| HistoricalRetention::Observed {
202        observed_at_ms: case.config.now_ms,
203        bundle_expires_at_ms: case.bundle.expires_at_ms,
204        unavailable,
205    };
206    let link = match &live.historical {
207        Historical::Linked(l) => Some(l),
208        Historical::None => None,
209    };
210    let change_or_unknown = |f: &dyn Fn(&HistoricalLink) -> Change| link.map_or(Change::Unknown, f);
211    let mut comparison = LiveComparison {
212        definition: ComparisonDefinition {
213            adapter: case.adapter.name.clone(),
214            adapter_version: case.adapter.version.clone(),
215            fields: case.agreement.params.clone(),
216        },
217        compared_at_ms: case.config.now_ms,
218        outcome: incomparable("not-compared"),
219        historical_result: None,
220        live_result: judgment(case.live_run).map(|j| CaseOutcome::of(j, case.agreement).name()),
221        result: ResultChange::NotCompared,
222        identities: IdentityChanges {
223            snapshot: change_or_unknown(&|l| Change::of(&l.snapshot_id, &live.snapshot_id)),
224            plan: change_or_unknown(&|l| Change::of(&l.plan_id, &live.plan_id)),
225            evaluation_time: change_or_unknown(&|l| {
226                Change::of(&l.evaluation_time_ms, &live.evaluation_time_ms)
227            }),
228            artifacts: Change::Unknown,
229            served_identity: Change::Unknown,
230        },
231        observations: ObservationChanges {
232            backends: Change::Unknown,
233            policy: Change::Unknown,
234            execution_policy: Change::Unknown,
235            calibration: Change::Unknown,
236            historical_runtime: None,
237            live_runtime: live_obs,
238        },
239        cost: CostComparison {
240            historical: None,
241            live: case.live_run.cost,
242        },
243        other: OtherChanges {
244            termination: Change::Unknown,
245            cancellation: Change::Unknown,
246            remote_state: Change::Unknown,
247            retention: Change::Unknown,
248            delivery: Change::Unknown,
249        },
250        exclusions: EXCLUSIONS.iter().map(|s| s.to_string()).collect(),
251        limitations: LIMITATIONS.iter().map(|s| s.to_string()).collect(),
252    };
253
254    // The live side: a well-formed document bound to this run and plan.
255    if live.check().is_err() {
256        comparison.outcome = incomparable("live-document-invalid");
257        return (comparison, HistoricalRetention::NotObserved);
258    }
259    if !live.binds(case.live_run) {
260        comparison.outcome = incomparable("live-run-mismatch");
261        return (comparison, HistoricalRetention::NotObserved);
262    }
263    if case.live_plan.id().ok().as_ref() != Some(&live.plan_id) {
264        comparison.outcome = incomparable("live-plan-mismatch");
265        return (comparison, HistoricalRetention::NotObserved);
266    }
267    // The historical side: exactly the linked bundle.
268    let Some(link) = link else {
269        comparison.outcome = incomparable("no-historical-case");
270        return (comparison, HistoricalRetention::NoHistory);
271    };
272    if HistoricalLink::of(case.bundle).ok().as_ref() != Some(link) {
273        comparison.outcome = incomparable("historical-link-mismatch");
274        return (comparison, HistoricalRetention::NotObserved);
275    }
276    // Verify the historical case under spec 004 first.
277    let reproduced = match reproduce(case.bundle, case.resolver, case.config) {
278        ReplayOutcome::Reproduced(r) => r,
279        ReplayOutcome::Diverged { .. } => {
280            comparison.outcome = incomparable("historical-diverged");
281            return (comparison, observed_retention(vec![]));
282        }
283        ReplayOutcome::Incomparable(Incomparable::Unavailable {
284            location,
285            availability,
286        }) => {
287            let location = location.to_string();
288            comparison.outcome = ComparisonOutcome::Unavailable {
289                location: location.clone(),
290                availability,
291            };
292            return (
293                comparison,
294                observed_retention(vec![ItemObservation {
295                    location,
296                    availability,
297                }]),
298            );
299        }
300        ReplayOutcome::Incomparable(i) => {
301            comparison.outcome = incomparable(&format!("historical-{}", i.code()));
302            return (comparison, observed_retention(vec![]));
303        }
304    };
305    let historical_run = reproduced.run();
306    let historical_plan = reproduced.plan();
307    let hist_obs = observations(historical_run);
308
309    comparison.identities.artifacts = match (
310        output_artifacts(historical_plan, historical_run),
311        output_artifacts(case.live_plan, case.live_run),
312    ) {
313        (Some(h), Some(l)) => Change::of(&h, &l),
314        _ => Change::Unknown,
315    };
316    comparison.observations = ObservationChanges {
317        backends: Change::of(&backends(historical_run), &backends(case.live_run)),
318        policy: Change::of(
319            &historical_plan.definition_id,
320            &case.live_plan.definition_id,
321        ),
322        execution_policy: Change::of(&historical_run.execution, &case.live_run.execution),
323        calibration: Change::of(
324            &calibrations(historical_plan),
325            &calibrations(case.live_plan),
326        ),
327        historical_runtime: Some(hist_obs),
328        live_runtime: live_obs,
329    };
330    comparison.cost.historical = Some(historical_run.cost);
331    comparison.other = OtherChanges {
332        termination: Change::of(&historical_run.termination, &case.live_run.termination),
333        cancellation: Change::of(
334            &(hist_obs.cancellations_requested > 0),
335            &(live_obs.cancellations_requested > 0),
336        ),
337        remote_state: Change::of(
338            &(hist_obs.possibly_continuing > 0),
339            &(live_obs.possibly_continuing > 0),
340        ),
341        retention: retention_change(&live.retention.capture),
342        delivery: delivery_change(&live.delivery),
343    };
344    let historical = CaseOutcome::of(reproduced.judgment(), case.agreement);
345    comparison.historical_result = Some(historical.name());
346    match judgment(case.live_run) {
347        // A cancelled live run reached a terminal outcome without a
348        // judgment: nothing to agree or diverge with.
349        None => comparison.outcome = incomparable("live-not-judged"),
350        Some(j) => {
351            let live_outcome = CaseOutcome::of(j, case.agreement);
352            comparison.result = result_change(&historical, &live_outcome);
353            comparison.outcome = match comparison.result {
354                ResultChange::Unchanged => ComparisonOutcome::Agreement,
355                _ => ComparisonOutcome::LiveDiverged,
356            };
357        }
358    }
359    (comparison, observed_retention(vec![]))
360}
361
362/// A copy of `live` carrying the comparison and the observed retention.
363/// The historical sources and `live` itself are left as they were.
364pub fn with_comparison(
365    live: &LiveReexecution,
366    comparison: LiveComparison,
367    historical: HistoricalRetention,
368) -> LiveReexecution {
369    let mut out = live.clone();
370    out.retention.historical = historical;
371    out.comparison = ComparisonSection::Completed(Box::new(comparison));
372    out
373}
rustev_eval/live.rs

rustev_eval/
live.rs
