1use super::CONFIDENCE_ASSESSMENT_CONTRACT_VERSION;
4use super::CONFIDENCE_ASSESSMENT_SCHEMA;
5use super::CONFIDENCE_INPUT_CONTRACT_VERSION;
6use super::CONFIDENCE_INPUT_SCHEMA;
7use super::KnowledgeError;
8use super::canonical_optional_f64;
9use super::f64_column;
10use super::fixed_column;
11use super::invalid;
12use super::normalize_zero;
13use super::optional_f64;
14use super::require_schema;
15use super::require_uuid;
16use super::require_v7;
17use super::required_i64;
18use super::required_text;
19use super::required_u32;
20use super::string_column;
21use super::timestamp_column;
22use super::u32_column;
23use super::uuid_at;
24use super::validate_confidence;
25use arrow::array::FixedSizeBinaryBuilder;
26use arrow::array::Float64Array;
27use arrow::array::StringArray;
28use arrow::array::TimestampMicrosecondArray;
29use arrow::array::UInt32Array;
30use arrow::record_batch::RecordBatch;
31use graphforge_core::canonical::CANONICAL_CONTRACT_VERSION;
32use graphforge_core::canonical::CanonicalDomain;
33use graphforge_core::canonical::CanonicalWriter;
34use graphforge_core::canonical::fingerprint;
35use std::collections::HashMap;
36use std::collections::HashSet;
37use std::sync::Arc;
38use uuid::Uuid;
39
40#[cfg(test)]
41thread_local! {
42 static CONFIDENCE_ROW_VALIDATION_COUNT: std::cell::Cell<usize> = const { std::cell::Cell::new(0) };
43}
44
45#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, serde::Serialize, serde::Deserialize)]
47#[serde(rename_all = "snake_case")]
48pub enum ConfidencePolicy {
49 Explicit,
51 ConservativeMin,
53}
54
55impl ConfidencePolicy {
56 #[must_use]
58 pub const fn as_str(self) -> &'static str {
59 match self {
60 Self::Explicit => "explicit",
61 Self::ConservativeMin => "conservative_min",
62 }
63 }
64
65 pub(super) fn parse(value: &str) -> Result<Self, KnowledgeError> {
66 match value {
67 "explicit" => Ok(Self::Explicit),
68 "conservative_min" => Ok(Self::ConservativeMin),
69 _ => Err(invalid("policy", "unknown closed value")),
70 }
71 }
72}
73
74#[derive(Clone, Debug, PartialEq)]
76pub struct ConfidenceAssessment {
77 pub confidence_uuid: Uuid,
79 pub assertion_uuid: Uuid,
81 pub policy: ConfidencePolicy,
83 pub policy_version: u32,
85 pub value: Option<f64>,
87 pub provenance_uuid: Uuid,
89 pub recorded_at_micros: i64,
91 pub contract_version: u32,
93}
94
95impl ConfidenceAssessment {
96 pub fn new(
98 confidence_uuid: Uuid,
99 assertion_uuid: Uuid,
100 policy: ConfidencePolicy,
101 value: Option<f64>,
102 provenance_uuid: Uuid,
103 recorded_at_micros: i64,
104 ) -> Result<Self, KnowledgeError> {
105 require_v7(confidence_uuid, "confidence_uuid")?;
106 require_v7(assertion_uuid, "assertion_uuid")?;
107 require_uuid(provenance_uuid, "provenance_uuid")?;
108 validate_confidence(value, "value")?;
109 Ok(Self {
110 confidence_uuid,
111 assertion_uuid,
112 policy,
113 policy_version: 1,
114 value: value.map(normalize_zero),
115 provenance_uuid,
116 recorded_at_micros,
117 contract_version: CONFIDENCE_ASSESSMENT_CONTRACT_VERSION,
118 })
119 }
120}
121
122#[derive(Clone, Debug, PartialEq)]
124pub struct ConfidenceInput {
125 pub confidence_uuid: Uuid,
127 pub input_confidence_uuid: Uuid,
129 pub input_value: Option<f64>,
131 pub ordinal: u32,
133 pub contract_version: u32,
135}
136
137impl ConfidenceInput {
138 pub fn new(
140 confidence_uuid: Uuid,
141 input_confidence_uuid: Uuid,
142 input_value: Option<f64>,
143 ordinal: u32,
144 ) -> Result<Self, KnowledgeError> {
145 require_v7(confidence_uuid, "confidence_uuid")?;
146 require_v7(input_confidence_uuid, "input_confidence_uuid")?;
147 validate_confidence(input_value, "input_value")?;
148 Ok(Self {
149 confidence_uuid,
150 input_confidence_uuid,
151 input_value: input_value.map(normalize_zero),
152 ordinal,
153 contract_version: CONFIDENCE_INPUT_CONTRACT_VERSION,
154 })
155 }
156}
157
158#[derive(Clone, Debug, Default, PartialEq)]
170pub struct ConfidenceLedger {
171 assessments: Vec<ConfidenceAssessment>,
173 inputs: Vec<ConfidenceInput>,
175}
176
177impl ConfidenceLedger {
178 #[must_use]
180 pub fn assessments(&self) -> &[ConfidenceAssessment] {
181 &self.assessments
182 }
183
184 #[must_use]
186 pub fn inputs(&self) -> &[ConfidenceInput] {
187 &self.inputs
188 }
189
190 #[must_use]
192 pub fn into_parts(self) -> (Vec<ConfidenceAssessment>, Vec<ConfidenceInput>) {
193 (self.assessments, self.inputs)
194 }
195
196 pub fn new(
198 mut assessments: Vec<ConfidenceAssessment>,
199 mut inputs: Vec<ConfidenceInput>,
200 ) -> Result<Self, KnowledgeError> {
201 inputs.sort_by_key(|row| (row.confidence_uuid, row.ordinal, row.input_confidence_uuid));
202 validate_confidence_rows(&assessments, &inputs)?;
203 sort_confidence_rows(&mut assessments, &mut inputs);
204 Ok(Self {
205 assessments,
206 inputs,
207 })
208 }
209
210 pub fn explicit(
212 confidence_uuid: Uuid,
213 assertion_uuid: Uuid,
214 value: f64,
215 provenance_uuid: Uuid,
216 recorded_at_micros: i64,
217 ) -> Result<Self, KnowledgeError> {
218 Self::new(
219 vec![ConfidenceAssessment::new(
220 confidence_uuid,
221 assertion_uuid,
222 ConfidencePolicy::Explicit,
223 Some(value),
224 provenance_uuid,
225 recorded_at_micros,
226 )?],
227 vec![],
228 )
229 }
230
231 pub fn conservative_min(
233 &self,
234 confidence_uuid: Uuid,
235 assertion_uuid: Uuid,
236 mut requested: Vec<Uuid>,
237 provenance_uuid: Uuid,
238 recorded_at_micros: i64,
239 ) -> Result<Self, KnowledgeError> {
240 requested.sort_unstable();
241 if requested.windows(2).any(|pair| pair[0] == pair[1]) {
242 return Err(KnowledgeError::Duplicate("input_confidence_uuid"));
243 }
244 let values = self
245 .assessments
246 .iter()
247 .map(|row| (row.confidence_uuid, row.value))
248 .collect::<HashMap<_, _>>();
249 let mut minimum = None;
250 let mut complete = !requested.is_empty();
251 let mut inputs = Vec::with_capacity(requested.len());
252 for (ordinal, input_uuid) in requested.into_iter().enumerate() {
253 require_v7(input_uuid, "input_confidence_uuid")?;
254 let observed = values.get(&input_uuid).copied().flatten();
255 if let Some(value) = observed {
256 minimum = Some(minimum.map_or(value, |current: f64| current.min(value)));
257 } else {
258 complete = false;
259 }
260 inputs.push(ConfidenceInput::new(
261 confidence_uuid,
262 input_uuid,
263 observed,
264 u32::try_from(ordinal).map_err(|_| KnowledgeError::Limit {
265 participant: "confidence_inputs",
266 observed: ordinal,
267 limit: u32::MAX as usize,
268 })?,
269 )?);
270 }
271 Self::new(
272 vec![ConfidenceAssessment::new(
273 confidence_uuid,
274 assertion_uuid,
275 ConfidencePolicy::ConservativeMin,
276 complete.then_some(minimum).flatten(),
277 provenance_uuid,
278 recorded_at_micros,
279 )?],
280 inputs,
281 )
282 }
283
284 pub fn merge(&self, staged: &Self) -> Result<Self, KnowledgeError> {
286 self.merge_with_limits(staged, super::MAX_KNOWLEDGE_ROWS, super::MAX_KNOWLEDGE_ROWS)
287 }
288
289 fn merge_with_limits(
290 &self,
291 staged: &Self,
292 assessment_limit: usize,
293 input_limit: usize,
294 ) -> Result<Self, KnowledgeError> {
295 let mut assessments = self.assessments.clone();
296 let mut inputs = self.inputs.clone();
297 for row in &staged.assessments {
298 if let Some(existing) = assessments
299 .iter()
300 .find(|existing| existing.confidence_uuid == row.confidence_uuid)
301 {
302 if existing != row
303 || inputs_for(&inputs, row.confidence_uuid)
304 != inputs_for(&staged.inputs, row.confidence_uuid)
305 {
306 return Err(KnowledgeError::Conflict("confidence_uuid"));
307 }
308 } else {
309 assessments.push(row.clone());
310 inputs.extend(
311 staged
312 .inputs
313 .iter()
314 .filter(|input| input.confidence_uuid == row.confidence_uuid)
315 .cloned(),
316 );
317 }
318 }
319 check_confidence_row_limit(
320 "confidence_assessments",
321 assessments.len(),
322 assessment_limit,
323 )?;
324 check_confidence_row_limit("confidence_inputs", inputs.len(), input_limit)?;
325 sort_confidence_rows(&mut assessments, &mut inputs);
326 Ok(Self {
327 assessments,
328 inputs,
329 })
330 }
331
332 pub fn assessment_fingerprint(
334 &self,
335 confidence_uuid: Uuid,
336 ) -> Result<[u8; 32], KnowledgeError> {
337 let row = self
338 .assessments
339 .iter()
340 .find(|row| row.confidence_uuid == confidence_uuid)
341 .ok_or(KnowledgeError::Dangling("confidence_uuid"))?;
342 assessment_fingerprint_for_row(row, &inputs_for(&self.inputs, confidence_uuid))
343 }
344
345 pub fn assessment_fingerprints(&self) -> Result<Vec<(Uuid, [u8; 32])>, KnowledgeError> {
347 let mut inputs = HashMap::<Uuid, Vec<&ConfidenceInput>>::new();
348 for input in &self.inputs {
349 inputs.entry(input.confidence_uuid).or_default().push(input);
350 }
351 self.assessments
352 .iter()
353 .map(|row| {
354 let row_inputs = inputs
355 .get(&row.confidence_uuid)
356 .map_or(&[][..], Vec::as_slice);
357 Ok((
358 row.confidence_uuid,
359 assessment_fingerprint_for_row(row, row_inputs)?,
360 ))
361 })
362 .collect()
363 }
364
365 pub fn assessment_batch(&self) -> Result<RecordBatch, KnowledgeError> {
367 confidence_assessment_batch(&self.assessments)
368 }
369
370 pub fn input_batch(&self) -> Result<RecordBatch, KnowledgeError> {
372 confidence_input_batch(&self.inputs)
373 }
374
375 pub fn from_batches(
377 assessment_batches: &[RecordBatch],
378 input_batches: &[RecordBatch],
379 ) -> Result<Self, KnowledgeError> {
380 let mut assessments = Vec::new();
381 for batch in assessment_batches {
382 require_schema(batch, &CONFIDENCE_ASSESSMENT_SCHEMA, "confidence.schema")?;
383 let ids = fixed_column(batch, "confidence_uuid")?;
384 let assertions = fixed_column(batch, "assertion_uuid")?;
385 let policies = string_column(batch, "policy")?;
386 let policy_versions = u32_column(batch, "policy_version")?;
387 let values = f64_column(batch, "value")?;
388 let provenance = fixed_column(batch, "provenance_uuid")?;
389 let recorded = timestamp_column(batch, "recorded_at")?;
390 let versions = u32_column(batch, "contract_version")?;
391 for row in 0..batch.num_rows() {
392 assessments.push(ConfidenceAssessment {
393 confidence_uuid: uuid_at(ids, row, "confidence_uuid")?,
394 assertion_uuid: uuid_at(assertions, row, "assertion_uuid")?,
395 policy: ConfidencePolicy::parse(required_text(policies, row, "policy")?)?,
396 policy_version: required_u32(policy_versions, row, "policy_version")?,
397 value: optional_f64(values, row),
398 provenance_uuid: uuid_at(provenance, row, "provenance_uuid")?,
399 recorded_at_micros: required_i64(recorded, row, "recorded_at")?,
400 contract_version: required_u32(versions, row, "contract_version")?,
401 });
402 }
403 }
404 let mut inputs = Vec::new();
405 for batch in input_batches {
406 require_schema(batch, &CONFIDENCE_INPUT_SCHEMA, "confidence_input.schema")?;
407 let owners = fixed_column(batch, "confidence_uuid")?;
408 let ids = fixed_column(batch, "input_confidence_uuid")?;
409 let values = f64_column(batch, "input_value")?;
410 let ordinals = u32_column(batch, "ordinal")?;
411 let versions = u32_column(batch, "contract_version")?;
412 for row in 0..batch.num_rows() {
413 inputs.push(ConfidenceInput {
414 confidence_uuid: uuid_at(owners, row, "confidence_uuid")?,
415 input_confidence_uuid: uuid_at(ids, row, "input_confidence_uuid")?,
416 input_value: optional_f64(values, row),
417 ordinal: required_u32(ordinals, row, "ordinal")?,
418 contract_version: required_u32(versions, row, "contract_version")?,
419 });
420 }
421 }
422 Self::new(assessments, inputs)
423 }
424}
425
426fn confidence_assessment_batch(
427 rows: &[ConfidenceAssessment],
428) -> Result<RecordBatch, KnowledgeError> {
429 let mut ids = FixedSizeBinaryBuilder::with_capacity(rows.len(), 16);
430 let mut assertions = FixedSizeBinaryBuilder::with_capacity(rows.len(), 16);
431 let mut provenance = FixedSizeBinaryBuilder::with_capacity(rows.len(), 16);
432 for row in rows {
433 ids.append_value(row.confidence_uuid.as_bytes())?;
434 assertions.append_value(row.assertion_uuid.as_bytes())?;
435 provenance.append_value(row.provenance_uuid.as_bytes())?;
436 }
437 RecordBatch::try_new(
438 Arc::clone(&CONFIDENCE_ASSESSMENT_SCHEMA),
439 vec![
440 Arc::new(ids.finish()),
441 Arc::new(assertions.finish()),
442 Arc::new(StringArray::from_iter_values(
443 rows.iter().map(|row| row.policy.as_str()),
444 )),
445 Arc::new(UInt32Array::from_iter_values(
446 rows.iter().map(|row| row.policy_version),
447 )),
448 Arc::new(Float64Array::from(
449 rows.iter().map(|row| row.value).collect::<Vec<_>>(),
450 )),
451 Arc::new(provenance.finish()),
452 Arc::new(
453 TimestampMicrosecondArray::from_iter_values(
454 rows.iter().map(|row| row.recorded_at_micros),
455 )
456 .with_timezone("UTC"),
457 ),
458 Arc::new(UInt32Array::from_iter_values(
459 rows.iter().map(|row| row.contract_version),
460 )),
461 ],
462 )
463 .map_err(Into::into)
464}
465
466fn confidence_input_batch(rows: &[ConfidenceInput]) -> Result<RecordBatch, KnowledgeError> {
467 let mut owners = FixedSizeBinaryBuilder::with_capacity(rows.len(), 16);
468 let mut ids = FixedSizeBinaryBuilder::with_capacity(rows.len(), 16);
469 for row in rows {
470 owners.append_value(row.confidence_uuid.as_bytes())?;
471 ids.append_value(row.input_confidence_uuid.as_bytes())?;
472 }
473 RecordBatch::try_new(
474 Arc::clone(&CONFIDENCE_INPUT_SCHEMA),
475 vec![
476 Arc::new(owners.finish()),
477 Arc::new(ids.finish()),
478 Arc::new(Float64Array::from(
479 rows.iter().map(|row| row.input_value).collect::<Vec<_>>(),
480 )),
481 Arc::new(UInt32Array::from_iter_values(
482 rows.iter().map(|row| row.ordinal),
483 )),
484 Arc::new(UInt32Array::from_iter_values(
485 rows.iter().map(|row| row.contract_version),
486 )),
487 ],
488 )
489 .map_err(Into::into)
490}
491
492fn validate_confidence_rows(
493 assessments: &[ConfidenceAssessment],
494 inputs: &[ConfidenceInput],
495) -> Result<(), KnowledgeError> {
496 validate_confidence_rows_with_limits(
497 assessments,
498 inputs,
499 super::MAX_KNOWLEDGE_ROWS,
500 super::MAX_KNOWLEDGE_ROWS,
501 )
502}
503
504fn validate_confidence_rows_with_limits(
505 assessments: &[ConfidenceAssessment],
506 inputs: &[ConfidenceInput],
507 assessment_limit: usize,
508 input_limit: usize,
509) -> Result<(), KnowledgeError> {
510 check_confidence_row_limit(
511 "confidence_assessments",
512 assessments.len(),
513 assessment_limit,
514 )?;
515 check_confidence_row_limit("confidence_inputs", inputs.len(), input_limit)?;
516 let mut ids = HashSet::with_capacity(assessments.len());
517 let mut policies = HashMap::with_capacity(assessments.len());
518 for row in assessments {
519 #[cfg(test)]
520 CONFIDENCE_ROW_VALIDATION_COUNT.with(|count| count.set(count.get() + 1));
521 require_v7(row.confidence_uuid, "confidence_uuid")?;
522 require_v7(row.assertion_uuid, "assertion_uuid")?;
523 require_uuid(row.provenance_uuid, "provenance_uuid")?;
524 validate_confidence(row.value, "value")?;
525 if row.policy_version != 1 {
526 return Err(invalid("policy_version", "unsupported version"));
527 }
528 if row.contract_version != CONFIDENCE_ASSESSMENT_CONTRACT_VERSION {
529 return Err(invalid(
530 "confidence.contract_version",
531 "unsupported version",
532 ));
533 }
534 if !ids.insert(row.confidence_uuid) {
535 return Err(KnowledgeError::Duplicate("confidence_uuid"));
536 }
537 policies.insert(row.confidence_uuid, row.policy);
538 }
539 let mut input_ids = HashSet::with_capacity(inputs.len());
540 let mut normalized_inputs: HashMap<Uuid, Vec<(u32, Uuid, Option<f64>)>> = HashMap::new();
541 for input in inputs {
542 #[cfg(test)]
543 CONFIDENCE_ROW_VALIDATION_COUNT.with(|count| count.set(count.get() + 1));
544 require_v7(input.confidence_uuid, "confidence_uuid")?;
545 require_v7(input.input_confidence_uuid, "input_confidence_uuid")?;
546 validate_confidence(input.input_value, "input_value")?;
547 if input.contract_version != CONFIDENCE_INPUT_CONTRACT_VERSION {
548 return Err(invalid(
549 "confidence_input.contract_version",
550 "unsupported version",
551 ));
552 }
553 if !ids.contains(&input.confidence_uuid) {
554 return Err(KnowledgeError::Dangling("confidence_uuid"));
555 }
556 if !input_ids.insert((input.confidence_uuid, input.input_confidence_uuid)) {
557 return Err(KnowledgeError::Duplicate("input_confidence_uuid"));
558 }
559 normalized_inputs
560 .entry(input.confidence_uuid)
561 .or_default()
562 .push((
563 input.ordinal,
564 input.input_confidence_uuid,
565 input.input_value,
566 ));
567 }
568 for (confidence_uuid, policy) in policies {
569 let assessment = assessments
570 .iter()
571 .find(|row| row.confidence_uuid == confidence_uuid)
572 .expect("validated assessment identity");
573 let values = normalized_inputs.entry(confidence_uuid).or_default();
574 values.sort_by_key(|(ordinal, _, _)| *ordinal);
575 if values
576 .iter()
577 .enumerate()
578 .any(|(expected, (actual, _, _))| usize::try_from(*actual) != Ok(expected))
579 {
580 return Err(invalid("ordinal", "must be contiguous from zero"));
581 }
582 if values.windows(2).any(|pair| pair[0].1 >= pair[1].1) {
583 return Err(invalid(
584 "input_confidence_uuid",
585 "must be unique UUID-normalized order",
586 ));
587 }
588 validate_policy_snapshot(assessment, policy, values)?;
589 }
590 Ok(())
591}
592
593fn check_confidence_row_limit(
594 participant: &'static str,
595 observed: usize,
596 limit: usize,
597) -> Result<(), KnowledgeError> {
598 if observed > limit {
599 Err(KnowledgeError::Limit {
600 participant,
601 observed,
602 limit,
603 })
604 } else {
605 Ok(())
606 }
607}
608
609fn sort_confidence_rows(assessments: &mut [ConfidenceAssessment], inputs: &mut [ConfidenceInput]) {
610 let times = assessments
611 .iter()
612 .map(|row| (row.confidence_uuid, row.recorded_at_micros))
613 .collect::<HashMap<_, _>>();
614 assessments.sort_by_key(|row| (row.recorded_at_micros, row.confidence_uuid));
615 inputs.sort_by_key(|row| {
616 (
617 times[&row.confidence_uuid],
618 row.confidence_uuid,
619 row.ordinal,
620 row.input_confidence_uuid,
621 )
622 });
623}
624
625#[cfg(test)]
626fn reset_confidence_row_validation_count() {
627 CONFIDENCE_ROW_VALIDATION_COUNT.with(|count| count.set(0));
628}
629
630#[cfg(test)]
631fn confidence_row_validation_count() -> usize {
632 CONFIDENCE_ROW_VALIDATION_COUNT.with(std::cell::Cell::get)
633}
634
635fn validate_policy_snapshot(
636 assessment: &ConfidenceAssessment,
637 policy: ConfidencePolicy,
638 values: &[(u32, Uuid, Option<f64>)],
639) -> Result<(), KnowledgeError> {
640 match policy {
641 ConfidencePolicy::Explicit => {
642 if !values.is_empty() {
643 return Err(invalid(
644 "confidence_inputs",
645 "explicit policy has no inputs",
646 ));
647 }
648 if assessment.value.is_none() {
649 return Err(invalid("value", "explicit policy requires a value"));
650 }
651 }
652 ConfidencePolicy::ConservativeMin => {
653 let expected =
654 if values.is_empty() || values.iter().any(|(_, _, value)| value.is_none()) {
655 None
656 } else {
657 values
658 .iter()
659 .filter_map(|(_, _, value)| *value)
660 .reduce(f64::min)
661 };
662 if assessment.value != expected {
663 return Err(invalid(
664 "value",
665 "does not match conservative_min input snapshot",
666 ));
667 }
668 }
669 }
670 Ok(())
671}
672
673fn assessment_fingerprint_for_row(
674 row: &ConfidenceAssessment,
675 inputs: &[&ConfidenceInput],
676) -> Result<[u8; 32], KnowledgeError> {
677 let mut writer = CanonicalWriter::new();
678 writer.raw(b"GFCA")?;
679 writer.u32(CONFIDENCE_ASSESSMENT_CONTRACT_VERSION)?;
680 writer.raw(row.assertion_uuid.as_bytes())?;
681 writer.text(row.policy.as_str())?;
682 writer.u32(row.policy_version)?;
683 canonical_optional_f64(&mut writer, row.value)?;
684 writer.u64(inputs.len() as u64)?;
685 for input in inputs {
686 writer.raw(input.input_confidence_uuid.as_bytes())?;
687 canonical_optional_f64(&mut writer, input.input_value)?;
688 writer.u32(input.ordinal)?;
689 }
690 Ok(fingerprint(
691 CanonicalDomain::ConfidenceAssessment,
692 CANONICAL_CONTRACT_VERSION,
693 &writer.finish(),
694 )?)
695}
696
697fn inputs_for(rows: &[ConfidenceInput], confidence_uuid: Uuid) -> Vec<&ConfidenceInput> {
698 let mut inputs = rows
699 .iter()
700 .filter(|row| row.confidence_uuid == confidence_uuid)
701 .collect::<Vec<_>>();
702 inputs.sort_by_key(|row| (row.ordinal, row.input_confidence_uuid));
703 inputs
704}
705
706#[cfg(test)]
707mod tests;