rustev_runtime/
live.rs

1//! Live re-execution with fresh observations (spec 026).
2//!
3//! [`Runtime::reexecute_live`] is the only entry point that may request
4//! fresh external work for a previously recorded decision shape, and the
5//! only one that accepts a host [`LiveAuthorizer`]. It runs a new decision:
6//! a decision id unequal to the historical one, a snapshot whose identity the
7//! runtime computes, a fresh evaluation time, fresh admission, deadline,
8//! concurrency, retry, fallback and budget state, and new observations.
9//! Nothing is read from a replay bundle; a historical case is a link (a
10//! digest plus identities) carried into the evidence for later comparison.
11//!
12//! Immediately before every external attempt, retries and fallbacks
13//! included, and after its full worst-case exposure is reserved, the host is
14//! asked about that exact attempt. A missing, expired, mismatched, errored
15//! or refused answer prevents dispatch and is recorded as
16//! `authorization_refused`; the runtime never downgrades scope, switches
17//! targets, retries or falls back to avoid a refusal. The answer is scoped
18//! to that attempt, is never exported, and authorizes nothing downstream.
19
20use std::collections::BTreeMap;
21use std::fmt;
22use std::panic::{AssertUnwindSafe, catch_unwind};
23use std::sync::{Arc, Mutex};
24
25use rustev_contract::definition::{Operation, StepBody};
26use rustev_contract::ids::{ContentDigest, SnapshotId};
27use rustev_contract::live::{
28    Admission, AttemptRole, AuthorizationOutcome, AuthorizationReference, AuthorizationRefusal,
29    AuthorizationRequest, CapturePosture, ComparisonSection, DeliveryObservation, Dispatch,
30    Exposure, Historical, HistoricalLink, HistoricalRetention, LiveEffect, LiveReexecution,
31    MAX_EFFECTS, RetentionPosture, RunLink,
32};
33use rustev_contract::plan::{PlanExecution, PlanStepDetail};
34use rustev_contract::run::{Charge, CoreEvidence, CostBound, RunRecord, Termination};
35use rustev_contract::scope::{Handle, Scope};
36use rustev_contract::snapshot::Snapshot;
37use rustev_contract::time::Timestamp;
38use rustev_contract::{Document, Identified, schema};
39use rustev_core::evaluate::SemanticRequest;
40use rustev_core::seams::CancelSignal;
41
42use crate::driver::{DETAIL_BYTES, DecisionCtx};
43use crate::wait::cut;
44use crate::{
45    CaptureConfig, CaptureConfigError, CaptureOutcome, Completion, DecideError, Decided,
46    DecisionRequest, Delivery, DeliveryFailure, PreparedPlan, Rejection, Runtime, Target, capture,
47};
48
49/// The host's answer about one proposed attempt. In process only: it is
50/// neither serializable nor clonable, the runtime keeps nothing of it but
51/// the reference, and it authorizes no action outside that one dispatch.
52#[derive(Debug)]
53pub enum DispatchAnswer {
54    /// Send exactly the attempt whose request digest is `answers`, if it is
55    /// sent within `valid_for_ms` of the question, by the runtime's clock.
56    Allow {
57        answers: ContentDigest,
58        reference: AuthorizationReference,
59        valid_for_ms: u64,
60    },
61    Refuse {
62        reference: Option<AuthorizationReference>,
63        detail: String,
64    },
65}
66
67/// The host's authorization seam (spec 026, 3.3). Rustev mints no token:
68/// the host decides, from its own trusted facts, whether this exact attempt
69/// may be sent. `Ok(None)` is no answer (missing); `Err` is a failure of the
70/// authorizer (errored); a panic is contained and errored. Every one of
71/// them prevents dispatch.
72pub trait LiveAuthorizer: Send + Sync {
73    fn authorize(&self, request: &AuthorizationRequest) -> Result<Option<DispatchAnswer>, String>;
74}
75
76/// One live re-execution to run. Nothing here is taken from a replay bundle.
77#[derive(Debug, Clone)]
78pub struct LiveRequest {
79    /// A new decision id: never the historical one, and not running.
80    pub decision_id: String,
81    /// The current snapshot, validated by the normal snapshot contract.
82    pub snapshot: Snapshot,
83    /// The identity the caller believes the snapshot has, if it states one.
84    /// Never trusted: the runtime computes the identity, and a different
85    /// stated identity is refused.
86    pub expected_snapshot_id: Option<SnapshotId>,
87    /// Supplied anew; never read from a clock or a historical record.
88    pub evaluation_time: Timestamp,
89    /// Can only tighten the plan's `deadline_ms`.
90    pub deadline_ms: Option<u64>,
91    /// Passed through to backends, as for `decide`.
92    pub principal_handle: Vec<u8>,
93    /// Opaque, non-secret subject and scope presented to the authorizer.
94    pub subject: Handle,
95    pub scope: Scope,
96    /// The new run's own capture; off unless the host chooses it.
97    pub capture: Option<CaptureConfig>,
98    /// The historical case this run may later be compared with, by link.
99    pub historical: Option<HistoricalLink>,
100}
101
102/// Why a live re-execution was refused or failed. Except for
103/// `GenerationFailed`, nothing was presented to the authorizer, dispatched
104/// or spent.
105#[derive(Debug, Clone, PartialEq, Eq)]
106pub enum LiveRefusal {
107    /// Source operations need their own authorization and spend identities.
108    /// This backend-only gated entry point refuses them before admission.
109    SpendAcquisitionUnsupported,
110    /// The decision id is the historical decision's.
111    HistoricalDecisionId,
112    /// The historical link names a decision id outside 1 to 256 bytes.
113    InvalidHistoricalLink,
114    /// The snapshot's computed identity is not the identity stated for it.
115    SnapshotIdentity {
116        stated: SnapshotId,
117        computed: SnapshotId,
118    },
119    /// The snapshot has no canonical identity.
120    InvalidSnapshot(String),
121    Capture(CaptureConfigError),
122    /// The plan can propose more attempts than one live document records.
123    Unbounded {
124        worst_case: u64,
125        max: u64,
126    },
127    /// The runtime's own admission refusal (spec 003, 3.4.1).
128    Rejected(Rejection),
129    /// A pure generation invariant failed: no text result and no record.
130    /// This is a runtime failure, not a request defect.
131    GenerationFailed(String),
132}
133
134impl fmt::Display for LiveRefusal {
135    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
136        match self {
137            LiveRefusal::SpendAcquisitionUnsupported => {
138                f.write_str("gated live acquisition dispatch is not implemented")
139            }
140            LiveRefusal::HistoricalDecisionId => {
141                f.write_str("a live re-execution needs a new decision id, not the historical one")
142            }
143            LiveRefusal::InvalidHistoricalLink => {
144                f.write_str("the historical link's decision id is not 1 to 256 bytes")
145            }
146            LiveRefusal::SnapshotIdentity { stated, computed } => write!(
147                f,
148                "identity mismatch: the snapshot is {computed}, not the stated {stated}"
149            ),
150            LiveRefusal::InvalidSnapshot(e) => write!(f, "the snapshot has no identity: {e}"),
151            LiveRefusal::Capture(e) => write!(f, "{e}"),
152            LiveRefusal::Unbounded { worst_case, max } => write!(
153                f,
154                "the plan can propose {worst_case} attempts; a live document records at most {max}"
155            ),
156            LiveRefusal::Rejected(r) => write!(f, "not admitted: {r:?}"),
157            LiveRefusal::GenerationFailed(detail) => write!(f, "generation failed: {detail}"),
158        }
159    }
160}
161
162impl std::error::Error for LiveRefusal {}
163
164/// A finished live re-execution: the ordinary result of a new decision and
165/// its live document, returned also when evidence delivery failed, so every
166/// attempt, charge and remote state stays with the caller (spec 026, 5).
167#[derive(Debug)]
168pub struct LiveResult {
169    /// First spend refusal; later attempts of this decision were not sent.
170    pub spend_refusal: Option<crate::SpendRefusal>,
171    pub completion: Completion,
172    /// The new run record; the evidence of execution.
173    pub record: Arc<RunRecord>,
174    pub delivery: Result<Delivery, DeliveryFailure>,
175    /// `None` when capture was off.
176    pub capture: Option<CaptureOutcome>,
177    pub document: LiveReexecution,
178}
179
180/// One proposed attempt as presented, and the host's answer.
181struct Presented {
182    step: String,
183    instance: Vec<String>,
184    target: u32,
185    role: AttemptRole,
186    attempt_id: String,
187    backend_id: String,
188    artifact: rustev_contract::ids::ArtifactId,
189    request: rustev_contract::ids::RequestId,
190    presented: ContentDigest,
191    exposure: Exposure,
192    remaining_ms: u64,
193    authorization: AuthorizationOutcome,
194}
195
196/// The live path's state within one decision.
197pub(crate) struct LiveCtx<'l> {
198    authorizer: &'l dyn LiveAuthorizer,
199    spend: Option<crate::LiveSpend<'l>>,
200    spend_refusal: Mutex<Option<crate::SpendRefusal>>,
201    subject: &'l Handle,
202    scope: &'l Scope,
203    operations: BTreeMap<String, Operation>,
204    presented: Mutex<Vec<Presented>>,
205}
206
207/// What [`LiveCtx::authorize`] needs to know about the proposed attempt.
208pub(crate) struct Proposal<'p> {
209    pub req: &'p SemanticRequest,
210    pub target_index: usize,
211    pub target: &'p Target,
212    pub role: AttemptRole,
213    pub attempt_id: &'p str,
214    pub bound: CostBound,
215    pub reserved: u64,
216}
217
218impl LiveCtx<'_> {
219    pub(crate) fn spend_refusal(&self) -> Option<crate::SpendRefusal> {
220        self.spend_refusal
221            .lock()
222            .unwrap_or_else(|e| e.into_inner())
223            .clone()
224    }
225
226    pub(crate) fn reserve_spend(
227        &self,
228        cx: &DecisionCtx<'_, '_>,
229        p: Proposal<'_>,
230    ) -> Result<Option<crate::spend::HeldSpend<'_>>, crate::SpendRefusal> {
231        let Some(spend) = self.spend else {
232            return Ok(None);
233        };
234        let mut refusal = self.spend_refusal.lock().unwrap_or_else(|e| e.into_inner());
235        if let Some(r) = &*refusal {
236            return Err(r.clone());
237        }
238        let binding = spend
239            .bindings
240            .iter()
241            .find(|b| b.backend_id == p.target.backend_id && b.artifact == p.target.artifact);
242        let result = match binding {
243            None => Err(crate::SpendRefusal::new("spend-unit")),
244            Some(binding) => spend.gate.reserve(&crate::SpendAttempt {
245                decision_id: cx.decision_id.to_string(),
246                attempt_id: p.attempt_id.to_string(),
247                backend_id: p.target.backend_id.clone(),
248                target: p.target_index as u32,
249                artifact: p.target.artifact.clone(),
250                operation: self.operations[&p.req.step],
251                bound: p.bound,
252                cost_unit: binding.cost_unit,
253            }),
254        };
255        match result {
256            Ok(ticket) => Ok(Some(crate::spend::HeldSpend::new(spend.gate, ticket))),
257            Err(r) => {
258                *refusal = Some(r.clone());
259                Err(r)
260            }
261        }
262    }
263
264    pub(crate) fn expire_authorization(&self, attempt_id: &str) {
265        let mut presented = self.presented.lock().unwrap_or_else(|e| e.into_inner());
266        if let Some(p) = presented.iter_mut().find(|p| p.attempt_id == attempt_id)
267            && let AuthorizationOutcome::Allowed { reference } = &p.authorization
268        {
269            p.authorization = AuthorizationOutcome::Refused {
270                refusal: AuthorizationRefusal::Expired,
271                reference: Some(reference.clone()),
272                detail: "authorization expired during spend reservation".into(),
273            };
274        }
275    }
276
277    /// Present one proposed attempt to the host, after its reservation and
278    /// immediately before dispatch, and record the answer. `Ok` only for an
279    /// allow that names this exact request and is still valid now.
280    pub(crate) fn authorize(
281        &self,
282        cx: &DecisionCtx<'_, '_>,
283        p: Proposal<'_>,
284    ) -> Result<u64, AuthorizationRefusal> {
285        let now = cx.rt.clock.now();
286        let identity = cx
287            .ev()
288            .request_document(
289                &p.req.step,
290                &p.req.instance,
291                p.target_index as u32,
292                self.scope,
293            )
294            .map_err(|e| e.to_string())
295            .and_then(|d| d.id().map_err(|e| e.to_string()));
296        let request = match identity {
297            Ok(id) => id,
298            // The request is pending and its target bound, so this is a
299            // defect; it still prevents dispatch rather than guessing.
300            Err(_) => return Err(AuthorizationRefusal::Errored),
301        };
302        let asked = AuthorizationRequest {
303            decision_id: cx.decision_id.to_string(),
304            subject: self.subject.clone(),
305            scope: self.scope.clone(),
306            step: p.req.step.clone(),
307            instance: p.req.instance.clone(),
308            operation: self.operations[&p.req.step],
309            target: p.target_index as u32,
310            role: p.role,
311            backend_id: p.target.backend_id.clone(),
312            artifact: p.target.artifact.clone(),
313            request: request.clone(),
314            attempt_id: p.attempt_id.to_string(),
315            exposure: Exposure {
316                bound: p.bound,
317                reserved_units: p.reserved,
318            },
319            remaining_ms: cx.deadline.saturating_sub(now),
320        };
321        let Ok(digest) = asked.digest() else {
322            return Err(AuthorizationRefusal::Errored);
323        };
324        let refused = |refusal, reference, detail: &str| AuthorizationOutcome::Refused {
325            refusal,
326            reference,
327            detail: cut(detail, DETAIL_BYTES),
328        };
329        let mut valid_until = now;
330        let outcome = match catch_unwind(AssertUnwindSafe(|| self.authorizer.authorize(&asked))) {
331            Err(p) => refused(
332                AuthorizationRefusal::Errored,
333                None,
334                &format!(
335                    "authorizer panicked: {}",
336                    crate::wait::panic_detail(p.as_ref())
337                ),
338            ),
339            Ok(Err(detail)) => refused(AuthorizationRefusal::Errored, None, &detail),
340            Ok(Ok(None)) => refused(AuthorizationRefusal::Missing, None, "no answer"),
341            Ok(Ok(Some(DispatchAnswer::Refuse { reference, detail }))) => {
342                refused(AuthorizationRefusal::Refused, reference, &detail)
343            }
344            Ok(Ok(Some(DispatchAnswer::Allow {
345                answers,
346                reference,
347                valid_for_ms,
348            }))) => {
349                if answers != digest {
350                    refused(
351                        AuthorizationRefusal::Mismatched,
352                        Some(reference),
353                        "the answer names another request",
354                    )
355                } else if cx.rt.clock.now() >= now.saturating_add(valid_for_ms) {
356                    refused(
357                        AuthorizationRefusal::Expired,
358                        Some(reference),
359                        &format!("the answer was valid for {valid_for_ms} ms"),
360                    )
361                } else {
362                    valid_until = now.saturating_add(valid_for_ms);
363                    AuthorizationOutcome::Allowed { reference }
364                }
365            }
366        };
367        let result = match &outcome {
368            AuthorizationOutcome::Allowed { .. } => Ok(valid_until),
369            AuthorizationOutcome::Refused { refusal, .. } => Err(*refusal),
370        };
371        self.presented
372            .lock()
373            .unwrap_or_else(|e| e.into_inner())
374            .push(Presented {
375                step: asked.step,
376                instance: asked.instance,
377                target: asked.target,
378                role: asked.role,
379                attempt_id: asked.attempt_id,
380                backend_id: asked.backend_id,
381                artifact: asked.artifact,
382                request,
383                presented: digest,
384                exposure: asked.exposure,
385                remaining_ms: asked.remaining_ms,
386                authorization: outcome,
387            });
388        result
389    }
390}
391
392/// The worst-case number of attempts a plan can propose: its execution
393/// policy's bound, or one per possible request without a policy. A refused
394/// proposal ends its request, so it replaces an attempt, never adds one.
395fn worst_case_attempts(plan: &PreparedPlan) -> u64 {
396    match &plan.compiled().plan.execution {
397        PlanExecution::Declared(d) => d.policy.max_attempts_per_decision,
398        PlanExecution::None => plan
399            .compiled()
400            .plan
401            .steps
402            .iter()
403            .filter_map(|s| match &s.detail {
404                PlanStepDetail::Semantic(b) => Some(b.max_requests),
405                _ => None,
406            })
407            .fold(0, u64::saturating_add),
408    }
409}
410
411impl Runtime {
412    /// Re-execute a decision shape live (spec 026): a new decision, under
413    /// fresh bounds, with the host authorizing every external attempt
414    /// immediately before it is sent. The only entry point that accepts an
415    /// authorizer; `decide`, capture, replay and comparison never dispatch
416    /// on its behalf.
417    ///
418    /// Refused before admission, with nothing presented or spent: a reused
419    /// historical decision id, a snapshot whose computed identity is not the
420    /// stated one, an invalid capture configuration, a plan whose proposals
421    /// exceed the document bound, and every admission refusal of `decide`.
422    ///
423    /// ```no_run
424    /// # async fn f(rt: &rustev_runtime::Runtime, plan: &rustev_runtime::PreparedPlan,
425    /// #     live: rustev_runtime::LiveRequest, a: &dyn rustev_runtime::LiveAuthorizer,
426    /// #     req: rustev_runtime::DecisionRequest) {
427    /// let cancel = rustev_core::seams::CancelSignal::new();
428    /// let _ = rt.reexecute_live(plan, live, a, &cancel).await;
429    /// let _ = rt.decide(plan, req, &cancel).await;
430    /// # }
431    /// ```
432    ///
433    /// ```compile_fail
434    /// # async fn f(rt: &rustev_runtime::Runtime, plan: &rustev_runtime::PreparedPlan,
435    /// #     req: rustev_runtime::DecisionRequest, a: &dyn rustev_runtime::LiveAuthorizer) {
436    /// // `decide` takes no authorizer: only `reexecute_live` does.
437    /// rt.decide(plan, req, &rustev_core::seams::CancelSignal::new(), a).await;
438    /// # }
439    /// ```
440    ///
441    /// ```compile_fail
442    /// # async fn f(rt: &rustev_runtime::Runtime, plan: &rustev_runtime::PreparedPlan,
443    /// #     req: rustev_runtime::DecisionRequest, c: &rustev_runtime::CaptureConfig,
444    /// #     a: &dyn rustev_runtime::LiveAuthorizer) {
445    /// // Nor does capture.
446    /// rt.decide_with_capture(plan, req, &rustev_core::seams::CancelSignal::new(), c, a).await;
447    /// # }
448    /// ```
449    pub async fn reexecute_live(
450        &self,
451        plan: &PreparedPlan,
452        req: LiveRequest,
453        authorizer: &dyn LiveAuthorizer,
454        cancel: &CancelSignal,
455    ) -> Result<LiveResult, LiveRefusal> {
456        self.reexecute_live_inner(plan, req, authorizer, cancel, None)
457            .await
458    }
459
460    /// Live re-execution with a durable spend gate, after budget and host
461    /// authorization on every attempt. Existing ungated callers are unchanged.
462    /// Plans with acquisitions are refused before admission until source
463    /// authorization and spend identities are supported.
464    pub async fn reexecute_live_with_gate(
465        &self,
466        plan: &PreparedPlan,
467        req: LiveRequest,
468        authorizer: &dyn LiveAuthorizer,
469        cancel: &CancelSignal,
470        spend: crate::LiveSpend<'_>,
471    ) -> Result<LiveResult, LiveRefusal> {
472        self.reexecute_live_inner(plan, req, authorizer, cancel, Some(spend))
473            .await
474    }
475
476    async fn reexecute_live_inner(
477        &self,
478        plan: &PreparedPlan,
479        req: LiveRequest,
480        authorizer: &dyn LiveAuthorizer,
481        cancel: &CancelSignal,
482        spend: Option<crate::LiveSpend<'_>>,
483    ) -> Result<LiveResult, LiveRefusal> {
484        if spend.is_some()
485            && plan
486                .compiled()
487                .plan
488                .acquisitions
489                .as_ref()
490                .is_some_and(|a| !a.is_empty())
491        {
492            return Err(LiveRefusal::SpendAcquisitionUnsupported);
493        }
494        if req
495            .historical
496            .as_ref()
497            .is_some_and(|h| h.decision_id == req.decision_id)
498        {
499            return Err(LiveRefusal::HistoricalDecisionId);
500        }
501        if req
502            .historical
503            .as_ref()
504            .is_some_and(|h| h.decision_id.is_empty() || h.decision_id.len() > 256)
505        {
506            return Err(LiveRefusal::InvalidHistoricalLink);
507        }
508        let snapshot_id = req
509            .snapshot
510            .id()
511            .map_err(|e| LiveRefusal::InvalidSnapshot(e.to_string()))?;
512        if let Some(stated) = &req.expected_snapshot_id
513            && *stated != snapshot_id
514        {
515            return Err(LiveRefusal::SnapshotIdentity {
516                stated: stated.clone(),
517                computed: snapshot_id,
518            });
519        }
520        if let Some(c) = &req.capture {
521            c.check().map_err(LiveRefusal::Capture)?;
522        }
523        let worst_case = worst_case_attempts(plan);
524        if worst_case > MAX_EFFECTS as u64 {
525            return Err(LiveRefusal::Unbounded {
526                worst_case,
527                max: MAX_EFFECTS as u64,
528            });
529        }
530
531        let operations = plan
532            .compiled()
533            .plan
534            .definition
535            .steps
536            .iter()
537            .filter_map(|s| match &s.body {
538                StepBody::Semantic(d) => Some((s.id.clone(), d.operation)),
539                _ => None,
540            })
541            .collect();
542        let live = LiveCtx {
543            authorizer,
544            spend,
545            spend_refusal: Mutex::new(None),
546            subject: &req.subject,
547            scope: &req.scope,
548            operations,
549            presented: Mutex::new(vec![]),
550        };
551        let decision = DecisionRequest {
552            decision_id: req.decision_id.clone(),
553            snapshot: req.snapshot.clone(),
554            evaluation_time: req.evaluation_time,
555            deadline_ms: req.deadline_ms,
556            principal_handle: req.principal_handle.clone(),
557        };
558        let mut capturing = req.capture.as_ref().map(capture::Capturing::new);
559        let result = self
560            .run(
561                plan,
562                decision,
563                cancel,
564                capturing.as_mut(),
565                Some(&live),
566                None,
567                None,
568            )
569            .await;
570        let (record, delivery) = match result {
571            Err(DecideError::GenerationFailed(detail)) => {
572                return Err(LiveRefusal::GenerationFailed(detail));
573            }
574            Err(DecideError::Rejected(r)) => return Err(LiveRefusal::Rejected(r)),
575            Ok(Decided {
576                record, delivery, ..
577            }) => (record, Ok(delivery)),
578            Err(DecideError::EvidenceNotDelivered { record, failure }) => (record, Err(failure)),
579        };
580        let capture = capturing.map(|c| {
581            CaptureOutcome::Captured(c.finish(
582                &req.decision_id,
583                record.termination == Termination::Cancelled,
584            ))
585        });
586        let spend_refusal = live.spend_refusal();
587        let presented = live
588            .presented
589            .into_inner()
590            .unwrap_or_else(|e| e.into_inner());
591        let document = document(&req, snapshot_id, &record, &delivery, &capture, presented);
592        let completion = match &record.core {
593            CoreEvidence::Recorded(e) if record.termination == Termination::Judged => {
594                Completion::Judged(Box::new(e.judgment.clone()))
595            }
596            _ => Completion::Cancelled,
597        };
598        Ok(LiveResult {
599            spend_refusal,
600            completion,
601            record,
602            delivery,
603            capture,
604            document,
605        })
606    }
607}
608
609fn document(
610    req: &LiveRequest,
611    snapshot_id: SnapshotId,
612    record: &RunRecord,
613    delivery: &Result<Delivery, DeliveryFailure>,
614    capture: &Option<CaptureOutcome>,
615    presented: Vec<Presented>,
616) -> LiveReexecution {
617    let attempts: BTreeMap<&str, &rustev_contract::run::AttemptRecord> = record
618        .requests
619        .iter()
620        .flat_map(|r| &r.attempts)
621        .map(|a| (a.attempt_id.as_str(), a))
622        .collect();
623    let effects = presented
624        .into_iter()
625        .map(|p| {
626            let dispatch = match attempts.get(p.attempt_id.as_str()) {
627                Some(a) => Dispatch::Dispatched {
628                    end: a.end.clone(),
629                    cancellation: a.cancellation,
630                    remote: a.remote,
631                    charge: a.cost.charge,
632                },
633                None => Dispatch::NotDispatched,
634            };
635            LiveEffect {
636                step: p.step,
637                instance: p.instance,
638                target: p.target,
639                role: p.role,
640                attempt_id: p.attempt_id,
641                backend_id: p.backend_id,
642                artifact: p.artifact,
643                request: p.request,
644                presented: p.presented,
645                exposure: p.exposure,
646                remaining_ms: p.remaining_ms,
647                authorization: p.authorization,
648                dispatch,
649            }
650        })
651        .collect();
652    let unreconciled = attempts
653        .values()
654        .filter(|a| a.cost.charge == Charge::Unknown)
655        .map(|a| a.attempt_id.clone())
656        .collect();
657    let delivery = match delivery {
658        Ok(Delivery::Acknowledged { receipt }) => DeliveryObservation::Acknowledged {
659            receipt: cut(receipt, DETAIL_BYTES),
660        },
661        Ok(Delivery::Queued(_)) => DeliveryObservation::Queued,
662        Ok(Delivery::Dropped { dropped_total }) => DeliveryObservation::Dropped {
663            dropped_total: *dropped_total,
664        },
665        Err(f) => DeliveryObservation::NotDelivered {
666            detail: cut(&f.detail, DETAIL_BYTES),
667            uncertain: f.uncertain,
668        },
669    };
670    let capture = match (&req.capture, capture) {
671        (Some(c), Some(CaptureOutcome::Captured(k))) => CapturePosture::On {
672            scope: c.scope.clone(),
673            max_bytes: c.max_bytes,
674            status: k.status,
675        },
676        _ => CapturePosture::Off,
677    };
678    LiveReexecution {
679        schema: schema::LIVE_REEXECUTION.into(),
680        decision_id: req.decision_id.clone(),
681        plan_id: record.plan_id.clone(),
682        snapshot_id,
683        evaluation_time_ms: req.evaluation_time,
684        subject: req.subject.clone(),
685        scope: req.scope.clone(),
686        historical: match &req.historical {
687            Some(h) => Historical::Linked(h.clone()),
688            None => Historical::None,
689        },
690        admission: Admission {
691            queued_ms: record.timing.queued_ms,
692            deadline_ms: record.timing.deadline_ms,
693        },
694        run_digest: record.record_digest().ok(),
695        run: RunLink {
696            termination: record.termination,
697            cost: record.cost,
698        },
699        effects,
700        unreconciled,
701        delivery,
702        retention: RetentionPosture {
703            capture,
704            historical: if req.historical.is_some() {
705                HistoricalRetention::NotObserved
706            } else {
707                HistoricalRetention::NoHistory
708            },
709        },
710        comparison: ComparisonSection::NotRequested,
711    }
712}
rustev_runtime/live.rs

rustev_runtime/
live.rs
