use std::collections::{BTreeMap, BTreeSet};
use std::fmt;
use rustev_contract::calibration::CalibrationArtifact;
use rustev_contract::descriptor::BackendDescriptor;
use rustev_contract::generation_result::TextResult;
use rustev_contract::ids::{ArtifactId, RequestId};
use rustev_contract::judgment::{Judgment, Unresolved};
use rustev_contract::output::{BackendOutputDoc, RawOutput};
use rustev_contract::plan::{Plan, PlanExecution, PlanStepDetail};
use rustev_contract::reason::RuntimeReasonDoc;
use rustev_contract::replay::{
BundleError, CaptureStatus, ItemLocation, MAX_BUNDLE_ITEMS, ReplayBundle,
};
use rustev_contract::retention::Availability;
use rustev_contract::run::{CoreEvidence, RequestResult, RunRecord, Termination};
use rustev_contract::scope::Scope;
use rustev_contract::snapshot::Snapshot;
use rustev_contract::time::Timestamp;
use rustev_contract::{Document, Identified, schema};
use rustev_core::evaluate::{Evaluation, Supplied};
use rustev_core::generation_eval::GenerationEvaluation;
use rustev_core::{Compiled, LoadError};
use crate::acquisition_replay::{MissingAcquisition, PageValue, fill_missing, supply_page};
use crate::resolve::{Budget, ItemFailure, Resolver, check_bytes, item_bytes};
#[derive(Debug, Clone)]
pub struct ReplayConfig {
pub scope: Scope,
pub now_ms: Timestamp,
pub host_cap_ms: Option<u64>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Inconsistency {
DecisionId,
PlanId,
SnapshotId,
EvaluationTime,
TooManySupplies {
max: u64,
},
NotPending,
NotInRun,
Origin,
UnknownTarget,
RequestIdentity,
Value,
OutputBinding,
OutputRefused,
MissingSupply {
step: String,
instance: Vec<String>,
},
RequestOrder,
StartRefused,
FinishRefused,
NoRecordForm,
}
#[derive(Debug, Clone, PartialEq)]
pub enum Incomparable {
AcquisitionNotCaptured {
missing: Vec<MissingAcquisition>,
judgment: Box<Judgment>,
},
ScopeMismatch,
InvalidBundle(BundleError),
CaptureIncomplete(CaptureStatus),
Unavailable {
location: ItemLocation,
availability: Availability,
},
Oversized {
location: ItemLocation,
},
NoExpectedJudgment,
CompilerChanged {
recorded_compiler: String,
current_compiler: String,
recorded_registry: String,
current_registry: String,
},
MissingDependency {
detail: String,
},
PlanMismatch,
Inconsistent {
location: ItemLocation,
what: Inconsistency,
},
}
impl Incomparable {
pub fn code(&self) -> String {
match self {
Incomparable::AcquisitionNotCaptured { .. } => "acquisition-not-captured".into(),
Incomparable::ScopeMismatch => "scope-mismatch".into(),
Incomparable::InvalidBundle(_) => "invalid-bundle".into(),
Incomparable::CaptureIncomplete(_) => "capture-incomplete".into(),
Incomparable::Unavailable {
location: ItemLocation::AcquisitionPage(_),
availability: Availability::Missing,
} => "content-not-retained".into(),
Incomparable::Unavailable { availability, .. } => availability.name().into(),
Incomparable::Oversized { .. } => "oversized".into(),
Incomparable::NoExpectedJudgment => "no-expected-judgment".into(),
Incomparable::CompilerChanged { .. } => "compiler-changed".into(),
Incomparable::MissingDependency { .. } => "missing-dependency".into(),
Incomparable::PlanMismatch => "plan-mismatch".into(),
Incomparable::Inconsistent { .. } => "inconsistent".into(),
}
}
}
impl fmt::Display for Incomparable {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Incomparable::Unavailable {
location,
availability,
} => write!(f, "{location} is {availability}"),
Incomparable::Oversized { location } => write!(f, "{location} is oversized"),
Incomparable::Inconsistent { location, what } => {
write!(f, "inconsistent at {location}: {what:?}")
}
Incomparable::InvalidBundle(e) => write!(f, "invalid bundle: {e}"),
other => f.write_str(&other.code()),
}
}
}
#[derive(Debug, Clone, PartialEq)]
pub enum RetainedValue {
Output {
artifact: ArtifactId,
output: RawOutput,
},
Reason(Unresolved),
}
#[derive(Debug, Clone, PartialEq)]
pub struct RetainedSupply {
pub step: String,
pub instance: Vec<String>,
pub request: RequestId,
pub target: u32,
pub value: RetainedValue,
}
#[derive(Debug, Clone)]
pub struct Reproduced {
pub(crate) decision_id: String,
pub(crate) scope: Scope,
pub(crate) plan: Plan,
pub(crate) calibrations: Vec<CalibrationArtifact>,
pub(crate) snapshot: Snapshot,
pub(crate) evaluation_time: Timestamp,
pub(crate) judgment: Judgment,
pub(crate) run: RunRecord,
pub(crate) supplies: Vec<RetainedSupply>,
pub(crate) pages: Vec<(rustev_contract::ids::AcquisitionRequestId, PageValue)>,
pub(crate) text_result: Option<TextResult>,
}
impl Reproduced {
pub fn text_result(&self) -> Option<&TextResult> {
self.text_result.as_ref()
}
pub fn decision_id(&self) -> &str {
&self.decision_id
}
pub fn scope(&self) -> &Scope {
&self.scope
}
pub fn plan(&self) -> &Plan {
&self.plan
}
pub fn calibrations(&self) -> &[CalibrationArtifact] {
&self.calibrations
}
pub fn snapshot(&self) -> &Snapshot {
&self.snapshot
}
pub fn evaluation_time(&self) -> Timestamp {
self.evaluation_time
}
pub fn judgment(&self) -> &Judgment {
&self.judgment
}
pub fn run(&self) -> &RunRecord {
&self.run
}
pub fn supplies(&self) -> &[RetainedSupply] {
&self.supplies
}
}
#[derive(Debug, Clone)]
pub enum ReplayOutcome {
Reproduced(Box<Reproduced>),
Diverged {
expected: Box<Judgment>,
actual: Box<Judgment>,
},
Incomparable(Incomparable),
}
fn inconsistent(location: ItemLocation, what: Inconsistency) -> Incomparable {
Incomparable::Inconsistent { location, what }
}
fn item_failure(location: ItemLocation, f: ItemFailure) -> Incomparable {
match f {
ItemFailure::Unavailable(availability) => Incomparable::Unavailable {
location,
availability,
},
ItemFailure::Oversized => Incomparable::Oversized { location },
}
}
pub fn reproduce(
bundle: &ReplayBundle,
resolver: &dyn Resolver,
config: &ReplayConfig,
) -> ReplayOutcome {
match reproduce_inner(bundle, resolver, config) {
Ok(o) => o,
Err(i) => ReplayOutcome::Incomparable(i),
}
}
struct Resolved {
plan: Plan,
descriptors: Vec<BackendDescriptor>,
calibrations: Vec<CalibrationArtifact>,
snapshot: Snapshot,
run: RunRecord,
values: Vec<Value>,
sources: Vec<rustev_contract::source::SourceDescriptor>,
pages: Vec<PageValue>,
}
enum Value {
Output(BackendOutputDoc),
Reason(RuntimeReasonDoc),
}
fn resolve_all(
bundle: &ReplayBundle,
resolver: &dyn Resolver,
now: Timestamp,
) -> Result<Resolved, Incomparable> {
let mut budget = Budget::default();
let mut get = |location: ItemLocation, item| {
item_bytes(item, now, resolver, &mut budget)
.map_err(|f| item_failure(location, f))
.map(|b| (location, item, b))
};
fn typed<T: Document + PartialEq>(
(location, item, bytes): (
ItemLocation,
&rustev_contract::retention::RetentionItem,
Vec<u8>,
),
) -> Result<T, Incomparable> {
check_bytes::<T>(item, &bytes).map_err(|f| item_failure(location, f))
}
let plan: Plan = typed(get(ItemLocation::Plan, &bundle.plan)?)?;
let mut descriptors = vec![];
for (i, d) in bundle.descriptors.iter().enumerate() {
descriptors.push(typed(get(ItemLocation::Descriptor(i), d)?)?);
}
let mut calibrations = vec![];
for (i, c) in bundle.calibrations.iter().enumerate() {
calibrations.push(typed(get(ItemLocation::Calibration(i), c)?)?);
}
let snapshot: Snapshot = typed(get(ItemLocation::Snapshot, &bundle.snapshot)?)?;
let run: RunRecord = typed(get(ItemLocation::Run, &bundle.run)?)?;
let mut values = vec![];
for (i, s) in bundle.supplies.iter().enumerate() {
let location = ItemLocation::Supply(i);
let got = get(location, &s.value)?;
values.push(if s.value.document_schema == schema::RUNTIME_REASON {
let r: RuntimeReasonDoc = typed(got)?;
if r.check().is_err() {
return Err(Incomparable::Unavailable {
location,
availability: Availability::Mismatched,
});
}
Value::Reason(r)
} else {
Value::Output(typed(got)?)
});
}
let mut sources = vec![];
for (i, source) in bundle.sources.iter().flatten().enumerate() {
sources.push(typed(get(ItemLocation::Source(i), source)?)?);
}
let mut pages = vec![];
for (i, page) in bundle.acquisition_pages.iter().flatten().enumerate() {
let got = get(ItemLocation::AcquisitionPage(i), &page.value)?;
pages.push(if page.value.document_schema == schema::RUNTIME_REASON {
PageValue::Reason(typed::<RuntimeReasonDoc>(got)?.reason)
} else {
PageValue::Page(typed(got)?)
});
}
Ok(Resolved {
sources,
pages,
plan,
descriptors,
calibrations,
snapshot,
run,
values,
})
}
fn target_artifact(plan: &Plan, step: &str, target: u32) -> Option<ArtifactId> {
if let Some((_, b)) = rustev_contract::generation::output_screen_binding(plan, step) {
return (target == 0).then(|| b.artifact.clone());
}
let artifact = match &plan.steps.iter().find(|s| s.id == step)?.detail {
PlanStepDetail::Semantic(b) => &b.artifact,
PlanStepDetail::Generate(b) => &b.artifact,
PlanStepDetail::Extract(b) => &b.artifact,
_ => return None,
};
if target == 0 {
return Some(artifact.clone());
}
match &plan.execution {
PlanExecution::Declared(d) => d
.fallbacks
.iter()
.find(|f| f.step == step && f.target == target)
.map(|f| f.artifact.clone()),
PlanExecution::None => None,
}
}
enum ReplayEvaluation<'a> {
Decision(Box<Evaluation<'a>>),
Generation(Box<GenerationEvaluation<'a>>),
}
impl<'a> ReplayEvaluation<'a> {
fn advance(self, decision: &str) -> Result<Self, Incomparable> {
match self {
Self::Decision(ev)
if ev.pending().is_empty() && ev.pending_acquisitions().is_empty() =>
{
ev.begin_generation(decision)
.map(Box::new)
.map(Self::Generation)
.map_err(|_| inconsistent(ItemLocation::Bundle, Inconsistency::FinishRefused))
}
other => Ok(other),
}
}
fn pending(&self) -> Vec<(String, Vec<String>)> {
match self {
Self::Decision(ev) => ev
.pending()
.iter()
.map(|r| (r.step.clone(), r.instance.clone()))
.collect(),
Self::Generation(g) => g
.pending()
.iter()
.map(|r| (r.document.step.clone(), r.document.instance.clone()))
.collect(),
}
}
fn request_document(
&self,
step: &str,
instance: &[String],
target: u32,
scope: &Scope,
) -> Result<rustev_contract::request::RequestDoc, rustev_core::RequestIdentityError> {
match self {
Self::Decision(ev) => ev.request_document(step, instance, target, scope),
Self::Generation(g) => g.request_document(step, instance, target, scope),
}
}
fn accepts(&self, step: &str, instance: &[String], output: &RawOutput) -> bool {
match self {
Self::Decision(ev) => matches!(ev.check_output(step, instance, output), Ok(Ok(()))),
Self::Generation(g) => matches!(g.check_output(step, instance, output), Ok(Ok(()))),
}
}
fn supply(
&mut self,
step: &str,
instance: &[String],
value: Supplied,
q: &rustev_contract::run::RequestRecord,
decision: &str,
) -> bool {
match self {
Self::Decision(ev) => ev.supply(step, instance, value).is_ok(),
Self::Generation(g) => {
let from = q.supplied_from().expect("checked supplied request");
let attempt = from
.attempt
.map(|a| a.attempt_id.as_str())
.or(from.reused_from);
let fallback = format!("{decision}/{step}/refused");
let attempt = attempt.unwrap_or(&fallback);
if g.supply_attempt(
step,
instance,
value,
attempt,
(q.attempts.len() as u32).max(1),
)
.is_err()
{
return false;
}
if let Some(source) = from.reused_from {
let decision = source
.split('/')
.next()
.unwrap_or("")
.replace("%2F", "/")
.replace("%2C", ",")
.replace("%25", "%");
g.note_cached_attempt(attempt, &decision);
}
true
}
}
}
}
fn list_new(
ev: &ReplayEvaluation<'_>,
listed: &mut Vec<(String, Vec<String>)>,
seen: &mut BTreeSet<(String, Vec<String>)>,
) {
for key in ev.pending() {
if seen.insert(key.clone()) {
listed.push(key);
}
}
}
fn reproduce_inner(
bundle: &ReplayBundle,
resolver: &dyn Resolver,
config: &ReplayConfig,
) -> Result<ReplayOutcome, Incomparable> {
if config.scope != bundle.scope {
return Err(Incomparable::ScopeMismatch);
}
bundle
.check(config.host_cap_ms)
.map_err(Incomparable::InvalidBundle)?;
if bundle.capture != CaptureStatus::Complete {
return Err(Incomparable::CaptureIncomplete(bundle.capture));
}
let r = resolve_all(bundle, resolver, config.now_ms)?;
let run = &r.run;
let run_at = ItemLocation::Run;
if run.decision_id != bundle.decision_id {
return Err(inconsistent(run_at, Inconsistency::DecisionId));
}
if run.plan_id != bundle.plan_id {
return Err(inconsistent(run_at, Inconsistency::PlanId));
}
let evidence = match (&run.termination, &run.core) {
(Termination::Judged, CoreEvidence::Recorded(e)) => e,
_ => return Err(Incomparable::NoExpectedJudgment),
};
if evidence.decision_id != bundle.decision_id {
return Err(inconsistent(run_at, Inconsistency::DecisionId));
}
if evidence.plan_id != bundle.plan_id || evidence.judgment.plan_id != bundle.plan_id {
return Err(inconsistent(run_at, Inconsistency::PlanId));
}
if evidence.snapshot_id != bundle.snapshot_id {
return Err(inconsistent(run_at, Inconsistency::SnapshotId));
}
if evidence.evaluation_time_ms != bundle.evaluation_time_ms {
return Err(inconsistent(run_at, Inconsistency::EvaluationTime));
}
if r.plan.id().ok().as_ref() != Some(&bundle.plan_id) {
return Err(inconsistent(ItemLocation::Plan, Inconsistency::PlanId));
}
if r.snapshot.id().ok().as_ref() != Some(&bundle.snapshot_id) {
return Err(inconsistent(
ItemLocation::Snapshot,
Inconsistency::SnapshotId,
));
}
let max = r
.plan
.definition
.limits
.max_semantic_requests
.min(MAX_BUNDLE_ITEMS as u64);
if bundle.supplies.len() as u64 > max {
return Err(inconsistent(
ItemLocation::Bundle,
Inconsistency::TooManySupplies { max },
));
}
let max_pages = r
.plan
.definition
.limits
.acquisition
.as_ref()
.map_or(0, |l| l.max_calls);
if r.pages.len() as u64 > max_pages {
return Err(inconsistent(
ItemLocation::Bundle,
Inconsistency::TooManySupplies { max: max_pages },
));
}
let compiled =
Compiled::load_checked_with_sources(&r.plan, &r.descriptors, &r.calibrations, &r.sources)
.map_err(|e| match e {
LoadError::CompilerChanged {
recorded_compiler,
current_compiler,
recorded_registry,
current_registry,
} => Incomparable::CompilerChanged {
recorded_compiler,
current_compiler: current_compiler.into(),
recorded_registry,
current_registry: current_registry.into(),
},
e @ (LoadError::MissingDescriptor { .. } | LoadError::MissingCalibration { .. }) => {
Incomparable::MissingDependency {
detail: e.to_string(),
}
}
LoadError::PlanMismatch { .. } => Incomparable::PlanMismatch,
})?;
let ev = Evaluation::start(&compiled, &r.snapshot, bundle.evaluation_time_ms)
.map_err(|_| inconsistent(ItemLocation::Snapshot, Inconsistency::StartRefused))?;
let mut ev = ReplayEvaluation::Decision(Box::new(ev)).advance(&bundle.decision_id)?;
let by_key: BTreeMap<(&str, &[String]), usize> = run
.requests
.iter()
.enumerate()
.map(|(i, q)| ((q.step.as_str(), q.instance.as_slice()), i))
.collect();
let mut listed = vec![];
let mut seen = BTreeSet::new();
list_new(&ev, &mut listed, &mut seen);
let mut supplies = vec![];
let pages = bundle.acquisition_pages.as_deref().unwrap_or(&[]);
let mut missing = vec![];
let (mut semantic_index, mut page_index) = (0, 0);
loop {
if let ReplayEvaluation::Decision(decision) = &mut ev {
fill_missing(decision, &pages[page_index..], &mut missing)?;
}
ev = ev.advance(&bundle.decision_id)?;
list_new(&ev, &mut listed, &mut seen);
let semantic = bundle.supplies.get(semantic_index);
let page = pages.get(page_index);
if let Some(page) = page
&& semantic.is_none_or(|s| page.sequence < s.sequence)
{
let ReplayEvaluation::Decision(decision) = &mut ev else {
if !missing.is_empty() {
page_index += 1;
continue;
}
return Err(inconsistent(
ItemLocation::AcquisitionPage(page_index),
Inconsistency::NotPending,
));
};
if missing.is_empty()
|| decision.pending_acquisitions().iter().any(|p| {
p.step == page.step && p.tier_index == page.tier && p.page_index == page.page
})
{
supply_page(decision, page, &r.pages[page_index], run, page_index)?;
}
page_index += 1;
continue;
}
let Some(entry) = semantic else {
break;
};
let i = semantic_index;
let value = &r.values[i];
semantic_index += 1;
if !missing.is_empty()
&& !ev
.pending()
.iter()
.any(|p| p.0 == entry.step && p.1 == entry.instance)
{
continue;
}
let at = ItemLocation::Supply(i);
let bad = |what| Err(inconsistent(at, what));
let Some(q) = by_key
.get(&(entry.step.as_str(), entry.instance.as_slice()))
.map(|&k| &run.requests[k])
else {
return bad(Inconsistency::NotInRun);
};
let Some(from) = q.supplied_from() else {
return bad(Inconsistency::NotInRun);
};
let origin = from
.attempt
.map(|a| a.attempt_id.as_str())
.or(from.reused_from);
let output_target = from
.attempt
.map(|a| a.target)
.or(from.reused_from.map(|_| from.target));
if from.target != entry.target
|| origin != entry.attempt_id.as_deref()
|| matches!(q.result, RequestResult::Output { .. })
&& output_target != Some(entry.target)
{
return bad(Inconsistency::Origin);
}
let doc =
match ev.request_document(&entry.step, &entry.instance, entry.target, &bundle.scope) {
Ok(d) => d,
Err(rustev_core::RequestIdentityError::UnknownTarget { .. }) => {
return bad(Inconsistency::UnknownTarget);
}
Err(_) => return bad(Inconsistency::NotPending),
};
if doc.id().ok().as_ref() != Some(&entry.request) {
return bad(Inconsistency::RequestIdentity);
}
let (supplied, retained) = match (value, &q.result) {
(Value::Output(o), RequestResult::Output { target }) if *target == entry.target => {
let bound = target_artifact(&r.plan, &entry.step, entry.target);
if o.step != entry.step
|| o.instance != entry.instance
|| bound.as_ref() != Some(&o.artifact)
|| from.attempt.is_some_and(|a| a.artifact != o.artifact)
{
return bad(Inconsistency::OutputBinding);
}
if !ev.accepts(&entry.step, &entry.instance, &o.output) {
return bad(Inconsistency::OutputRefused);
}
(
Supplied::Output(o.output.clone()),
RetainedValue::Output {
artifact: o.artifact.clone(),
output: o.output.clone(),
},
)
}
(Value::Reason(d), RequestResult::Failed(u)) if &d.reason == u => (
Supplied::Failed(d.reason.clone()),
RetainedValue::Reason(d.reason.clone()),
),
_ => return bad(Inconsistency::Value),
};
if let Some(context) = entry.failure_context
&& let ReplayEvaluation::Decision(decision) = &mut ev
{
decision.note_failure(
&entry.step,
&entry.instance,
rustev_core::FailureContext {
origin: context.origin,
determinism: context.determinism,
},
);
}
if !ev.supply(
&entry.step,
&entry.instance,
supplied,
q,
&bundle.decision_id,
) {
return bad(Inconsistency::NotPending);
}
ev = ev.advance(&bundle.decision_id)?;
list_new(&ev, &mut listed, &mut seen);
supplies.push(RetainedSupply {
step: entry.step.clone(),
instance: entry.instance.clone(),
request: entry.request.clone(),
target: entry.target,
value: retained,
});
}
if let Some(p) = ev.pending().first() {
return Err(inconsistent(
ItemLocation::Bundle,
Inconsistency::MissingSupply {
step: p.0.clone(),
instance: p.1.clone(),
},
));
}
let recorded: Vec<(String, Vec<String>)> = run
.requests
.iter()
.map(|q| (q.step.clone(), q.instance.clone()))
.collect();
if missing.is_empty() && recorded != listed {
return Err(inconsistent(run_at, Inconsistency::RequestOrder));
}
let ReplayEvaluation::Generation(stage) = ev else {
return Err(inconsistent(
ItemLocation::Bundle,
Inconsistency::FinishRefused,
));
};
let (judgment, actual_evidence, text_result) = stage
.finish()
.map_err(|_| inconsistent(ItemLocation::Bundle, Inconsistency::FinishRefused))?;
if !missing.is_empty() {
return Err(Incomparable::AcquisitionNotCaptured {
missing,
judgment: Box::new(judgment),
});
}
let expected = &evidence.judgment;
let same = match (judgment.record_canonical(), expected.record_canonical()) {
(Ok(a), Ok(b)) => a == b,
_ => {
return Err(inconsistent(ItemLocation::Run, Inconsistency::NoRecordForm));
}
};
let generation_bytes =
|g: &Option<Vec<rustev_contract::generation_result::GenerationEvidence>>| {
let mut g = g.clone();
if let Some(steps) = &mut g {
for step in steps {
step.notices.retain(|n| {
matches!(
n.kind,
rustev_contract::generation_result::TextNoticeKind::BackstopTruncated
)
});
}
}
rustev_contract::canonical::record_canonical_bytes(&g)
};
let generation_same = match (
generation_bytes(&actual_evidence.generation),
generation_bytes(&evidence.generation),
) {
(Ok(a), Ok(b)) => a == b,
_ => return Err(inconsistent(ItemLocation::Run, Inconsistency::NoRecordForm)),
};
if !same || !generation_same || actual_evidence.screening != evidence.screening {
return Ok(ReplayOutcome::Diverged {
expected: Box::new(expected.clone()),
actual: Box::new(judgment),
});
}
Ok(ReplayOutcome::Reproduced(Box::new(Reproduced {
decision_id: bundle.decision_id.clone(),
scope: bundle.scope.clone(),
plan: r.plan.clone(),
calibrations: r.calibrations.clone(),
snapshot: r.snapshot,
evaluation_time: bundle.evaluation_time_ms,
judgment,
run: r.run.clone(),
supplies,
pages: pages
.iter()
.zip(&r.pages)
.map(|(p, v)| (p.request.clone(), v.clone()))
.collect(),
text_result,
})))
}