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                PlanStepDetail::Rank(b) => Some(b.max_requests),
406                _ => None,
407            })
408            .fold(0, u64::saturating_add),
409    }
410}
411
412impl Runtime {
413    /// Re-execute a decision shape live (spec 026): a new decision, under
414    /// fresh bounds, with the host authorizing every external attempt
415    /// immediately before it is sent. The only entry point that accepts an
416    /// authorizer; `decide`, capture, replay and comparison never dispatch
417    /// on its behalf.
418    ///
419    /// Refused before admission, with nothing presented or spent: a reused
420    /// historical decision id, a snapshot whose computed identity is not the
421    /// stated one, an invalid capture configuration, a plan whose proposals
422    /// exceed the document bound, and every admission refusal of `decide`.
423    ///
424    /// ```no_run
425    /// # async fn f(rt: &rustev_runtime::Runtime, plan: &rustev_runtime::PreparedPlan,
426    /// #     live: rustev_runtime::LiveRequest, a: &dyn rustev_runtime::LiveAuthorizer,
427    /// #     req: rustev_runtime::DecisionRequest) {
428    /// let cancel = rustev_core::seams::CancelSignal::new();
429    /// let _ = rt.reexecute_live(plan, live, a, &cancel).await;
430    /// let _ = rt.decide(plan, req, &cancel).await;
431    /// # }
432    /// ```
433    ///
434    /// ```compile_fail
435    /// # async fn f(rt: &rustev_runtime::Runtime, plan: &rustev_runtime::PreparedPlan,
436    /// #     req: rustev_runtime::DecisionRequest, a: &dyn rustev_runtime::LiveAuthorizer) {
437    /// // `decide` takes no authorizer: only `reexecute_live` does.
438    /// rt.decide(plan, req, &rustev_core::seams::CancelSignal::new(), a).await;
439    /// # }
440    /// ```
441    ///
442    /// ```compile_fail
443    /// # async fn f(rt: &rustev_runtime::Runtime, plan: &rustev_runtime::PreparedPlan,
444    /// #     req: rustev_runtime::DecisionRequest, c: &rustev_runtime::CaptureConfig,
445    /// #     a: &dyn rustev_runtime::LiveAuthorizer) {
446    /// // Nor does capture.
447    /// rt.decide_with_capture(plan, req, &rustev_core::seams::CancelSignal::new(), c, a).await;
448    /// # }
449    /// ```
450    pub async fn reexecute_live(
451        &self,
452        plan: &PreparedPlan,
453        req: LiveRequest,
454        authorizer: &dyn LiveAuthorizer,
455        cancel: &CancelSignal,
456    ) -> Result<LiveResult, LiveRefusal> {
457        self.reexecute_live_inner(plan, req, authorizer, cancel, None)
458            .await
459    }
460
461    /// Live re-execution with a durable spend gate, after budget and host
462    /// authorization on every attempt. Existing ungated callers are unchanged.
463    /// Plans with acquisitions are refused before admission until source
464    /// authorization and spend identities are supported.
465    pub async fn reexecute_live_with_gate(
466        &self,
467        plan: &PreparedPlan,
468        req: LiveRequest,
469        authorizer: &dyn LiveAuthorizer,
470        cancel: &CancelSignal,
471        spend: crate::LiveSpend<'_>,
472    ) -> Result<LiveResult, LiveRefusal> {
473        self.reexecute_live_inner(plan, req, authorizer, cancel, Some(spend))
474            .await
475    }
476
477    async fn reexecute_live_inner(
478        &self,
479        plan: &PreparedPlan,
480        req: LiveRequest,
481        authorizer: &dyn LiveAuthorizer,
482        cancel: &CancelSignal,
483        spend: Option<crate::LiveSpend<'_>>,
484    ) -> Result<LiveResult, LiveRefusal> {
485        if spend.is_some()
486            && plan
487                .compiled()
488                .plan
489                .acquisitions
490                .as_ref()
491                .is_some_and(|a| !a.is_empty())
492        {
493            return Err(LiveRefusal::SpendAcquisitionUnsupported);
494        }
495        if req
496            .historical
497            .as_ref()
498            .is_some_and(|h| h.decision_id == req.decision_id)
499        {
500            return Err(LiveRefusal::HistoricalDecisionId);
501        }
502        if req
503            .historical
504            .as_ref()
505            .is_some_and(|h| h.decision_id.is_empty() || h.decision_id.len() > 256)
506        {
507            return Err(LiveRefusal::InvalidHistoricalLink);
508        }
509        let snapshot_id = req
510            .snapshot
511            .id()
512            .map_err(|e| LiveRefusal::InvalidSnapshot(e.to_string()))?;
513        if let Some(stated) = &req.expected_snapshot_id
514            && *stated != snapshot_id
515        {
516            return Err(LiveRefusal::SnapshotIdentity {
517                stated: stated.clone(),
518                computed: snapshot_id,
519            });
520        }
521        if let Some(c) = &req.capture {
522            c.check().map_err(LiveRefusal::Capture)?;
523        }
524        let worst_case = worst_case_attempts(plan);
525        if worst_case > MAX_EFFECTS as u64 {
526            return Err(LiveRefusal::Unbounded {
527                worst_case,
528                max: MAX_EFFECTS as u64,
529            });
530        }
531
532        let operations = plan
533            .compiled()
534            .plan
535            .definition
536            .steps
537            .iter()
538            .filter_map(|s| match &s.body {
539                StepBody::Semantic(d) => Some((s.id.clone(), d.operation)),
540                StepBody::Rank(_) => {
541                    Some((s.id.clone(), rustev_contract::definition::Operation::Rank))
542                }
543                _ => None,
544            })
545            .collect();
546        let live = LiveCtx {
547            authorizer,
548            spend,
549            spend_refusal: Mutex::new(None),
550            subject: &req.subject,
551            scope: &req.scope,
552            operations,
553            presented: Mutex::new(vec![]),
554        };
555        let decision = DecisionRequest {
556            decision_id: req.decision_id.clone(),
557            snapshot: req.snapshot.clone(),
558            evaluation_time: req.evaluation_time,
559            deadline_ms: req.deadline_ms,
560            principal_handle: req.principal_handle.clone(),
561        };
562        let mut capturing = req.capture.as_ref().map(capture::Capturing::new);
563        let result = self
564            .run(
565                plan,
566                decision,
567                cancel,
568                capturing.as_mut(),
569                Some(&live),
570                None,
571                None,
572                &BTreeMap::new(),
573            )
574            .await;
575        let (record, delivery) = match result {
576            Err(DecideError::GenerationFailed(detail)) => {
577                return Err(LiveRefusal::GenerationFailed(detail));
578            }
579            Err(DecideError::Rejected(r)) => return Err(LiveRefusal::Rejected(r)),
580            Ok(Decided {
581                record, delivery, ..
582            }) => (record, Ok(delivery)),
583            Err(DecideError::EvidenceNotDelivered { record, failure }) => (record, Err(failure)),
584        };
585        let capture = capturing.map(|c| {
586            CaptureOutcome::Captured(c.finish(
587                &req.decision_id,
588                record.termination == Termination::Cancelled,
589            ))
590        });
591        let spend_refusal = live.spend_refusal();
592        let presented = live
593            .presented
594            .into_inner()
595            .unwrap_or_else(|e| e.into_inner());
596        let document = document(&req, snapshot_id, &record, &delivery, &capture, presented);
597        let completion = match &record.core {
598            CoreEvidence::Recorded(e) if record.termination == Termination::Judged => {
599                Completion::Judged(Box::new(e.judgment.clone()))
600            }
601            _ => Completion::Cancelled,
602        };
603        Ok(LiveResult {
604            spend_refusal,
605            completion,
606            record,
607            delivery,
608            capture,
609            document,
610        })
611    }
612}
613
614fn document(
615    req: &LiveRequest,
616    snapshot_id: SnapshotId,
617    record: &RunRecord,
618    delivery: &Result<Delivery, DeliveryFailure>,
619    capture: &Option<CaptureOutcome>,
620    presented: Vec<Presented>,
621) -> LiveReexecution {
622    let attempts: BTreeMap<&str, &rustev_contract::run::AttemptRecord> = record
623        .requests
624        .iter()
625        .flat_map(|r| &r.attempts)
626        .map(|a| (a.attempt_id.as_str(), a))
627        .collect();
628    let effects = presented
629        .into_iter()
630        .map(|p| {
631            let dispatch = match attempts.get(p.attempt_id.as_str()) {
632                Some(a) => Dispatch::Dispatched {
633                    end: a.end.clone(),
634                    cancellation: a.cancellation,
635                    remote: a.remote,
636                    charge: a.cost.charge,
637                },
638                None => Dispatch::NotDispatched,
639            };
640            LiveEffect {
641                step: p.step,
642                instance: p.instance,
643                target: p.target,
644                role: p.role,
645                attempt_id: p.attempt_id,
646                backend_id: p.backend_id,
647                artifact: p.artifact,
648                request: p.request,
649                presented: p.presented,
650                exposure: p.exposure,
651                remaining_ms: p.remaining_ms,
652                authorization: p.authorization,
653                dispatch,
654            }
655        })
656        .collect();
657    let unreconciled = attempts
658        .values()
659        .filter(|a| a.cost.charge == Charge::Unknown)
660        .map(|a| a.attempt_id.clone())
661        .collect();
662    let delivery = match delivery {
663        Ok(Delivery::Acknowledged { receipt }) => DeliveryObservation::Acknowledged {
664            receipt: cut(receipt, DETAIL_BYTES),
665        },
666        Ok(Delivery::Queued(_)) => DeliveryObservation::Queued,
667        Ok(Delivery::Dropped { dropped_total }) => DeliveryObservation::Dropped {
668            dropped_total: *dropped_total,
669        },
670        Err(f) => DeliveryObservation::NotDelivered {
671            detail: cut(&f.detail, DETAIL_BYTES),
672            uncertain: f.uncertain,
673        },
674    };
675    let capture = match (&req.capture, capture) {
676        (Some(c), Some(CaptureOutcome::Captured(k))) => CapturePosture::On {
677            scope: c.scope.clone(),
678            max_bytes: c.max_bytes,
679            status: k.status,
680        },
681        _ => CapturePosture::Off,
682    };
683    LiveReexecution {
684        schema: schema::LIVE_REEXECUTION.into(),
685        decision_id: req.decision_id.clone(),
686        plan_id: record.plan_id.clone(),
687        snapshot_id,
688        evaluation_time_ms: req.evaluation_time,
689        subject: req.subject.clone(),
690        scope: req.scope.clone(),
691        historical: match &req.historical {
692            Some(h) => Historical::Linked(h.clone()),
693            None => Historical::None,
694        },
695        admission: Admission {
696            queued_ms: record.timing.queued_ms,
697            deadline_ms: record.timing.deadline_ms,
698        },
699        run_digest: record.record_digest().ok(),
700        run: RunLink {
701            termination: record.termination,
702            cost: record.cost,
703        },
704        effects,
705        unreconciled,
706        delivery,
707        retention: RetentionPosture {
708            capture,
709            historical: if req.historical.is_some() {
710                HistoricalRetention::NotObserved
711            } else {
712                HistoricalRetention::NoHistory
713            },
714        },
715        comparison: ComparisonSection::NotRequested,
716    }
717}
rustev_runtime/live.rs

rustev_runtime/
live.rs
