use std::collections::BTreeMap;
use std::fmt;
use rustev_contract::canonical::{CanonicalError, canonical_bytes};
use rustev_contract::decimal::Decimal;
use rustev_contract::dynamic::{
AbstentionDecision, AbstentionPolicy, AbstentionRule, DynamicAnswer, DynamicCapability,
DynamicIdentity, DynamicLineage, DynamicOperation, DynamicOutputKind, DynamicProjection,
DynamicProposal, DynamicQuestion, DynamicRawOutput, DynamicRefusal, DynamicUnresolved,
DynamicValidation, HiddenState, LabelMass, MAX_CLASSIFY_LABELS, MAX_RUBRIC_LEVELS, MIN_LABELS,
PROPOSITION_LABELS, TaskFamily,
};
use rustev_contract::ids::DynamicRequestId;
use rustev_contract::judgment::Derivation;
use rustev_contract::remote::{Disclosure, ServedIdentity};
use rustev_contract::run::{CostBound, CostModel};
use rustev_contract::{Document, DocumentError};
use crate::kinds::Distribution;
pub const DISTRIBUTION_TOLERANCE: &str = "0.000001";
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum DynamicRequestError {
Parse(DocumentError),
Bounds(String),
PropositionLabels,
LabelCount {
operation: DynamicOperation,
count: usize,
max: usize,
},
OutputKind {
operation: DynamicOperation,
output: DynamicOutputKind,
},
Abstention(String),
ZeroDeadline,
ZeroCostCap,
TaskFamilyNotAllowed(TaskFamily),
}
impl fmt::Display for DynamicRequestError {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
DynamicRequestError::Parse(e) => write!(f, "{e}"),
DynamicRequestError::Bounds(m) => f.write_str(m),
DynamicRequestError::PropositionLabels => {
f.write_str("proposition takes exactly the labels [\"false\", \"true\"]")
}
DynamicRequestError::LabelCount {
operation,
count,
max,
} => write!(
f,
"{} takes {MIN_LABELS} to {max} labels, not {count}",
operation.name()
),
DynamicRequestError::OutputKind { operation, output } => {
write!(f, "{} does not produce {}", operation.name(), output.name())
}
DynamicRequestError::Abstention(m) => write!(f, "abstention: {m}"),
DynamicRequestError::ZeroDeadline => f.write_str("the deadline must be positive"),
DynamicRequestError::ZeroCostCap => f.write_str("the hard cost cap must be positive"),
DynamicRequestError::TaskFamilyNotAllowed(t) => write!(
f,
"task family {:?} version {:?} is not allowed",
t.name, t.version
),
}
}
}
impl std::error::Error for DynamicRequestError {}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ValidatedQuestion {
q: DynamicQuestion,
}
impl ValidatedQuestion {
pub fn question(&self) -> &DynamicQuestion {
&self.q
}
pub fn into_question(self) -> DynamicQuestion {
self.q
}
}
pub fn parse_question(
bytes: &[u8],
allowlist: &[TaskFamily],
) -> Result<ValidatedQuestion, DynamicRequestError> {
let q = DynamicQuestion::parse(bytes).map_err(DynamicRequestError::Parse)?;
validate(q, allowlist)
}
pub fn validate(
q: DynamicQuestion,
allowlist: &[TaskFamily],
) -> Result<ValidatedQuestion, DynamicRequestError> {
if q.schema != rustev_contract::schema::DYNAMIC_QUESTION {
return Err(DynamicRequestError::Parse(DocumentError::Schema {
expected: rustev_contract::schema::DYNAMIC_QUESTION,
found: q.schema.chars().take(80).collect(),
}));
}
q.check_bounds()
.map_err(|e| DynamicRequestError::Bounds(e.0))?;
match q.operation {
DynamicOperation::Proposition => {
if q.labels.len() != PROPOSITION_LABELS.len()
|| q.labels.iter().zip(PROPOSITION_LABELS).any(|(l, p)| l != p)
{
return Err(DynamicRequestError::PropositionLabels);
}
}
op @ (DynamicOperation::Classify | DynamicOperation::Rubric) => {
let max = if op == DynamicOperation::Classify {
MAX_CLASSIFY_LABELS
} else {
MAX_RUBRIC_LEVELS
};
if q.labels.len() < MIN_LABELS || q.labels.len() > max {
return Err(DynamicRequestError::LabelCount {
operation: op,
count: q.labels.len(),
max,
});
}
}
}
let allowed = match q.operation {
DynamicOperation::Proposition | DynamicOperation::Classify => {
matches!(
q.output,
DynamicOutputKind::Label | DynamicOutputKind::Distribution
)
}
DynamicOperation::Rubric => matches!(
q.output,
DynamicOutputKind::Label | DynamicOutputKind::OrdinalDistribution
),
};
if !allowed {
return Err(DynamicRequestError::OutputKind {
operation: q.operation,
output: q.output,
});
}
match (&q.abstention, q.output) {
(AbstentionPolicy::LabelOnly, DynamicOutputKind::Label) => {}
(AbstentionPolicy::LabelOnly, _) => {
return Err(DynamicRequestError::Abstention(
"label-only posture with a distribution output".into(),
));
}
(AbstentionPolicy::Thresholds { .. }, DynamicOutputKind::Label) => {
return Err(DynamicRequestError::Abstention(
"a label output is accepted only under an explicit label-only posture".into(),
));
}
(
AbstentionPolicy::Thresholds {
min_top_mass,
min_margin,
},
_,
) => {
let zero = Decimal::parse("0").expect("literal");
let one = Decimal::parse("1").expect("literal");
for (name, t) in [("min_top_mass", min_top_mass), ("min_margin", min_margin)] {
if *t < zero || *t > one {
return Err(DynamicRequestError::Abstention(format!(
"{name} {t} is outside [0, 1]"
)));
}
}
}
}
if q.deadline_ms == 0 {
return Err(DynamicRequestError::ZeroDeadline);
}
if q.max_cost_units == 0 {
return Err(DynamicRequestError::ZeroCostCap);
}
if !allowlist.contains(&q.task_family) {
return Err(DynamicRequestError::TaskFamilyNotAllowed(
q.task_family.clone(),
));
}
Ok(ValidatedQuestion { q })
}
pub fn request_identity(
q: &ValidatedQuestion,
capability: &DynamicCapability,
) -> Result<DynamicRequestId, CanonicalError> {
DynamicIdentity::of(&q.q, capability).id()
}
pub fn projection(q: &ValidatedQuestion) -> Vec<u8> {
canonical_bytes(&DynamicProjection::of(&q.q)).expect("a validated request has a canonical form")
}
fn exceeds(limit: &str, declared: u64, requested: usize) -> Result<(), DynamicRefusal> {
let requested = requested as u64;
if requested > declared {
return Err(DynamicRefusal::ExceedsDeclaredLimit {
limit: limit.into(),
declared,
requested,
});
}
Ok(())
}
pub fn admit<'c>(
q: &ValidatedQuestion,
capability: Option<&'c DynamicCapability>,
) -> Result<&'c DynamicCapability, DynamicRefusal> {
let q = &q.q;
let c = capability.ok_or(DynamicRefusal::NoDynamicCapability)?;
if c.schema != rustev_contract::schema::DYNAMIC_CAPABILITY {
return Err(DynamicRefusal::CapabilityInvalid {
detail: "not a rustev.dynamic-capability/1 declaration".into(),
});
}
c.check_bounds()
.map_err(|e| DynamicRefusal::CapabilityInvalid { detail: e.0 })?;
if c.artifact != q.artifact {
return Err(DynamicRefusal::ArtifactMismatch {
declared: c.artifact.clone(),
});
}
let support = c
.support(q.operation)
.filter(|s| s.outputs.contains(&q.output))
.ok_or(DynamicRefusal::OperationUnsupported {
operation: q.operation,
output: q.output,
})?;
if !c.task_families.contains(&q.task_family) {
return Err(DynamicRefusal::TaskFamilyUndeclared);
}
exceeds("labels", support.max_labels, q.labels.len())?;
exceeds("question_bytes", c.max_question_bytes, q.question.len())?;
exceeds("context_fields", c.max_context_fields, q.context.len())?;
exceeds(
"context_bytes",
c.max_context_bytes,
q.context_value_bytes(),
)?;
if let Some(p) = q.privacy.iter().find(|p| !c.privacy.contains(p)) {
return Err(DynamicRefusal::PrivacyUnsupported { option: p.clone() });
}
if c.cost != CostModel::Bounded {
return Err(DynamicRefusal::CostUnenforceable { model: c.cost });
}
Ok(c)
}
pub fn admit_cost(bound: CostBound, cap: u64) -> Result<u64, DynamicRefusal> {
match bound {
CostBound::Bounded { max_units } if max_units <= cap => Ok(max_units),
CostBound::Bounded { max_units } => Err(DynamicRefusal::CostAboveCap {
bound: max_units,
cap,
}),
CostBound::Estimated { .. } | CostBound::Unknown => Err(DynamicRefusal::CostUnknown),
}
}
pub fn qualifiable(capability: &DynamicCapability) -> bool {
capability.hidden_state == HiddenState::None
}
#[derive(Debug, Clone, PartialEq)]
pub struct Judged {
pub validation: DynamicValidation,
pub abstention: AbstentionDecision,
pub result: Result<DynamicAnswer, DynamicUnresolved>,
}
fn invalid(detail: String) -> Judged {
Judged {
validation: DynamicValidation::Invalid {
detail: detail.clone(),
},
abstention: AbstentionDecision::NotApplied,
result: Err(DynamicUnresolved::InvalidOutput { detail }),
}
}
fn short(s: &str) -> String {
s.chars().take(64).collect()
}
pub fn judge(
q: &ValidatedQuestion,
capability: &DynamicCapability,
served: &ServedIdentity,
raw: &DynamicRawOutput,
) -> Judged {
let q = &q.q;
if capability.served.disclosure == Disclosure::Pinned
&& (served.disclosure != Disclosure::Pinned
|| served.identity != capability.served.identity)
{
let pinned = capability.served.identity.clone().unwrap_or_default();
return Judged {
validation: DynamicValidation::Invalid {
detail: "served identity differs from the pinned identity".into(),
},
abstention: AbstentionDecision::NotApplied,
result: Err(DynamicUnresolved::ServedIdentityMismatch {
pinned,
served: served.identity.as_ref().map(|s| short(s)),
}),
};
}
match raw {
DynamicRawOutput::Unrecognized { shape, .. } => invalid(format!(
"response material of shape {} is not an answer; it was not acted on",
shape.name()
)),
DynamicRawOutput::Label(l) => {
if q.output != DynamicOutputKind::Label {
return invalid(format!(
"a label where {} is required, and label-only posture is not permitted",
q.output.name()
));
}
if !q.labels.contains(l) {
return invalid(format!("undeclared label {:?}", short(l)));
}
Judged {
validation: DynamicValidation::Valid,
abstention: AbstentionDecision::LabelOnly,
result: Ok(DynamicAnswer::Label { label: l.clone() }),
}
}
DynamicRawOutput::Distribution(map) => judge_distribution(q, map),
}
}
fn judge_distribution(q: &DynamicQuestion, map: &BTreeMap<String, f64>) -> Judged {
if q.output == DynamicOutputKind::Label {
return invalid("a distribution where a label is required".into());
}
let tolerance = Decimal::parse(DISTRIBUTION_TOLERANCE).expect("literal");
let d = match Distribution::new(&q.labels, map, tolerance) {
Ok(d) => d,
Err(e) => {
let detail: String = e.0.chars().take(200).collect();
return invalid(format!("outside the declared support: {detail}"));
}
};
let (top_label, top_mass) = d.top();
let top_label = top_label.to_string();
let second = d
.options()
.iter()
.zip(d.masses())
.filter(|(o, _)| **o != top_label)
.map(|(_, m)| *m)
.fold(0.0_f64, f64::max);
let margin = top_mass - second;
let AbstentionPolicy::Thresholds {
min_top_mass,
min_margin,
} = &q.abstention
else {
return invalid("no thresholds for a distribution output".into());
};
let masses: Vec<LabelMass> = d
.options()
.iter()
.zip(d.masses())
.map(|(l, m)| LabelMass {
label: l.clone(),
mass: *m,
})
.collect();
let rule = if top_mass < min_top_mass.to_f64_nearest() {
Some(AbstentionRule::MinTopMass)
} else if margin < min_margin.to_f64_nearest() {
Some(AbstentionRule::MinMargin)
} else {
None
};
if let Some(rule) = rule {
return Judged {
validation: DynamicValidation::Valid,
abstention: AbstentionDecision::Abstained {
rule,
top_mass,
margin,
},
result: Err(DynamicUnresolved::Abstained { rule }),
};
}
let answer = if q.output == DynamicOutputKind::OrdinalDistribution {
DynamicAnswer::OrdinalDistribution {
level: top_label,
masses,
}
} else {
DynamicAnswer::UncalibratedDistribution {
label: top_label,
masses,
}
};
Judged {
validation: DynamicValidation::Valid,
abstention: AbstentionDecision::Passed { top_mass, margin },
result: Ok(answer),
}
}
pub fn proposal(
q: &ValidatedQuestion,
request: DynamicRequestId,
capability: &DynamicCapability,
attempt_id: &str,
answer: DynamicAnswer,
) -> DynamicProposal {
let mut exact_processing = vec!["validate".to_string()];
if !matches!(answer, DynamicAnswer::Label { .. }) {
exact_processing.push("abstention".into());
}
DynamicProposal {
answer,
derivation: Derivation::ModelDerived,
lineage: DynamicLineage {
request,
context: q.q.context.iter().map(|c| c.name.clone()).collect(),
backend_id: capability.backend_id.clone(),
artifact: capability.artifact.clone(),
attempt_id: attempt_id.to_string(),
exact_processing,
},
}
}