1#![forbid(unsafe_code)]
7
8mod algorithm_run;
9mod artifact;
10mod artifact_preference;
11mod belief_projection;
12mod confidence;
13mod derivation;
14mod hypothesis;
15mod reasoning;
16pub mod research;
17mod retention_dependency;
18mod source;
19mod status;
20mod supersession;
21mod valid_time;
22
23pub use algorithm_run::{AlgorithmRun, AlgorithmRunEvent, AlgorithmRunLedger, AlgorithmRunState};
24pub use confidence::{ConfidenceAssessment, ConfidenceInput, ConfidenceLedger, ConfidencePolicy};
25
26pub use artifact::{
27 ARTIFACT_AVAILABILITY_REGISTRY_VERSION, ARTIFACT_CONTRACT_VERSION,
28 ARTIFACT_KIND_REGISTRY_VERSION, ARTIFACT_PAYLOAD_KIND_REGISTRY_VERSION, ARTIFACT_SCHEMA,
29 Artifact, ArtifactAvailability, ArtifactKind, ArtifactLedger, ArtifactPayloadKind,
30 MAX_ARTIFACT_EXTERNAL_URI_BYTES, MAX_ARTIFACT_MEDIA_TYPE_BYTES,
31};
32pub use artifact_preference::{
33 ARTIFACT_PREFERENCE_CONTRACT_VERSION, ARTIFACT_PREFERENCE_SCHEMA, ArtifactPreferenceEvent,
34 ArtifactPreferenceLedger, MAX_ARTIFACT_PREFERENCE_REASON_BYTES,
35};
36pub use derivation::{
37 ARTIFACT_DERIVATION_CONTRACT_VERSION, ARTIFACT_DERIVATION_SCHEMA, ArtifactDerivation,
38 ArtifactDerivationLedger, DERIVATION_ROLE_REGISTRY_VERSION,
39 DERIVATION_SUBJECT_KIND_REGISTRY_VERSION, DerivationRole, DerivationSubjectKind,
40};
41pub use hypothesis::{
42 HYPOTHESIS_GROUP_CONTRACT_VERSION, HYPOTHESIS_GROUP_SCHEMA, HYPOTHESIS_KEY_POLICY_VERSION,
43 HYPOTHESIS_MEMBERSHIP_CONTRACT_VERSION, HYPOTHESIS_MEMBERSHIP_SCHEMA,
44 HYPOTHESIS_SELECTION_CONTRACT_VERSION, HYPOTHESIS_SELECTION_SCHEMA,
45 HYPOTHESIS_STATE_POLICY_VERSION, HypothesisGroup, HypothesisLedger, HypothesisMembershipAction,
46 HypothesisMembershipEvent, HypothesisSelectionEvent, MAX_HYPOTHESIS_QUESTION_KEY_BYTES,
47};
48#[cfg(feature = "test-support")]
49pub use reasoning::test_support as reasoning_test_support;
50pub use reasoning::{
51 EPISTEMIC_CAPABILITY_VERSION, MAX_REASONING_CONTENT_BYTES,
52 REASONING_CONTENT_FORMAT_REGISTRY_VERSION, REASONING_CONTRACT_VERSION,
53 REASONING_KIND_REGISTRY_VERSION, REASONING_SCHEMA, ReasoningContentFormat, ReasoningKind,
54 ReasoningLedger, ReasoningRecord,
55};
56pub use retention_dependency::{
57 RETENTION_DEPENDENCY_CLASS_REGISTRY_VERSION, RETENTION_DEPENDENCY_CONTRACT_VERSION,
58 RETENTION_DEPENDENCY_SCHEMA, RETENTION_REQUIRED_KIND_REGISTRY_VERSION, RetentionDependency,
59 RetentionDependencyClass, RetentionDependencyLedger, RetentionRequiredKind,
60};
61pub use source::{
62 MAX_SOURCE_IDENTITY_URI_BYTES, MAX_SOURCE_LABEL_BYTES, SOURCE_CONTRACT_VERSION,
63 SOURCE_KIND_REGISTRY_VERSION, SOURCE_SCHEMA, Source, SourceKind, SourceLedger,
64};
65pub use status::{
66 ASSERTION_STATUS_CONTRACT_VERSION, ASSERTION_STATUS_REGISTRY_VERSION, ASSERTION_STATUS_SCHEMA,
67 AssertionStatus, AssertionStatusEvent, AssertionStatusLedger,
68};
69pub use supersession::{
70 ASSERTION_SUPERSESSION_CONTRACT_VERSION, ASSERTION_SUPERSESSION_POLICY_VERSION,
71 ASSERTION_SUPERSESSION_SCHEMA, AssertionSupersession, AssertionSupersessionLedger,
72};
73pub use valid_time::{
74 ASSERTION_VALIDITY_CONTRACT_VERSION, ASSERTION_VALIDITY_POLICY_VERSION,
75 ASSERTION_VALIDITY_SCHEMA, AssertionValidityEvent, AssertionValidityLedger,
76 VALID_TIME_CAPABILITY_VERSION,
77};
78
79use std::collections::{HashMap, HashSet};
80use std::sync::{Arc, LazyLock};
81
82use arrow::array::{
83 Array, BinaryArray, FixedSizeBinaryArray, FixedSizeBinaryBuilder, Float64Array, StringArray,
84 TimestampMicrosecondArray, UInt32Array,
85};
86use arrow::datatypes::{DataType, Field, Schema, SchemaRef, TimeUnit};
87use arrow::record_batch::RecordBatch;
88use graphforge_core::canonical::{
89 CANONICAL_CONTRACT_VERSION, CanonicalDomain, CanonicalError, CanonicalWriter, fingerprint,
90};
91use uuid::{Uuid, Version};
92
93pub const KNOWLEDGE_CAPABILITY_VERSION: u32 = 1;
95pub const ASSERTION_CONTRACT_VERSION: u32 = 1;
97pub const ASSERTION_GRAPH_REF_CONTRACT_VERSION: u32 = 1;
99pub const CONFIDENCE_ASSESSMENT_CONTRACT_VERSION: u32 = 1;
101pub const CONFIDENCE_INPUT_CONTRACT_VERSION: u32 = 1;
103pub const EVIDENCE_LINK_CONTRACT_VERSION: u32 = 1;
105pub const ALGORITHM_RUN_CONTRACT_VERSION: u32 = 1;
107pub const ALGORITHM_RUN_EVENT_CONTRACT_VERSION: u32 = 1;
109pub const CONFIDENCE_POLICY_REGISTRY_VERSION: u32 = 1;
111pub const GRAPH_OBJECT_KIND_REGISTRY_VERSION: u32 = 1;
113pub const ASSERTION_GRAPH_ROLE_REGISTRY_VERSION: u32 = 1;
115pub const EVIDENCE_SOURCE_KIND_REGISTRY_VERSION: u32 = 2;
117pub const EVIDENCE_ROLE_REGISTRY_VERSION: u32 = 1;
119pub const ALGORITHM_RUN_STATE_REGISTRY_VERSION: u32 = 1;
121pub const MAX_KNOWLEDGE_ROWS: usize = 1_000_000;
123
124pub static ASSERTION_SCHEMA: LazyLock<SchemaRef> = LazyLock::new(|| {
126 Arc::new(Schema::new(vec![
127 uuid_field("assertion_uuid", false),
128 Field::new("claim", DataType::Utf8, false),
129 uuid_field("provenance_uuid", false),
130 Field::new(
131 "recorded_at",
132 DataType::Timestamp(TimeUnit::Microsecond, Some("UTC".into())),
133 false,
134 ),
135 Field::new("contract_version", DataType::UInt32, false),
136 ]))
137});
138
139pub static ASSERTION_GRAPH_REF_SCHEMA: LazyLock<SchemaRef> = LazyLock::new(|| {
141 Arc::new(Schema::new(vec![
142 uuid_field("assertion_uuid", false),
143 uuid_field("graph_uuid", false),
144 Field::new("graph_kind", DataType::Utf8, false),
145 Field::new("role", DataType::Utf8, false),
146 Field::new("ordinal", DataType::UInt32, false),
147 Field::new("contract_version", DataType::UInt32, false),
148 ]))
149});
150
151pub static CONFIDENCE_ASSESSMENT_SCHEMA: LazyLock<SchemaRef> = LazyLock::new(|| {
153 Arc::new(Schema::new(vec![
154 uuid_field("confidence_uuid", false),
155 uuid_field("assertion_uuid", false),
156 Field::new("policy", DataType::Utf8, false),
157 Field::new("policy_version", DataType::UInt32, false),
158 Field::new("value", DataType::Float64, true),
159 uuid_field("provenance_uuid", false),
160 Field::new(
161 "recorded_at",
162 DataType::Timestamp(TimeUnit::Microsecond, Some("UTC".into())),
163 false,
164 ),
165 Field::new("contract_version", DataType::UInt32, false),
166 ]))
167});
168
169pub static CONFIDENCE_INPUT_SCHEMA: LazyLock<SchemaRef> = LazyLock::new(|| {
171 Arc::new(Schema::new(vec![
172 uuid_field("confidence_uuid", false),
173 uuid_field("input_confidence_uuid", false),
174 Field::new("input_value", DataType::Float64, true),
175 Field::new("ordinal", DataType::UInt32, false),
176 Field::new("contract_version", DataType::UInt32, false),
177 ]))
178});
179
180pub static EVIDENCE_LINK_SCHEMA: LazyLock<SchemaRef> = LazyLock::new(|| {
182 Arc::new(Schema::new(vec![
183 uuid_field("evidence_uuid", false),
184 uuid_field("assertion_uuid", false),
185 uuid_field("source_uuid", false),
186 Field::new("source_kind", DataType::Utf8, false),
187 Field::new("role", DataType::Utf8, false),
188 Field::new("weight", DataType::Float64, true),
189 uuid_field("provenance_uuid", false),
190 Field::new(
191 "recorded_at",
192 DataType::Timestamp(TimeUnit::Microsecond, Some("UTC".into())),
193 false,
194 ),
195 Field::new("contract_version", DataType::UInt32, false),
196 ]))
197});
198
199pub static ALGORITHM_RUN_SCHEMA: LazyLock<SchemaRef> = LazyLock::new(|| {
201 Arc::new(Schema::new(vec![
202 uuid_field("run_uuid", false),
203 Field::new("algorithm", DataType::Utf8, false),
204 Field::new("algorithm_version", DataType::UInt32, false),
205 Field::new("descriptor_version", DataType::UInt32, false),
206 Field::new("descriptor", DataType::Binary, false),
207 Field::new(
208 "projection_fingerprint",
209 DataType::FixedSizeBinary(32),
210 false,
211 ),
212 uuid_field("provenance_uuid", false),
213 Field::new(
214 "started_at",
215 DataType::Timestamp(TimeUnit::Microsecond, Some("UTC".into())),
216 false,
217 ),
218 Field::new("contract_version", DataType::UInt32, false),
219 ]))
220});
221
222pub static ALGORITHM_RUN_EVENT_SCHEMA: LazyLock<SchemaRef> = LazyLock::new(|| {
224 Arc::new(Schema::new(vec![
225 uuid_field("event_uuid", false),
226 uuid_field("run_uuid", false),
227 Field::new("state", DataType::Utf8, false),
228 Field::new("result_fingerprint", DataType::FixedSizeBinary(32), true),
229 Field::new("error_code", DataType::Utf8, true),
230 Field::new(
231 "recorded_at",
232 DataType::Timestamp(TimeUnit::Microsecond, Some("UTC".into())),
233 false,
234 ),
235 uuid_field("provenance_uuid", false),
236 Field::new("contract_version", DataType::UInt32, false),
237 ]))
238});
239
240static ASSERTION_SCHEMA_FINGERPRINT: LazyLock<[u8; 32]> = LazyLock::new(|| {
241 fingerprint(
242 CanonicalDomain::Schema,
243 CANONICAL_CONTRACT_VERSION,
244 b"assertion/1|assertion_uuid:fixed[16]:required|claim:utf8:required|provenance_uuid:fixed[16]:required|recorded_at:timestamp_us_utc:required|contract_version:u32:required",
245 )
246 .expect("registered assertion schema is within canonical bounds")
247});
248
249static ASSERTION_GRAPH_REF_SCHEMA_FINGERPRINT: LazyLock<[u8; 32]> = LazyLock::new(|| {
250 fingerprint(
251 CanonicalDomain::Schema,
252 CANONICAL_CONTRACT_VERSION,
253 b"assertion_graph_ref/1|assertion_uuid:fixed[16]:required|graph_uuid:fixed[16]:required|graph_kind:utf8:required|role:utf8:required|ordinal:u32:required|contract_version:u32:required",
254 )
255 .expect("registered assertion graph-ref schema is within canonical bounds")
256});
257
258static CONFIDENCE_ASSESSMENT_SCHEMA_FINGERPRINT: LazyLock<[u8; 32]> = LazyLock::new(|| {
259 fingerprint(
260 CanonicalDomain::Schema,
261 CANONICAL_CONTRACT_VERSION,
262 b"confidence_assessment/1|confidence_uuid:fixed[16]:required|assertion_uuid:fixed[16]:required|policy:utf8:required|policy_version:u32:required|value:f64:nullable|provenance_uuid:fixed[16]:required|recorded_at:timestamp_us_utc:required|contract_version:u32:required",
263 )
264 .expect("registered confidence-assessment schema is within canonical bounds")
265});
266
267static CONFIDENCE_INPUT_SCHEMA_FINGERPRINT: LazyLock<[u8; 32]> = LazyLock::new(|| {
268 fingerprint(
269 CanonicalDomain::Schema,
270 CANONICAL_CONTRACT_VERSION,
271 b"confidence_input/1|confidence_uuid:fixed[16]:required|input_confidence_uuid:fixed[16]:required|input_value:f64:nullable|ordinal:u32:required|contract_version:u32:required",
272 )
273 .expect("registered confidence-input schema is within canonical bounds")
274});
275
276static EVIDENCE_LINK_SCHEMA_FINGERPRINT: LazyLock<[u8; 32]> = LazyLock::new(|| {
277 fingerprint(
278 CanonicalDomain::Schema,
279 CANONICAL_CONTRACT_VERSION,
280 b"evidence_link/1|evidence_uuid:fixed[16]:required|assertion_uuid:fixed[16]:required|source_uuid:fixed[16]:required|source_kind:utf8:required|role:utf8:required|weight:f64:nullable|provenance_uuid:fixed[16]:required|recorded_at:timestamp_us_utc:required|contract_version:u32:required",
281 )
282 .expect("registered evidence-link schema is within canonical bounds")
283});
284
285static ALGORITHM_RUN_SCHEMA_FINGERPRINT: LazyLock<[u8; 32]> = LazyLock::new(|| {
286 fingerprint(
287 CanonicalDomain::Schema,
288 CANONICAL_CONTRACT_VERSION,
289 b"algorithm_run/1|run_uuid:fixed[16]:required|algorithm:utf8:required|algorithm_version:u32:required|descriptor_version:u32:required|descriptor:binary:required|projection_fingerprint:fixed[32]:required|provenance_uuid:fixed[16]:required|started_at:timestamp_us_utc:required|contract_version:u32:required",
290 )
291 .expect("registered algorithm-run schema is within canonical bounds")
292});
293
294static ALGORITHM_RUN_EVENT_SCHEMA_FINGERPRINT: LazyLock<[u8; 32]> = LazyLock::new(|| {
295 fingerprint(
296 CanonicalDomain::Schema,
297 CANONICAL_CONTRACT_VERSION,
298 b"algorithm_run_event/1|event_uuid:fixed[16]:required|run_uuid:fixed[16]:required|state:utf8:required|result_fingerprint:fixed[32]:nullable|error_code:utf8:nullable|recorded_at:timestamp_us_utc:required|provenance_uuid:fixed[16]:required|contract_version:u32:required",
299 )
300 .expect("registered algorithm-run-event schema is within canonical bounds")
301});
302
303fn uuid_field(name: &str, nullable: bool) -> Field {
304 Field::new(name, DataType::FixedSizeBinary(16), nullable)
305}
306
307#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, serde::Serialize, serde::Deserialize)]
309#[serde(rename_all = "snake_case")]
310pub enum GraphObjectKind {
311 Node,
313 Edge,
315}
316
317impl GraphObjectKind {
318 #[must_use]
320 pub const fn as_str(self) -> &'static str {
321 match self {
322 Self::Node => "node",
323 Self::Edge => "edge",
324 }
325 }
326
327 fn parse(value: &str) -> Result<Self, KnowledgeError> {
328 match value {
329 "node" => Ok(Self::Node),
330 "edge" => Ok(Self::Edge),
331 _ => Err(invalid("graph_kind", "unknown closed value")),
332 }
333 }
334}
335
336#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, serde::Serialize, serde::Deserialize)]
338#[serde(rename_all = "snake_case")]
339pub enum AssertionGraphRole {
340 Subject,
342 Object,
344 Context,
346}
347
348impl AssertionGraphRole {
349 #[must_use]
351 pub const fn as_str(self) -> &'static str {
352 match self {
353 Self::Subject => "subject",
354 Self::Object => "object",
355 Self::Context => "context",
356 }
357 }
358
359 fn parse(value: &str) -> Result<Self, KnowledgeError> {
360 match value {
361 "subject" => Ok(Self::Subject),
362 "object" => Ok(Self::Object),
363 "context" => Ok(Self::Context),
364 _ => Err(invalid("role", "unknown closed value")),
365 }
366 }
367}
368
369#[derive(Clone, Debug, PartialEq, Eq)]
371pub struct Assertion {
372 pub assertion_uuid: Uuid,
374 pub claim: String,
376 pub provenance_uuid: Uuid,
378 pub recorded_at_micros: i64,
380 pub contract_version: u32,
382}
383
384impl Assertion {
385 pub fn new(
387 assertion_uuid: Uuid,
388 claim: String,
389 provenance_uuid: Uuid,
390 recorded_at_micros: i64,
391 ) -> Result<Self, KnowledgeError> {
392 require_v7(assertion_uuid, "assertion_uuid")?;
393 require_uuid(provenance_uuid, "provenance_uuid")?;
394 validate_claim(&claim)?;
395 Ok(Self {
396 assertion_uuid,
397 claim,
398 provenance_uuid,
399 recorded_at_micros,
400 contract_version: ASSERTION_CONTRACT_VERSION,
401 })
402 }
403}
404
405#[derive(Clone, Debug, PartialEq, Eq)]
407pub struct AssertionGraphRef {
408 pub assertion_uuid: Uuid,
410 pub graph_uuid: Uuid,
412 pub graph_kind: GraphObjectKind,
414 pub role: AssertionGraphRole,
416 pub ordinal: u32,
418 pub contract_version: u32,
420}
421
422impl AssertionGraphRef {
423 pub fn new(
425 assertion_uuid: Uuid,
426 graph_uuid: Uuid,
427 graph_kind: GraphObjectKind,
428 role: AssertionGraphRole,
429 ordinal: u32,
430 ) -> Result<Self, KnowledgeError> {
431 require_v7(assertion_uuid, "assertion_uuid")?;
432 require_uuid(graph_uuid, "graph_uuid")?;
433 Ok(Self {
434 assertion_uuid,
435 graph_uuid,
436 graph_kind,
437 role,
438 ordinal,
439 contract_version: ASSERTION_GRAPH_REF_CONTRACT_VERSION,
440 })
441 }
442}
443
444#[derive(Clone, Debug, Default, PartialEq, Eq)]
457pub struct AssertionLedger {
458 assertions: Vec<Assertion>,
460 graph_refs: Vec<AssertionGraphRef>,
462}
463
464#[cfg(test)]
465thread_local! {
466 static ASSERTION_ROW_VALIDATION_COUNT: std::cell::Cell<usize> = const { std::cell::Cell::new(0) };
467}
468
469impl AssertionLedger {
470 #[must_use]
472 pub fn assertions(&self) -> &[Assertion] {
473 &self.assertions
474 }
475
476 #[must_use]
478 pub fn graph_refs(&self) -> &[AssertionGraphRef] {
479 &self.graph_refs
480 }
481
482 #[must_use]
484 pub fn into_parts(self) -> (Vec<Assertion>, Vec<AssertionGraphRef>) {
485 (self.assertions, self.graph_refs)
486 }
487
488 pub fn new(
490 mut assertions: Vec<Assertion>,
491 mut graph_refs: Vec<AssertionGraphRef>,
492 ) -> Result<Self, KnowledgeError> {
493 validate_rows(&assertions, &graph_refs)?;
494 sort_assertion_rows(&mut assertions, &mut graph_refs);
495 Ok(Self {
496 assertions,
497 graph_refs,
498 })
499 }
500
501 pub fn merge(&self, staged: &Self) -> Result<Self, KnowledgeError> {
503 self.merge_with_limits(staged, MAX_KNOWLEDGE_ROWS, MAX_KNOWLEDGE_ROWS)
504 }
505
506 fn merge_with_limits(
507 &self,
508 staged: &Self,
509 assertion_limit: usize,
510 graph_ref_limit: usize,
511 ) -> Result<Self, KnowledgeError> {
512 let mut assertions = self.assertions.clone();
513 let mut refs = self.graph_refs.clone();
514 let mut by_id = assertions
515 .iter()
516 .cloned()
517 .map(|row| (row.assertion_uuid, row))
518 .collect::<HashMap<_, _>>();
519 for row in &staged.assertions {
520 if let Some(existing) = by_id.get(&row.assertion_uuid)
521 && (existing != row
522 || refs_for(&refs, row.assertion_uuid)
523 != refs_for(&staged.graph_refs, row.assertion_uuid))
524 {
525 return Err(KnowledgeError::Conflict("assertion_uuid"));
526 }
527 if by_id.insert(row.assertion_uuid, row.clone()).is_none() {
528 assertions.push(row.clone());
529 refs.extend(
530 staged
531 .graph_refs
532 .iter()
533 .filter(|reference| reference.assertion_uuid == row.assertion_uuid)
534 .cloned(),
535 );
536 }
537 }
538 check_assertion_row_limit("assertions", assertions.len(), assertion_limit)?;
539 check_assertion_row_limit("assertion_graph_refs", refs.len(), graph_ref_limit)?;
540 sort_assertion_rows(&mut assertions, &mut refs);
541 Ok(Self {
542 assertions,
543 graph_refs: refs,
544 })
545 }
546
547 pub fn assertion_fingerprint(&self, assertion_uuid: Uuid) -> Result<[u8; 32], KnowledgeError> {
549 let assertion = self
550 .assertions
551 .iter()
552 .find(|row| row.assertion_uuid == assertion_uuid)
553 .ok_or(KnowledgeError::Dangling("assertion_uuid"))?;
554 assertion_fingerprint_for_row(assertion, &refs_for(&self.graph_refs, assertion_uuid))
555 }
556
557 pub fn assertion_fingerprints(&self) -> Result<Vec<(Uuid, [u8; 32])>, KnowledgeError> {
559 let mut refs = HashMap::<Uuid, Vec<&AssertionGraphRef>>::new();
560 for reference in &self.graph_refs {
561 refs.entry(reference.assertion_uuid)
562 .or_default()
563 .push(reference);
564 }
565 self.assertions
566 .iter()
567 .map(|row| {
568 let row_refs = refs.get(&row.assertion_uuid).map_or(&[][..], Vec::as_slice);
569 Ok((
570 row.assertion_uuid,
571 assertion_fingerprint_for_row(row, row_refs)?,
572 ))
573 })
574 .collect()
575 }
576
577 pub fn assertion_batch(&self) -> Result<RecordBatch, KnowledgeError> {
579 assertion_batch(&self.assertions)
580 }
581
582 pub fn graph_ref_batch(&self) -> Result<RecordBatch, KnowledgeError> {
584 graph_ref_batch(&self.graph_refs)
585 }
586
587 pub fn from_batches(
589 assertion_batches: &[RecordBatch],
590 graph_ref_batches: &[RecordBatch],
591 ) -> Result<Self, KnowledgeError> {
592 let mut assertions = Vec::new();
593 for batch in assertion_batches {
594 require_schema(batch, &ASSERTION_SCHEMA, "assertion.schema")?;
595 let ids = fixed_column(batch, "assertion_uuid")?;
596 let claims = string_column(batch, "claim")?;
597 let provenance = fixed_column(batch, "provenance_uuid")?;
598 let recorded = timestamp_column(batch, "recorded_at")?;
599 let versions = u32_column(batch, "contract_version")?;
600 for row in 0..batch.num_rows() {
601 assertions.push(Assertion {
602 assertion_uuid: uuid_at(ids, row, "assertion_uuid")?,
603 claim: required_text(claims, row, "claim")?.to_owned(),
604 provenance_uuid: uuid_at(provenance, row, "provenance_uuid")?,
605 recorded_at_micros: required_i64(recorded, row, "recorded_at")?,
606 contract_version: required_u32(versions, row, "contract_version")?,
607 });
608 }
609 }
610 let mut refs = Vec::new();
611 for batch in graph_ref_batches {
612 require_schema(
613 batch,
614 &ASSERTION_GRAPH_REF_SCHEMA,
615 "assertion_graph_ref.schema",
616 )?;
617 let assertions_col = fixed_column(batch, "assertion_uuid")?;
618 let graph_ids = fixed_column(batch, "graph_uuid")?;
619 let kinds = string_column(batch, "graph_kind")?;
620 let roles = string_column(batch, "role")?;
621 let ordinals = u32_column(batch, "ordinal")?;
622 let versions = u32_column(batch, "contract_version")?;
623 for row in 0..batch.num_rows() {
624 refs.push(AssertionGraphRef {
625 assertion_uuid: uuid_at(assertions_col, row, "assertion_uuid")?,
626 graph_uuid: uuid_at(graph_ids, row, "graph_uuid")?,
627 graph_kind: GraphObjectKind::parse(required_text(kinds, row, "graph_kind")?)?,
628 role: AssertionGraphRole::parse(required_text(roles, row, "role")?)?,
629 ordinal: required_u32(ordinals, row, "ordinal")?,
630 contract_version: required_u32(versions, row, "contract_version")?,
631 });
632 }
633 }
634 Self::new(assertions, refs)
635 }
636}
637
638#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, serde::Serialize, serde::Deserialize)]
640#[serde(rename_all = "snake_case")]
641pub enum EvidenceSourceKind {
642 Document,
644 Observation,
646 GraphNode,
648 GraphEdge,
650 Source,
652 Artifact,
654}
655
656impl EvidenceSourceKind {
657 #[must_use]
659 pub const fn as_str(self) -> &'static str {
660 match self {
661 Self::Document => "document",
662 Self::Observation => "observation",
663 Self::GraphNode => "graph_node",
664 Self::GraphEdge => "graph_edge",
665 Self::Source => "source",
666 Self::Artifact => "artifact",
667 }
668 }
669
670 fn parse(value: &str) -> Result<Self, KnowledgeError> {
671 match value {
672 "document" => Ok(Self::Document),
673 "observation" => Ok(Self::Observation),
674 "graph_node" => Ok(Self::GraphNode),
675 "graph_edge" => Ok(Self::GraphEdge),
676 "source" => Ok(Self::Source),
677 "artifact" => Ok(Self::Artifact),
678 _ => Err(invalid("source_kind", "unknown closed value")),
679 }
680 }
681}
682
683#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, serde::Serialize, serde::Deserialize)]
685#[serde(rename_all = "snake_case")]
686pub enum EvidenceRole {
687 Supports,
689 Contradicts,
691 Context,
693}
694
695impl EvidenceRole {
696 #[must_use]
698 pub const fn as_str(self) -> &'static str {
699 match self {
700 Self::Supports => "supports",
701 Self::Contradicts => "contradicts",
702 Self::Context => "context",
703 }
704 }
705
706 fn parse(value: &str) -> Result<Self, KnowledgeError> {
707 match value {
708 "supports" => Ok(Self::Supports),
709 "contradicts" => Ok(Self::Contradicts),
710 "context" => Ok(Self::Context),
711 _ => Err(invalid("role", "unknown closed value")),
712 }
713 }
714}
715
716#[derive(Clone, Debug, PartialEq)]
718pub struct EvidenceLink {
719 pub evidence_uuid: Uuid,
721 pub assertion_uuid: Uuid,
723 pub source_uuid: Uuid,
725 pub source_kind: EvidenceSourceKind,
727 pub role: EvidenceRole,
729 pub weight: Option<f64>,
731 pub provenance_uuid: Uuid,
733 pub recorded_at_micros: i64,
735 pub contract_version: u32,
737}
738
739impl EvidenceLink {
740 #[allow(clippy::too_many_arguments)]
742 pub fn new(
743 evidence_uuid: Uuid,
744 assertion_uuid: Uuid,
745 source_uuid: Uuid,
746 source_kind: EvidenceSourceKind,
747 role: EvidenceRole,
748 weight: Option<f64>,
749 provenance_uuid: Uuid,
750 recorded_at_micros: i64,
751 ) -> Result<Self, KnowledgeError> {
752 require_v7(evidence_uuid, "evidence_uuid")?;
753 require_v7(assertion_uuid, "assertion_uuid")?;
754 require_uuid(source_uuid, "source_uuid")?;
755 require_uuid(provenance_uuid, "provenance_uuid")?;
756 validate_confidence(weight, "weight")?;
757 let weight = weight.map(normalize_zero);
758 Ok(Self {
759 evidence_uuid,
760 assertion_uuid,
761 source_uuid,
762 source_kind,
763 role,
764 weight,
765 provenance_uuid,
766 recorded_at_micros,
767 contract_version: EVIDENCE_LINK_CONTRACT_VERSION,
768 })
769 }
770}
771
772#[derive(Clone, Debug, Default, PartialEq)]
784pub struct EvidenceLedger {
785 links: Vec<EvidenceLink>,
787}
788
789#[cfg(test)]
790thread_local! {
791 static EVIDENCE_ROW_VALIDATION_COUNT: std::cell::Cell<usize> = const { std::cell::Cell::new(0) };
792}
793
794impl EvidenceLedger {
795 pub fn new(mut links: Vec<EvidenceLink>) -> Result<Self, KnowledgeError> {
797 if links.len() > MAX_KNOWLEDGE_ROWS {
798 return Err(KnowledgeError::Limit {
799 participant: "evidence",
800 observed: links.len(),
801 limit: MAX_KNOWLEDGE_ROWS,
802 });
803 }
804 let mut ids = HashSet::new();
805 for link in &links {
806 #[cfg(test)]
807 EVIDENCE_ROW_VALIDATION_COUNT.with(|count| count.set(count.get() + 1));
808 require_v7(link.evidence_uuid, "evidence_uuid")?;
809 require_v7(link.assertion_uuid, "assertion_uuid")?;
810 require_uuid(link.source_uuid, "source_uuid")?;
811 require_uuid(link.provenance_uuid, "provenance_uuid")?;
812 if link.contract_version != EVIDENCE_LINK_CONTRACT_VERSION {
813 return Err(invalid("contract_version", "unsupported evidence version"));
814 }
815 if !ids.insert(link.evidence_uuid) {
816 return Err(KnowledgeError::Duplicate("evidence_uuid"));
817 }
818 if let Some(weight) = link.weight {
819 validate_confidence(Some(weight), "weight")?;
820 }
821 }
822 links.sort_by_key(|row| (row.recorded_at_micros, row.evidence_uuid));
823 Ok(Self { links })
824 }
825
826 #[must_use]
828 pub fn links(&self) -> &[EvidenceLink] {
829 &self.links
830 }
831
832 #[must_use]
834 pub fn into_links(self) -> Vec<EvidenceLink> {
835 self.links
836 }
837
838 pub fn merge(&self, staged: &Self) -> Result<Self, KnowledgeError> {
840 let mut links = self.links.clone();
841 for row in &staged.links {
842 if let Some(existing) = links
843 .iter()
844 .find(|existing| existing.evidence_uuid == row.evidence_uuid)
845 {
846 if existing != row {
847 return Err(KnowledgeError::Conflict("evidence_uuid"));
848 }
849 } else {
850 links.push(row.clone());
851 }
852 }
853 check_limit("evidence", links.len())?;
854 links.sort_by_key(|row| (row.recorded_at_micros, row.evidence_uuid));
855 Ok(Self { links })
856 }
857
858 pub fn evidence_fingerprint(&self, evidence_uuid: Uuid) -> Result<[u8; 32], KnowledgeError> {
860 let row = self
861 .links
862 .iter()
863 .find(|row| row.evidence_uuid == evidence_uuid)
864 .ok_or(KnowledgeError::Dangling("evidence_uuid"))?;
865 evidence_fingerprint_for_row(row)
866 }
867
868 pub fn evidence_fingerprints(&self) -> Result<Vec<(Uuid, [u8; 32])>, KnowledgeError> {
870 self.links
871 .iter()
872 .map(|row| Ok((row.evidence_uuid, evidence_fingerprint_for_row(row)?)))
873 .collect()
874 }
875
876 pub fn batch(&self) -> Result<RecordBatch, KnowledgeError> {
878 evidence_batch(&self.links)
879 }
880
881 pub fn from_batches(batches: &[RecordBatch]) -> Result<Self, KnowledgeError> {
883 let mut links = Vec::new();
884 for batch in batches {
885 require_schema(batch, &EVIDENCE_LINK_SCHEMA, "evidence.schema")?;
886 let ids = fixed_column(batch, "evidence_uuid")?;
887 let assertions = fixed_column(batch, "assertion_uuid")?;
888 let sources = fixed_column(batch, "source_uuid")?;
889 let kinds = string_column(batch, "source_kind")?;
890 let roles = string_column(batch, "role")?;
891 let weights = f64_column(batch, "weight")?;
892 let provenance = fixed_column(batch, "provenance_uuid")?;
893 let recorded = timestamp_column(batch, "recorded_at")?;
894 let versions = u32_column(batch, "contract_version")?;
895 for row in 0..batch.num_rows() {
896 links.push(EvidenceLink {
897 evidence_uuid: uuid_at(ids, row, "evidence_uuid")?,
898 assertion_uuid: uuid_at(assertions, row, "assertion_uuid")?,
899 source_uuid: uuid_at(sources, row, "source_uuid")?,
900 source_kind: EvidenceSourceKind::parse(required_text(
901 kinds,
902 row,
903 "source_kind",
904 )?)?,
905 role: EvidenceRole::parse(required_text(roles, row, "role")?)?,
906 weight: optional_f64(weights, row),
907 provenance_uuid: uuid_at(provenance, row, "provenance_uuid")?,
908 recorded_at_micros: required_i64(recorded, row, "recorded_at")?,
909 contract_version: required_u32(versions, row, "contract_version")?,
910 });
911 }
912 }
913 Self::new(links)
914 }
915}
916
917#[derive(Clone, Debug)]
919pub struct SchemaRegistryEntry {
920 pub capability_id: &'static str,
922 pub capability_version: u32,
924 pub record_family: &'static str,
926 pub record_version: u32,
928 pub schema: SchemaRef,
930 pub schema_fingerprint: [u8; 32],
932 pub enum_registry_versions: &'static [(&'static str, u32)],
934 pub sort_key: &'static [&'static str],
936 pub diff_identity_fields: &'static [&'static str],
938 pub diff_record_uuid_field: Option<&'static str>,
940 pub fingerprint_domain: CanonicalDomain,
942 pub owner: &'static str,
944 pub implementation_issue: u64,
946 pub max_rows: usize,
948}
949
950impl SchemaRegistryEntry {
951 #[must_use]
953 pub const fn diff_identity_fingerprint_domain(&self) -> CanonicalDomain {
954 CanonicalDomain::ArrowResult
955 }
956
957 #[must_use]
959 pub const fn diff_record_fingerprint_domain(&self) -> CanonicalDomain {
960 self.fingerprint_domain
961 }
962}
963
964#[must_use]
966pub fn schema_registry() -> Vec<SchemaRegistryEntry> {
967 let mut entries = base_schema_registry_entries();
968 entries.extend(algorithm_run_schema_entries());
969 entries.push(reasoning::schema_registry_entry());
970 entries.push(status::schema_registry_entry());
971 entries.push(supersession::schema_registry_entry());
972 entries.extend(hypothesis::schema_registry_entries());
973 entries.push(valid_time::schema_registry_entry());
974 entries.push(belief_projection::schema_registry_entry());
975 entries.push(source::schema_registry_entry());
976 entries.push(artifact::schema_registry_entry());
977 entries.push(derivation::schema_registry_entry());
978 entries.push(artifact_preference::schema_registry_entry());
979 entries.push(retention_dependency::schema_registry_entry());
980 entries.extend(research::schema_registry_entries());
981 entries
982}
983
984fn base_schema_registry_entries() -> Vec<SchemaRegistryEntry> {
985 vec![
986 SchemaRegistryEntry {
987 capability_id: "knowledge",
988 capability_version: KNOWLEDGE_CAPABILITY_VERSION,
989 record_family: "assertions",
990 record_version: ASSERTION_CONTRACT_VERSION,
991 schema: Arc::clone(&ASSERTION_SCHEMA),
992 schema_fingerprint: *ASSERTION_SCHEMA_FINGERPRINT,
993 enum_registry_versions: &[],
994 sort_key: &["recorded_at", "assertion_uuid"],
995 diff_identity_fields: &["assertion_uuid"],
996 diff_record_uuid_field: Some("assertion_uuid"),
997 fingerprint_domain: CanonicalDomain::Assertion,
998 owner: "graphforge-knowledge",
999 implementation_issue: 2411,
1000 max_rows: MAX_KNOWLEDGE_ROWS,
1001 },
1002 SchemaRegistryEntry {
1003 capability_id: "knowledge",
1004 capability_version: KNOWLEDGE_CAPABILITY_VERSION,
1005 record_family: "assertion_graph_refs",
1006 record_version: ASSERTION_GRAPH_REF_CONTRACT_VERSION,
1007 schema: Arc::clone(&ASSERTION_GRAPH_REF_SCHEMA),
1008 schema_fingerprint: *ASSERTION_GRAPH_REF_SCHEMA_FINGERPRINT,
1009 enum_registry_versions: &[
1010 ("graph_kind", GRAPH_OBJECT_KIND_REGISTRY_VERSION),
1011 ("role", ASSERTION_GRAPH_ROLE_REGISTRY_VERSION),
1012 ],
1013 sort_key: &[
1014 "assertion_uuid",
1015 "role",
1016 "ordinal",
1017 "graph_kind",
1018 "graph_uuid",
1019 ],
1020 diff_identity_fields: &["assertion_uuid", "graph_uuid", "role", "ordinal"],
1021 diff_record_uuid_field: None,
1022 fingerprint_domain: CanonicalDomain::Assertion,
1023 owner: "graphforge-knowledge",
1024 implementation_issue: 2411,
1025 max_rows: MAX_KNOWLEDGE_ROWS,
1026 },
1027 SchemaRegistryEntry {
1028 capability_id: "knowledge",
1029 capability_version: KNOWLEDGE_CAPABILITY_VERSION,
1030 record_family: "confidence_assessments",
1031 record_version: CONFIDENCE_ASSESSMENT_CONTRACT_VERSION,
1032 schema: Arc::clone(&CONFIDENCE_ASSESSMENT_SCHEMA),
1033 schema_fingerprint: *CONFIDENCE_ASSESSMENT_SCHEMA_FINGERPRINT,
1034 enum_registry_versions: &[("confidence_policy", CONFIDENCE_POLICY_REGISTRY_VERSION)],
1035 sort_key: &["recorded_at", "confidence_uuid"],
1036 diff_identity_fields: &["confidence_uuid"],
1037 diff_record_uuid_field: Some("confidence_uuid"),
1038 fingerprint_domain: CanonicalDomain::ConfidenceAssessment,
1039 owner: "graphforge-knowledge",
1040 implementation_issue: 774,
1041 max_rows: MAX_KNOWLEDGE_ROWS,
1042 },
1043 SchemaRegistryEntry {
1044 capability_id: "knowledge",
1045 capability_version: KNOWLEDGE_CAPABILITY_VERSION,
1046 record_family: "confidence_inputs",
1047 record_version: CONFIDENCE_INPUT_CONTRACT_VERSION,
1048 schema: Arc::clone(&CONFIDENCE_INPUT_SCHEMA),
1049 schema_fingerprint: *CONFIDENCE_INPUT_SCHEMA_FINGERPRINT,
1050 enum_registry_versions: &[],
1051 sort_key: &["confidence_uuid", "ordinal", "input_confidence_uuid"],
1052 diff_identity_fields: &["confidence_uuid", "input_confidence_uuid"],
1053 diff_record_uuid_field: None,
1054 fingerprint_domain: CanonicalDomain::ConfidenceAssessment,
1055 owner: "graphforge-knowledge",
1056 implementation_issue: 774,
1057 max_rows: MAX_KNOWLEDGE_ROWS,
1058 },
1059 SchemaRegistryEntry {
1060 capability_id: "knowledge",
1061 capability_version: KNOWLEDGE_CAPABILITY_VERSION,
1062 record_family: "evidence",
1063 record_version: EVIDENCE_LINK_CONTRACT_VERSION,
1064 schema: Arc::clone(&EVIDENCE_LINK_SCHEMA),
1065 schema_fingerprint: *EVIDENCE_LINK_SCHEMA_FINGERPRINT,
1066 enum_registry_versions: &[
1067 (
1068 "evidence_source_kind",
1069 EVIDENCE_SOURCE_KIND_REGISTRY_VERSION,
1070 ),
1071 ("evidence_role", EVIDENCE_ROLE_REGISTRY_VERSION),
1072 ],
1073 sort_key: &["recorded_at", "evidence_uuid"],
1074 diff_identity_fields: &["evidence_uuid"],
1075 diff_record_uuid_field: Some("evidence_uuid"),
1076 fingerprint_domain: CanonicalDomain::EvidenceLink,
1077 owner: "graphforge-knowledge",
1078 implementation_issue: 775,
1079 max_rows: MAX_KNOWLEDGE_ROWS,
1080 },
1081 ]
1082}
1083
1084fn algorithm_run_schema_entries() -> [SchemaRegistryEntry; 2] {
1085 [
1086 SchemaRegistryEntry {
1087 capability_id: "knowledge",
1088 capability_version: KNOWLEDGE_CAPABILITY_VERSION,
1089 record_family: "algorithm_runs",
1090 record_version: ALGORITHM_RUN_CONTRACT_VERSION,
1091 schema: Arc::clone(&ALGORITHM_RUN_SCHEMA),
1092 schema_fingerprint: *ALGORITHM_RUN_SCHEMA_FINGERPRINT,
1093 enum_registry_versions: &[],
1094 sort_key: &["started_at", "run_uuid"],
1095 diff_identity_fields: &["run_uuid"],
1096 diff_record_uuid_field: Some("run_uuid"),
1097 fingerprint_domain: CanonicalDomain::InvocationDescriptor,
1098 owner: "graphforge-knowledge",
1099 implementation_issue: 2003,
1100 max_rows: MAX_KNOWLEDGE_ROWS,
1101 },
1102 SchemaRegistryEntry {
1103 capability_id: "knowledge",
1104 capability_version: KNOWLEDGE_CAPABILITY_VERSION,
1105 record_family: "algorithm_run_events",
1106 record_version: ALGORITHM_RUN_EVENT_CONTRACT_VERSION,
1107 schema: Arc::clone(&ALGORITHM_RUN_EVENT_SCHEMA),
1108 schema_fingerprint: *ALGORITHM_RUN_EVENT_SCHEMA_FINGERPRINT,
1109 enum_registry_versions: &[(
1110 "algorithm_run_state",
1111 ALGORITHM_RUN_STATE_REGISTRY_VERSION,
1112 )],
1113 sort_key: &["recorded_at", "event_uuid"],
1114 diff_identity_fields: &["event_uuid"],
1115 diff_record_uuid_field: Some("event_uuid"),
1116 fingerprint_domain: CanonicalDomain::ArrowResult,
1117 owner: "graphforge-knowledge",
1118 implementation_issue: 2003,
1119 max_rows: MAX_KNOWLEDGE_ROWS,
1120 },
1121 ]
1122}
1123
1124#[derive(thiserror::Error, Debug)]
1126pub enum KnowledgeError {
1127 #[error("invalid knowledge {field}: {message}")]
1129 Invalid {
1130 field: &'static str,
1132 message: &'static str,
1134 },
1135 #[error("knowledge {participant} row limit exceeded: observed {observed}, limit {limit}")]
1137 Limit {
1138 participant: &'static str,
1140 observed: usize,
1142 limit: usize,
1144 },
1145 #[error("duplicate knowledge identity: {0}")]
1147 Duplicate(&'static str),
1148 #[error("dangling knowledge reference: {0}")]
1150 Dangling(&'static str),
1151 #[error("knowledge idempotency conflict: {0}")]
1153 Conflict(&'static str),
1154 #[error("knowledge transaction conflict: {0}")]
1156 TransactionConflict(&'static str),
1157 #[error(transparent)]
1159 Canonical(#[from] CanonicalError),
1160 #[error("knowledge Arrow failure: {0}")]
1162 Arrow(#[from] arrow::error::ArrowError),
1163}
1164
1165impl KnowledgeError {
1166 #[must_use]
1168 pub const fn code(&self) -> &'static str {
1169 match self {
1170 Self::Invalid { .. } => "GF_KNOWLEDGE_INVALID",
1171 Self::Limit { .. } => "GF_RESOURCE_LIMIT",
1172 Self::Duplicate(_) => "GF_KNOWLEDGE_DUPLICATE",
1173 Self::Dangling(_) => "GF_KNOWLEDGE_DANGLING",
1174 Self::Conflict(_) => "GF_IDEMPOTENCY_CONFLICT",
1175 Self::TransactionConflict(_) => "GF_TRANSACTION_CONFLICT",
1176 Self::Canonical(error) => error.code(),
1177 Self::Arrow(_) => "GF_SCHEMA_MISMATCH",
1178 }
1179 }
1180}
1181
1182fn validate_rows(
1183 assertions: &[Assertion],
1184 refs: &[AssertionGraphRef],
1185) -> Result<(), KnowledgeError> {
1186 validate_rows_with_limit(assertions, refs, MAX_KNOWLEDGE_ROWS)
1187}
1188
1189fn validate_rows_with_limit(
1190 assertions: &[Assertion],
1191 refs: &[AssertionGraphRef],
1192 row_limit: usize,
1193) -> Result<(), KnowledgeError> {
1194 validate_assertion_row_limits(assertions.len(), refs.len(), row_limit)?;
1195 let mut assertion_ids = HashSet::with_capacity(assertions.len());
1196 for assertion in assertions {
1197 #[cfg(test)]
1198 ASSERTION_ROW_VALIDATION_COUNT.with(|count| count.set(count.get() + 1));
1199 if assertion.contract_version != ASSERTION_CONTRACT_VERSION {
1200 return Err(invalid("assertion.contract_version", "unsupported version"));
1201 }
1202 require_v7(assertion.assertion_uuid, "assertion_uuid")?;
1203 require_uuid(assertion.provenance_uuid, "provenance_uuid")?;
1204 validate_claim(&assertion.claim)?;
1205 if !assertion_ids.insert(assertion.assertion_uuid) {
1206 return Err(KnowledgeError::Duplicate("assertion_uuid"));
1207 }
1208 }
1209 let mut tuples = HashSet::with_capacity(refs.len());
1210 let mut role_ordinals: HashMap<(Uuid, AssertionGraphRole), Vec<u32>> = HashMap::new();
1211 let mut covered_assertions = HashSet::with_capacity(assertions.len());
1212 for reference in refs {
1213 #[cfg(test)]
1214 ASSERTION_ROW_VALIDATION_COUNT.with(|count| count.set(count.get() + 1));
1215 if reference.contract_version != ASSERTION_GRAPH_REF_CONTRACT_VERSION {
1216 return Err(invalid(
1217 "assertion_graph_ref.contract_version",
1218 "unsupported version",
1219 ));
1220 }
1221 require_v7(reference.assertion_uuid, "assertion_uuid")?;
1222 require_uuid(reference.graph_uuid, "graph_uuid")?;
1223 if !assertion_ids.contains(&reference.assertion_uuid) {
1224 return Err(KnowledgeError::Dangling("assertion_uuid"));
1225 }
1226 if !tuples.insert((
1227 reference.assertion_uuid,
1228 reference.graph_uuid,
1229 reference.role,
1230 reference.ordinal,
1231 )) {
1232 return Err(KnowledgeError::Duplicate(
1233 "assertion_uuid/graph_uuid/role/ordinal",
1234 ));
1235 }
1236 role_ordinals
1237 .entry((reference.assertion_uuid, reference.role))
1238 .or_default()
1239 .push(reference.ordinal);
1240 covered_assertions.insert(reference.assertion_uuid);
1241 }
1242 for assertion_uuid in assertion_ids {
1243 if !covered_assertions.contains(&assertion_uuid) {
1244 return Err(KnowledgeError::Dangling("assertion.graph_refs"));
1245 }
1246 }
1247 for ordinals in role_ordinals.values_mut() {
1248 ordinals.sort_unstable();
1249 if ordinals
1250 .iter()
1251 .enumerate()
1252 .any(|(expected, actual)| usize::try_from(*actual) != Ok(expected))
1253 {
1254 return Err(invalid("ordinal", "must be contiguous from zero per role"));
1255 }
1256 }
1257 Ok(())
1258}
1259
1260#[cfg(test)]
1261fn reset_assertion_row_validation_count() {
1262 ASSERTION_ROW_VALIDATION_COUNT.with(|count| count.set(0));
1263}
1264
1265#[cfg(test)]
1266fn assertion_row_validation_count() -> usize {
1267 ASSERTION_ROW_VALIDATION_COUNT.with(std::cell::Cell::get)
1268}
1269
1270fn check_assertion_row_limit(
1271 participant: &'static str,
1272 observed: usize,
1273 limit: usize,
1274) -> Result<(), KnowledgeError> {
1275 if observed > limit {
1276 Err(KnowledgeError::Limit {
1277 participant,
1278 observed,
1279 limit,
1280 })
1281 } else {
1282 Ok(())
1283 }
1284}
1285
1286fn validate_assertion_row_limits(
1287 assertion_count: usize,
1288 graph_ref_count: usize,
1289 limit: usize,
1290) -> Result<(), KnowledgeError> {
1291 check_assertion_row_limit("assertions", assertion_count, limit)?;
1292 check_assertion_row_limit("assertion_graph_refs", graph_ref_count, limit)
1293}
1294
1295fn sort_assertion_rows(assertions: &mut [Assertion], refs: &mut [AssertionGraphRef]) {
1296 let times = assertions
1297 .iter()
1298 .map(|row| (row.assertion_uuid, row.recorded_at_micros))
1299 .collect::<HashMap<_, _>>();
1300 assertions.sort_by_key(|row| (row.recorded_at_micros, row.assertion_uuid));
1301 refs.sort_by_key(|row| {
1302 (
1303 times[&row.assertion_uuid],
1304 row.assertion_uuid,
1305 role_order(row.role),
1306 row.ordinal,
1307 kind_order(row.graph_kind),
1308 row.graph_uuid,
1309 )
1310 });
1311}
1312
1313fn assertion_batch(rows: &[Assertion]) -> Result<RecordBatch, KnowledgeError> {
1314 let mut ids = FixedSizeBinaryBuilder::with_capacity(rows.len(), 16);
1315 let mut provenance = FixedSizeBinaryBuilder::with_capacity(rows.len(), 16);
1316 for row in rows {
1317 ids.append_value(row.assertion_uuid.as_bytes())?;
1318 provenance.append_value(row.provenance_uuid.as_bytes())?;
1319 }
1320 RecordBatch::try_new(
1321 Arc::clone(&ASSERTION_SCHEMA),
1322 vec![
1323 Arc::new(ids.finish()),
1324 Arc::new(StringArray::from_iter_values(
1325 rows.iter().map(|row| row.claim.as_str()),
1326 )),
1327 Arc::new(provenance.finish()),
1328 Arc::new(
1329 TimestampMicrosecondArray::from_iter_values(
1330 rows.iter().map(|row| row.recorded_at_micros),
1331 )
1332 .with_timezone("UTC"),
1333 ),
1334 Arc::new(UInt32Array::from_iter_values(
1335 rows.iter().map(|row| row.contract_version),
1336 )),
1337 ],
1338 )
1339 .map_err(Into::into)
1340}
1341
1342fn graph_ref_batch(rows: &[AssertionGraphRef]) -> Result<RecordBatch, KnowledgeError> {
1343 let mut assertions = FixedSizeBinaryBuilder::with_capacity(rows.len(), 16);
1344 let mut graph_ids = FixedSizeBinaryBuilder::with_capacity(rows.len(), 16);
1345 for row in rows {
1346 assertions.append_value(row.assertion_uuid.as_bytes())?;
1347 graph_ids.append_value(row.graph_uuid.as_bytes())?;
1348 }
1349 RecordBatch::try_new(
1350 Arc::clone(&ASSERTION_GRAPH_REF_SCHEMA),
1351 vec![
1352 Arc::new(assertions.finish()),
1353 Arc::new(graph_ids.finish()),
1354 Arc::new(StringArray::from_iter_values(
1355 rows.iter().map(|row| row.graph_kind.as_str()),
1356 )),
1357 Arc::new(StringArray::from_iter_values(
1358 rows.iter().map(|row| row.role.as_str()),
1359 )),
1360 Arc::new(UInt32Array::from_iter_values(
1361 rows.iter().map(|row| row.ordinal),
1362 )),
1363 Arc::new(UInt32Array::from_iter_values(
1364 rows.iter().map(|row| row.contract_version),
1365 )),
1366 ],
1367 )
1368 .map_err(Into::into)
1369}
1370
1371fn evidence_batch(rows: &[EvidenceLink]) -> Result<RecordBatch, KnowledgeError> {
1372 let mut ids = FixedSizeBinaryBuilder::with_capacity(rows.len(), 16);
1373 let mut assertions = FixedSizeBinaryBuilder::with_capacity(rows.len(), 16);
1374 let mut sources = FixedSizeBinaryBuilder::with_capacity(rows.len(), 16);
1375 let mut provenance = FixedSizeBinaryBuilder::with_capacity(rows.len(), 16);
1376 for row in rows {
1377 ids.append_value(row.evidence_uuid.as_bytes())?;
1378 assertions.append_value(row.assertion_uuid.as_bytes())?;
1379 sources.append_value(row.source_uuid.as_bytes())?;
1380 provenance.append_value(row.provenance_uuid.as_bytes())?;
1381 }
1382 RecordBatch::try_new(
1383 Arc::clone(&EVIDENCE_LINK_SCHEMA),
1384 vec![
1385 Arc::new(ids.finish()),
1386 Arc::new(assertions.finish()),
1387 Arc::new(sources.finish()),
1388 Arc::new(StringArray::from_iter_values(
1389 rows.iter().map(|row| row.source_kind.as_str()),
1390 )),
1391 Arc::new(StringArray::from_iter_values(
1392 rows.iter().map(|row| row.role.as_str()),
1393 )),
1394 Arc::new(rows.iter().map(|row| row.weight).collect::<Float64Array>()),
1395 Arc::new(provenance.finish()),
1396 Arc::new(
1397 TimestampMicrosecondArray::from_iter_values(
1398 rows.iter().map(|row| row.recorded_at_micros),
1399 )
1400 .with_timezone("UTC"),
1401 ),
1402 Arc::new(UInt32Array::from_iter_values(
1403 rows.iter().map(|row| row.contract_version),
1404 )),
1405 ],
1406 )
1407 .map_err(Into::into)
1408}
1409
1410fn validate_confidence(value: Option<f64>, field: &'static str) -> Result<(), KnowledgeError> {
1411 if value.is_some_and(|value| !value.is_finite() || !(0.0..=1.0).contains(&value)) {
1412 Err(invalid(field, "must be finite and in [0,1]"))
1413 } else {
1414 Ok(())
1415 }
1416}
1417
1418fn normalize_zero(value: f64) -> f64 {
1419 if value == 0.0 { 0.0 } else { value }
1420}
1421
1422fn canonical_optional_f64(
1423 writer: &mut CanonicalWriter,
1424 value: Option<f64>,
1425) -> Result<(), KnowledgeError> {
1426 match value {
1427 None => writer.u8(0)?,
1428 Some(value) => {
1429 writer.u8(1)?;
1430 writer.u64(normalize_zero(value).to_bits())?;
1431 }
1432 }
1433 Ok(())
1434}
1435
1436fn refs_for(rows: &[AssertionGraphRef], assertion_uuid: Uuid) -> Vec<&AssertionGraphRef> {
1437 let mut refs = rows
1438 .iter()
1439 .filter(|row| row.assertion_uuid == assertion_uuid)
1440 .collect::<Vec<_>>();
1441 refs.sort_by_key(|row| {
1442 (
1443 role_order(row.role),
1444 row.ordinal,
1445 kind_order(row.graph_kind),
1446 row.graph_uuid,
1447 )
1448 });
1449 refs
1450}
1451
1452fn assertion_fingerprint_for_row(
1453 assertion: &Assertion,
1454 refs: &[&AssertionGraphRef],
1455) -> Result<[u8; 32], KnowledgeError> {
1456 let mut writer = CanonicalWriter::new();
1457 writer.raw(b"GFAS")?;
1458 writer.u32(ASSERTION_CONTRACT_VERSION)?;
1459 writer.text(&assertion.claim)?;
1460 writer.u64(
1461 u64::try_from(refs.len()).map_err(|_| KnowledgeError::Limit {
1462 participant: "assertion_graph_refs",
1463 observed: refs.len(),
1464 limit: MAX_KNOWLEDGE_ROWS,
1465 })?,
1466 )?;
1467 for reference in refs {
1468 writer.text(reference.role.as_str())?;
1469 writer.u32(reference.ordinal)?;
1470 writer.text(reference.graph_kind.as_str())?;
1471 writer.raw(reference.graph_uuid.as_bytes())?;
1472 }
1473 Ok(fingerprint(
1474 CanonicalDomain::Assertion,
1475 CANONICAL_CONTRACT_VERSION,
1476 &writer.finish(),
1477 )?)
1478}
1479
1480fn evidence_fingerprint_for_row(row: &EvidenceLink) -> Result<[u8; 32], KnowledgeError> {
1481 let mut writer = CanonicalWriter::new();
1482 writer.raw(b"GFEV")?;
1483 writer.u32(EVIDENCE_LINK_CONTRACT_VERSION)?;
1484 writer.raw(row.assertion_uuid.as_bytes())?;
1485 writer.raw(row.source_uuid.as_bytes())?;
1486 writer.text(row.source_kind.as_str())?;
1487 writer.text(row.role.as_str())?;
1488 canonical_optional_f64(&mut writer, row.weight)?;
1489 Ok(fingerprint(
1490 CanonicalDomain::EvidenceLink,
1491 CANONICAL_CONTRACT_VERSION,
1492 &writer.finish(),
1493 )?)
1494}
1495
1496const fn role_order(role: AssertionGraphRole) -> u8 {
1497 match role {
1498 AssertionGraphRole::Subject => 0,
1499 AssertionGraphRole::Object => 1,
1500 AssertionGraphRole::Context => 2,
1501 }
1502}
1503
1504const fn kind_order(kind: GraphObjectKind) -> u8 {
1505 match kind {
1506 GraphObjectKind::Node => 0,
1507 GraphObjectKind::Edge => 1,
1508 }
1509}
1510
1511fn validate_claim(claim: &str) -> Result<(), KnowledgeError> {
1512 if claim.is_empty() {
1513 return Err(invalid("claim", "must not be empty"));
1514 }
1515 if claim.len() as u64 > graphforge_core::canonical::MAX_CANONICAL_TEXT_BYTES {
1516 return Err(invalid("claim", "exceeds canonical UTF-8 limit"));
1517 }
1518 Ok(())
1519}
1520
1521fn require_uuid(value: Uuid, field: &'static str) -> Result<(), KnowledgeError> {
1522 if value.is_nil() {
1523 Err(invalid(field, "must not be nil"))
1524 } else {
1525 Ok(())
1526 }
1527}
1528
1529fn require_v7(value: Uuid, field: &'static str) -> Result<(), KnowledgeError> {
1530 require_uuid(value, field)?;
1531 if value.get_version() != Some(Version::SortRand) {
1532 return Err(invalid(field, "must be UUIDv7"));
1533 }
1534 Ok(())
1535}
1536
1537fn check_limit(participant: &'static str, observed: usize) -> Result<(), KnowledgeError> {
1538 if observed > MAX_KNOWLEDGE_ROWS {
1539 Err(KnowledgeError::Limit {
1540 participant,
1541 observed,
1542 limit: MAX_KNOWLEDGE_ROWS,
1543 })
1544 } else {
1545 Ok(())
1546 }
1547}
1548
1549const fn invalid(field: &'static str, message: &'static str) -> KnowledgeError {
1550 KnowledgeError::Invalid { field, message }
1551}
1552
1553fn require_schema(
1554 batch: &RecordBatch,
1555 expected: &SchemaRef,
1556 field: &'static str,
1557) -> Result<(), KnowledgeError> {
1558 if batch.schema().as_ref() == expected.as_ref() {
1559 Ok(())
1560 } else {
1561 Err(invalid(field, "schema mismatch"))
1562 }
1563}
1564
1565fn fixed_column<'a>(
1566 batch: &'a RecordBatch,
1567 name: &'static str,
1568) -> Result<&'a FixedSizeBinaryArray, KnowledgeError> {
1569 batch
1570 .column_by_name(name)
1571 .and_then(|column| column.as_any().downcast_ref())
1572 .ok_or_else(|| invalid(name, "missing or wrong Arrow type"))
1573}
1574
1575fn string_column<'a>(
1576 batch: &'a RecordBatch,
1577 name: &'static str,
1578) -> Result<&'a StringArray, KnowledgeError> {
1579 batch
1580 .column_by_name(name)
1581 .and_then(|column| column.as_any().downcast_ref())
1582 .ok_or_else(|| invalid(name, "missing or wrong Arrow type"))
1583}
1584
1585fn binary_column<'a>(
1586 batch: &'a RecordBatch,
1587 name: &'static str,
1588) -> Result<&'a BinaryArray, KnowledgeError> {
1589 batch
1590 .column_by_name(name)
1591 .and_then(|column| column.as_any().downcast_ref())
1592 .ok_or_else(|| invalid(name, "missing or wrong Arrow type"))
1593}
1594
1595fn timestamp_column<'a>(
1596 batch: &'a RecordBatch,
1597 name: &'static str,
1598) -> Result<&'a TimestampMicrosecondArray, KnowledgeError> {
1599 batch
1600 .column_by_name(name)
1601 .and_then(|column| column.as_any().downcast_ref())
1602 .ok_or_else(|| invalid(name, "missing or wrong Arrow type"))
1603}
1604
1605fn u32_column<'a>(
1606 batch: &'a RecordBatch,
1607 name: &'static str,
1608) -> Result<&'a UInt32Array, KnowledgeError> {
1609 batch
1610 .column_by_name(name)
1611 .and_then(|column| column.as_any().downcast_ref())
1612 .ok_or_else(|| invalid(name, "missing or wrong Arrow type"))
1613}
1614
1615fn f64_column<'a>(
1616 batch: &'a RecordBatch,
1617 name: &'static str,
1618) -> Result<&'a Float64Array, KnowledgeError> {
1619 batch
1620 .column_by_name(name)
1621 .and_then(|column| column.as_any().downcast_ref())
1622 .ok_or_else(|| invalid(name, "missing or wrong Arrow type"))
1623}
1624
1625fn optional_f64(array: &Float64Array, row: usize) -> Option<f64> {
1626 (!array.is_null(row)).then(|| normalize_zero(array.value(row)))
1627}
1628
1629fn uuid_at(
1630 array: &FixedSizeBinaryArray,
1631 row: usize,
1632 field: &'static str,
1633) -> Result<Uuid, KnowledgeError> {
1634 if array.is_null(row) {
1635 return Err(invalid(field, "must not be null"));
1636 }
1637 Uuid::from_slice(array.value(row)).map_err(|_| invalid(field, "malformed UUID"))
1638}
1639
1640fn fixed_32_at(
1641 array: &FixedSizeBinaryArray,
1642 row: usize,
1643 field: &'static str,
1644) -> Result<[u8; 32], KnowledgeError> {
1645 if array.is_null(row) || array.value_length() != 32 {
1646 return Err(invalid(field, "must be a 32-byte value"));
1647 }
1648 Ok(array
1649 .value(row)
1650 .try_into()
1651 .expect("validated fixed-size binary width"))
1652}
1653
1654fn optional_fixed_16(
1655 array: &FixedSizeBinaryArray,
1656 row: usize,
1657 field: &'static str,
1658) -> Result<Option<Uuid>, KnowledgeError> {
1659 if array.is_null(row) {
1660 Ok(None)
1661 } else {
1662 uuid_at(array, row, field).map(Some)
1663 }
1664}
1665
1666fn optional_fixed_32(
1667 array: &FixedSizeBinaryArray,
1668 row: usize,
1669 field: &'static str,
1670) -> Result<Option<[u8; 32]>, KnowledgeError> {
1671 if array.is_null(row) {
1672 Ok(None)
1673 } else {
1674 fixed_32_at(array, row, field).map(Some)
1675 }
1676}
1677
1678fn required_text<'a>(
1679 array: &'a StringArray,
1680 row: usize,
1681 field: &'static str,
1682) -> Result<&'a str, KnowledgeError> {
1683 if array.is_null(row) {
1684 Err(invalid(field, "must not be null"))
1685 } else {
1686 Ok(array.value(row))
1687 }
1688}
1689
1690fn optional_text(array: &StringArray, row: usize) -> Option<String> {
1691 (!array.is_null(row)).then(|| array.value(row).to_owned())
1692}
1693
1694fn required_binary<'a>(
1695 array: &'a BinaryArray,
1696 row: usize,
1697 field: &'static str,
1698) -> Result<&'a [u8], KnowledgeError> {
1699 if array.is_null(row) {
1700 Err(invalid(field, "must not be null"))
1701 } else {
1702 Ok(array.value(row))
1703 }
1704}
1705
1706fn required_i64(
1707 array: &TimestampMicrosecondArray,
1708 row: usize,
1709 field: &'static str,
1710) -> Result<i64, KnowledgeError> {
1711 if array.is_null(row) {
1712 Err(invalid(field, "must not be null"))
1713 } else {
1714 Ok(array.value(row))
1715 }
1716}
1717
1718fn required_u32(
1719 array: &UInt32Array,
1720 row: usize,
1721 field: &'static str,
1722) -> Result<u32, KnowledgeError> {
1723 if array.is_null(row) {
1724 Err(invalid(field, "must not be null"))
1725 } else {
1726 Ok(array.value(row))
1727 }
1728}
1729
1730#[cfg(test)]
1731mod tests {
1732 #[path = "assertion_mutation_tests.rs"]
1733 mod assertion_mutation_tests;
1734
1735 use super::*;
1736
1737 pub(super) fn uuid7(seed: u8) -> Uuid {
1738 let mut bytes = [seed; 16];
1739 bytes[6] = (bytes[6] & 0x0f) | 0x70;
1740 bytes[8] = (bytes[8] & 0x3f) | 0x80;
1741 Uuid::from_bytes(bytes)
1742 }
1743
1744 fn non_v7_uuid(seed: u8) -> Uuid {
1745 let mut bytes = [seed; 16];
1746 bytes[6] = (bytes[6] & 0x0f) | 0x40;
1747 bytes[8] = (bytes[8] & 0x3f) | 0x80;
1748 Uuid::from_bytes(bytes)
1749 }
1750
1751 fn fixture() -> AssertionLedger {
1752 let assertion_uuid = uuid7(1);
1753 AssertionLedger::new(
1754 vec![Assertion::new(assertion_uuid, "exact claim".into(), uuid7(2), 10).unwrap()],
1755 vec![
1756 AssertionGraphRef::new(
1757 assertion_uuid,
1758 uuid7(3),
1759 GraphObjectKind::Node,
1760 AssertionGraphRole::Subject,
1761 0,
1762 )
1763 .unwrap(),
1764 AssertionGraphRef::new(
1765 assertion_uuid,
1766 uuid7(4),
1767 GraphObjectKind::Edge,
1768 AssertionGraphRole::Context,
1769 0,
1770 )
1771 .unwrap(),
1772 ],
1773 )
1774 .unwrap()
1775 }
1776
1777 fn measurement_uuid(seed: u64) -> Uuid {
1778 let mut bytes = u128::from(seed).to_be_bytes();
1779 bytes[6] = (bytes[6] & 0x0f) | 0x70;
1780 bytes[8] = (bytes[8] & 0x3f) | 0x80;
1781 Uuid::from_bytes(bytes)
1782 }
1783
1784 fn assertion_measurement_digest(
1785 existing: &AssertionLedger,
1786 staged: &AssertionLedger,
1787 ) -> String {
1788 let mut writer = CanonicalWriter::new();
1789 writer.raw(b"GF_ASSERTION_MERGE_MEASUREMENT_V1").unwrap();
1790 for ledger in [existing, staged] {
1791 writer.u64(ledger.assertions.len() as u64).unwrap();
1792 for row in &ledger.assertions {
1793 writer.raw(row.assertion_uuid.as_bytes()).unwrap();
1794 writer.text(&row.claim).unwrap();
1795 writer.raw(row.provenance_uuid.as_bytes()).unwrap();
1796 writer.i64(row.recorded_at_micros).unwrap();
1797 writer.u32(row.contract_version).unwrap();
1798 }
1799 writer.u64(ledger.graph_refs.len() as u64).unwrap();
1800 for row in &ledger.graph_refs {
1801 writer.raw(row.assertion_uuid.as_bytes()).unwrap();
1802 writer.raw(row.graph_uuid.as_bytes()).unwrap();
1803 writer.text(row.graph_kind.as_str()).unwrap();
1804 writer.text(row.role.as_str()).unwrap();
1805 writer.u32(row.ordinal).unwrap();
1806 writer.u32(row.contract_version).unwrap();
1807 }
1808 }
1809 fingerprint(
1810 CanonicalDomain::Assertion,
1811 CANONICAL_CONTRACT_VERSION,
1812 &writer.finish(),
1813 )
1814 .unwrap()
1815 .iter()
1816 .map(|byte| format!("{byte:02x}"))
1817 .collect()
1818 }
1819
1820 fn median_nanos(values: &[u128]) -> u128 {
1821 let mut sorted = values.to_vec();
1822 sorted.sort_unstable();
1823 sorted[sorted.len() / 2]
1824 }
1825
1826 #[test]
1827 #[ignore = "manual quiet-host assertion merge cost measurement"]
1831 fn quiet_host_assertion_merge_cost_measurement() {
1832 use std::time::Instant;
1833
1834 let repetitions = 9;
1835 for existing_count in [1_000_u64, 10_000] {
1836 let existing = AssertionLedger::new(
1837 (0..existing_count)
1838 .map(|seed| {
1839 Assertion::new(
1840 measurement_uuid(seed + 1),
1841 format!("assertion claim {seed}"),
1842 measurement_uuid(seed + 2_000_000),
1843 i64::try_from(seed).unwrap(),
1844 )
1845 .unwrap()
1846 })
1847 .collect(),
1848 (0..existing_count)
1849 .map(|seed| {
1850 AssertionGraphRef::new(
1851 measurement_uuid(seed + 1),
1852 measurement_uuid(seed + 4_000_000),
1853 GraphObjectKind::Node,
1854 AssertionGraphRole::Subject,
1855 0,
1856 )
1857 .unwrap()
1858 })
1859 .collect(),
1860 )
1861 .unwrap();
1862 let staged = AssertionLedger::new(
1863 vec![
1864 Assertion::new(
1865 measurement_uuid(existing_count + 1),
1866 "staged assertion claim".into(),
1867 measurement_uuid(6_000_000),
1868 i64::try_from(existing_count).unwrap(),
1869 )
1870 .unwrap(),
1871 ],
1872 vec![
1873 AssertionGraphRef::new(
1874 measurement_uuid(existing_count + 1),
1875 measurement_uuid(8_000_000),
1876 GraphObjectKind::Node,
1877 AssertionGraphRole::Subject,
1878 0,
1879 )
1880 .unwrap(),
1881 ],
1882 )
1883 .unwrap();
1884
1885 let baseline_merge = || {
1886 AssertionLedger::new(
1887 [existing.assertions.clone(), staged.assertions.clone()].concat(),
1888 [existing.graph_refs.clone(), staged.graph_refs.clone()].concat(),
1889 )
1890 .unwrap()
1891 };
1892 let trusted_merge = || existing.merge(&staged).unwrap();
1893 assert_eq!(baseline_merge(), trusted_merge());
1894 let input_digest = assertion_measurement_digest(&existing, &staged);
1895 let expected_input_digest = match existing_count {
1896 1_000 => "eac191d14d0f1088c9cf1b6f8e87df5e353b31dfa48de6490123cf177b2a0496",
1897 10_000 => "5553b15b4fc68dd53ddf0b071daa392bc7e04cb74d78b39178c5a7a102df1a03",
1898 _ => unreachable!("measurement uses two pinned input sizes"),
1899 };
1900 assert_eq!(input_digest, expected_input_digest);
1901 let mut baseline = Vec::with_capacity(repetitions);
1902 let mut optimized = Vec::with_capacity(repetitions);
1903 let mut deltas = Vec::with_capacity(repetitions);
1904 for repetition in 0..repetitions {
1905 let (baseline_ns, optimized_ns) = if repetition % 2 == 0 {
1906 let start = Instant::now();
1907 std::hint::black_box(baseline_merge());
1908 let baseline_ns = start.elapsed().as_nanos();
1909
1910 let start = Instant::now();
1911 std::hint::black_box(trusted_merge());
1912 (baseline_ns, start.elapsed().as_nanos())
1913 } else {
1914 let start = Instant::now();
1915 std::hint::black_box(trusted_merge());
1916 let optimized_ns = start.elapsed().as_nanos();
1917
1918 let start = Instant::now();
1919 std::hint::black_box(baseline_merge());
1920 (start.elapsed().as_nanos(), optimized_ns)
1921 };
1922 baseline.push(baseline_ns);
1923 optimized.push(optimized_ns);
1924 deltas.push(
1925 i128::try_from(optimized_ns).unwrap() - i128::try_from(baseline_ns).unwrap(),
1926 );
1927 eprintln!(
1928 "existing_assertions={existing_count} existing_refs={existing_count} staged_assertions=1 staged_refs=1 rep={} order={} baseline_ns={baseline_ns} trusted_merge_ns={optimized_ns} delta_ns={}",
1929 repetition + 1,
1930 if repetition % 2 == 0 {
1931 "baseline-first"
1932 } else {
1933 "trusted-first"
1934 },
1935 optimized_ns as i128 - baseline_ns as i128,
1936 );
1937 }
1938 let baseline_median = median_nanos(&baseline);
1939 let optimized_median = median_nanos(&optimized);
1940 let mut sorted_deltas = deltas.clone();
1941 sorted_deltas.sort_unstable();
1942 let existing_rows = (existing.assertions.len() + existing.graph_refs.len()) as f64;
1943 eprintln!(
1944 "summary existing_assertions={existing_count} existing_refs={existing_count} existing_rows={} staged_assertions=1 staged_refs=1 repetitions={repetitions} input_digest_sha256={input_digest} baseline_median_ns={baseline_median} baseline_ns_per_existing_row={:.3} trusted_merge_median_ns={optimized_median} trusted_merge_ns_per_existing_row={:.3} median_delta_ns={} delta_range_ns={}..{}",
1945 existing_rows as usize,
1946 baseline_median as f64 / existing_rows,
1947 optimized_median as f64 / existing_rows,
1948 optimized_median as i128 - baseline_median as i128,
1949 sorted_deltas[0],
1950 sorted_deltas[sorted_deltas.len() - 1],
1951 );
1952 }
1953 }
1954
1955 #[test]
1956 fn exact_claim_and_sorted_refs_have_stable_fingerprint() {
1957 let first = fixture();
1958 let mut reversed = first.graph_refs.clone();
1959 reversed.reverse();
1960 let second = AssertionLedger::new(first.assertions.clone(), reversed).unwrap();
1961 let fingerprint = first.assertion_fingerprint(uuid7(1)).unwrap();
1962 assert_eq!(fingerprint, second.assertion_fingerprint(uuid7(1)).unwrap());
1963 let encoded = fingerprint
1964 .iter()
1965 .map(|byte| format!("{byte:02x}"))
1966 .collect::<String>();
1967 assert_eq!(
1970 encoded,
1971 "6024d4f2e22b35d5850f6a2f1c9f0cf77d28ab1e7ba83342416a1e7660f7b4bb"
1972 );
1973 }
1974
1975 #[test]
1976 fn arrow_round_trip_preserves_exact_records_across_chunking() {
1977 let ledger = fixture();
1978 let assertions = ledger.assertion_batch().unwrap();
1979 let refs = ledger.graph_ref_batch().unwrap();
1980 assert_eq!(
1981 AssertionLedger::from_batches(
1982 &[assertions.slice(0, 0), assertions.slice(0, 1)],
1983 &[refs.slice(0, 1), refs.slice(1, 1)],
1984 )
1985 .unwrap(),
1986 ledger
1987 );
1988 }
1989
1990 #[test]
1991 fn invalid_claims_refs_and_ordinals_fail_structurally() {
1992 let assertion_uuid = uuid7(1);
1993 let assertion = Assertion::new(assertion_uuid, "claim".into(), uuid7(2), 10).unwrap();
1994 assert!(AssertionLedger::new(vec![assertion.clone()], vec![]).is_err());
1995 assert!(matches!(
1996 Assertion::new(assertion_uuid, String::new(), uuid7(2), 10),
1997 Err(KnowledgeError::Invalid { field: "claim", .. })
1998 ));
1999 let oversized =
2000 "x".repeat(graphforge_core::canonical::MAX_CANONICAL_TEXT_BYTES as usize + 1);
2001 assert!(matches!(
2002 Assertion::new(assertion_uuid, oversized, uuid7(2), 10),
2003 Err(KnowledgeError::Invalid { field: "claim", .. })
2004 ));
2005
2006 let non_contiguous = AssertionGraphRef::new(
2007 assertion_uuid,
2008 uuid7(3),
2009 GraphObjectKind::Node,
2010 AssertionGraphRole::Subject,
2011 1,
2012 )
2013 .unwrap();
2014 assert!(matches!(
2015 AssertionLedger::new(vec![assertion.clone()], vec![non_contiguous]),
2016 Err(KnowledgeError::Invalid {
2017 field: "ordinal",
2018 ..
2019 })
2020 ));
2021
2022 let duplicate = AssertionGraphRef::new(
2023 assertion_uuid,
2024 uuid7(3),
2025 GraphObjectKind::Node,
2026 AssertionGraphRole::Subject,
2027 0,
2028 )
2029 .unwrap();
2030 assert!(matches!(
2031 AssertionLedger::new(vec![assertion], vec![duplicate.clone(), duplicate]),
2032 Err(KnowledgeError::Duplicate(
2033 "assertion_uuid/graph_uuid/role/ordinal"
2034 ))
2035 ));
2036 }
2037
2038 #[test]
2039 fn assertion_row_validation_rejects_bad_identity_version_and_duplicate() {
2040 let valid = fixture().assertions[0].clone();
2041 let mut invalid = valid.clone();
2042 invalid.contract_version += 1;
2043 assert!(matches!(
2044 AssertionLedger::new(vec![invalid], vec![]),
2045 Err(KnowledgeError::Invalid {
2046 field: "assertion.contract_version",
2047 ..
2048 })
2049 ));
2050
2051 invalid = valid.clone();
2052 invalid.assertion_uuid = Uuid::nil();
2053 assert!(matches!(
2054 AssertionLedger::new(vec![invalid], vec![]),
2055 Err(KnowledgeError::Invalid {
2056 field: "assertion_uuid",
2057 ..
2058 })
2059 ));
2060
2061 invalid = valid.clone();
2062 invalid.assertion_uuid = non_v7_uuid(5);
2063 assert!(matches!(
2064 AssertionLedger::new(vec![invalid], vec![]),
2065 Err(KnowledgeError::Invalid {
2066 field: "assertion_uuid",
2067 message: "must be UUIDv7",
2068 })
2069 ));
2070
2071 invalid = valid.clone();
2072 invalid.provenance_uuid = Uuid::nil();
2073 assert!(matches!(
2074 AssertionLedger::new(vec![invalid], vec![]),
2075 Err(KnowledgeError::Invalid {
2076 field: "provenance_uuid",
2077 ..
2078 })
2079 ));
2080
2081 assert!(matches!(
2082 AssertionLedger::new(vec![valid.clone(), valid], vec![]),
2083 Err(KnowledgeError::Duplicate("assertion_uuid"))
2084 ));
2085 }
2086
2087 #[test]
2088 fn assertion_ledger_rejects_an_empty_claim_after_row_construction() {
2089 let valid = fixture();
2090 let mut assertion = valid.assertions[0].clone();
2091 assertion.claim.clear();
2092 assert!(matches!(
2093 AssertionLedger::new(vec![assertion], valid.graph_refs.clone()),
2094 Err(KnowledgeError::Invalid {
2095 field: "claim",
2096 message: "must not be empty",
2097 })
2098 ));
2099 }
2100
2101 #[test]
2102 fn assertion_ledger_enforces_claim_byte_limit_after_row_construction() {
2103 let valid = fixture();
2104 let mut assertion = valid.assertions[0].clone();
2105 assertion.claim =
2106 "x".repeat(graphforge_core::canonical::MAX_CANONICAL_TEXT_BYTES as usize + 1);
2107 assert!(matches!(
2108 AssertionLedger::new(vec![assertion], valid.graph_refs.clone()),
2109 Err(KnowledgeError::Invalid {
2110 field: "claim",
2111 message: "exceeds canonical UTF-8 limit",
2112 })
2113 ));
2114 }
2115
2116 #[test]
2117 fn graph_reference_row_validation_rejects_bad_identity_and_version() {
2118 let assertion = fixture().assertions[0].clone();
2119 let valid = AssertionGraphRef::new(
2120 assertion.assertion_uuid,
2121 uuid7(90),
2122 GraphObjectKind::Node,
2123 AssertionGraphRole::Subject,
2124 0,
2125 )
2126 .unwrap();
2127 let mut invalid = valid.clone();
2128 invalid.contract_version += 1;
2129 assert!(matches!(
2130 AssertionLedger::new(vec![assertion.clone()], vec![invalid]),
2131 Err(KnowledgeError::Invalid {
2132 field: "assertion_graph_ref.contract_version",
2133 ..
2134 })
2135 ));
2136
2137 invalid = valid.clone();
2138 invalid.assertion_uuid = Uuid::nil();
2139 assert!(matches!(
2140 AssertionLedger::new(vec![assertion.clone()], vec![invalid]),
2141 Err(KnowledgeError::Invalid {
2142 field: "assertion_uuid",
2143 ..
2144 })
2145 ));
2146
2147 invalid = valid.clone();
2148 invalid.assertion_uuid = non_v7_uuid(6);
2149 assert!(matches!(
2150 AssertionLedger::new(vec![assertion.clone()], vec![invalid]),
2151 Err(KnowledgeError::Invalid {
2152 field: "assertion_uuid",
2153 message: "must be UUIDv7",
2154 })
2155 ));
2156
2157 invalid = valid.clone();
2158 invalid.graph_uuid = Uuid::nil();
2159 assert!(matches!(
2160 AssertionLedger::new(vec![assertion], vec![invalid]),
2161 Err(KnowledgeError::Invalid {
2162 field: "graph_uuid",
2163 ..
2164 })
2165 ));
2166 }
2167
2168 #[test]
2169 fn graph_reference_validation_requires_an_existing_assertion_owner() {
2170 let assertion = fixture().assertions[0].clone();
2171 let reference = AssertionGraphRef::new(
2172 uuid7(99),
2173 uuid7(90),
2174 GraphObjectKind::Node,
2175 AssertionGraphRole::Subject,
2176 0,
2177 )
2178 .unwrap();
2179 assert!(matches!(
2180 AssertionLedger::new(vec![assertion], vec![reference]),
2181 Err(KnowledgeError::Dangling("assertion_uuid"))
2182 ));
2183 }
2184
2185 #[test]
2186 fn assertion_row_limits_reject_the_first_excess_count() {
2187 assert!(validate_assertion_row_limits(2, 2, 2).is_ok());
2188 assert!(matches!(
2189 validate_assertion_row_limits(3, 2, 2),
2190 Err(KnowledgeError::Limit {
2191 participant: "assertions",
2192 observed: 3,
2193 limit: 2,
2194 })
2195 ));
2196 assert!(matches!(
2197 validate_assertion_row_limits(2, 3, 2),
2198 Err(KnowledgeError::Limit {
2199 participant: "assertion_graph_refs",
2200 observed: 3,
2201 limit: 2,
2202 })
2203 ));
2204 }
2205
2206 #[test]
2207 fn assertion_merge_skips_validation_and_preserves_order_and_limits() {
2208 let base = fixture();
2209 let second_id = uuid7(20);
2210 let staged = AssertionLedger::new(
2211 vec![Assertion::new(second_id, "second".into(), uuid7(21), 20).unwrap()],
2212 vec![
2213 AssertionGraphRef::new(
2214 second_id,
2215 uuid7(22),
2216 GraphObjectKind::Node,
2217 AssertionGraphRole::Subject,
2218 0,
2219 )
2220 .unwrap(),
2221 ],
2222 )
2223 .unwrap();
2224
2225 reset_assertion_row_validation_count();
2226 let merged = base.merge(&staged).unwrap();
2227 assert_eq!(assertion_row_validation_count(), 0);
2228 assert_eq!(merged.assertions()[0].assertion_uuid, uuid7(1));
2229 assert_eq!(merged.assertions()[1].assertion_uuid, second_id);
2230 assert_eq!(
2231 merged,
2232 AssertionLedger::new(
2233 [base.assertions.clone(), staged.assertions.clone()].concat(),
2234 [base.graph_refs.clone(), staged.graph_refs.clone()].concat(),
2235 )
2236 .unwrap()
2237 );
2238
2239 assert!(matches!(
2240 base.merge_with_limits(&staged, 1, MAX_KNOWLEDGE_ROWS),
2241 Err(KnowledgeError::Limit {
2242 participant: "assertions",
2243 observed: 2,
2244 limit: 1,
2245 })
2246 ));
2247 assert!(matches!(
2248 base.merge_with_limits(&staged, MAX_KNOWLEDGE_ROWS, 2),
2249 Err(KnowledgeError::Limit {
2250 participant: "assertion_graph_refs",
2251 observed: 3,
2252 limit: 2,
2253 })
2254 ));
2255 }
2256
2257 #[test]
2258 fn assertion_merge_sorts_backdated_new_assertions_and_references() {
2259 let base = fixture();
2260 let earlier_uuid = uuid7(20);
2261 let staged = AssertionLedger::new(
2262 vec![Assertion::new(earlier_uuid, "earlier".into(), uuid7(21), 5).unwrap()],
2263 vec![
2264 AssertionGraphRef::new(
2265 earlier_uuid,
2266 uuid7(22),
2267 GraphObjectKind::Node,
2268 AssertionGraphRole::Subject,
2269 0,
2270 )
2271 .unwrap(),
2272 ],
2273 )
2274 .unwrap();
2275
2276 let merged = base.merge(&staged).unwrap();
2277
2278 assert_eq!(
2279 merged
2280 .assertions()
2281 .iter()
2282 .map(|row| row.assertion_uuid)
2283 .collect::<Vec<_>>(),
2284 vec![earlier_uuid, uuid7(1)]
2285 );
2286 assert_eq!(
2287 merged
2288 .graph_refs()
2289 .iter()
2290 .map(|row| row.assertion_uuid)
2291 .collect::<Vec<_>>(),
2292 vec![earlier_uuid, uuid7(1), uuid7(1)]
2293 );
2294 }
2295
2296 #[test]
2297 fn closed_values_fail_during_arrow_decode() {
2298 let ledger = fixture();
2299 let assertions = ledger.assertion_batch().unwrap();
2300 let refs = ledger.graph_ref_batch().unwrap();
2301 let bad_kind = RecordBatch::try_new(
2302 Arc::clone(&ASSERTION_GRAPH_REF_SCHEMA),
2303 vec![
2304 Arc::clone(refs.column(0)),
2305 Arc::clone(refs.column(1)),
2306 Arc::new(StringArray::from(vec!["vertex", "edge"])),
2307 Arc::clone(refs.column(3)),
2308 Arc::clone(refs.column(4)),
2309 Arc::clone(refs.column(5)),
2310 ],
2311 )
2312 .unwrap();
2313 assert!(matches!(
2314 AssertionLedger::from_batches(&[assertions.clone()], &[bad_kind]),
2315 Err(KnowledgeError::Invalid {
2316 field: "graph_kind",
2317 ..
2318 })
2319 ));
2320
2321 let bad_role = RecordBatch::try_new(
2322 Arc::clone(&ASSERTION_GRAPH_REF_SCHEMA),
2323 vec![
2324 Arc::clone(refs.column(0)),
2325 Arc::clone(refs.column(1)),
2326 Arc::clone(refs.column(2)),
2327 Arc::new(StringArray::from(vec!["target", "context"])),
2328 Arc::clone(refs.column(4)),
2329 Arc::clone(refs.column(5)),
2330 ],
2331 )
2332 .unwrap();
2333 assert!(matches!(
2334 AssertionLedger::from_batches(&[assertions], &[bad_role]),
2335 Err(KnowledgeError::Invalid { field: "role", .. })
2336 ));
2337 }
2338
2339 #[test]
2340 fn knowledge_schema_registry_excludes_every_epistemic_field() {
2341 let fields = schema_registry()
2342 .into_iter()
2343 .filter(|entry| entry.capability_id == "knowledge")
2344 .flat_map(|entry| {
2345 entry
2346 .schema
2347 .fields()
2348 .iter()
2349 .map(|field| field.name().clone())
2350 .collect::<Vec<_>>()
2351 })
2352 .collect::<HashSet<_>>();
2353 for deferred in [
2354 "status",
2355 "confidence",
2356 "evidence",
2357 "hypothesis_uuid",
2358 "reasoning",
2359 "supersedes_uuid",
2360 "valid_from",
2361 "valid_to",
2362 ] {
2363 assert!(!fields.contains(deferred));
2364 }
2365 }
2366
2367 #[test]
2368 fn schema_registry_owns_record_diff_identity_contracts() {
2369 let identities = schema_registry()
2370 .into_iter()
2371 .map(|entry| {
2372 for field in entry.diff_identity_fields {
2373 assert!(entry.schema.field_with_name(field).is_ok());
2374 }
2375 if let Some(field) = entry.diff_record_uuid_field {
2376 assert!(entry.diff_identity_fields.contains(&field));
2377 assert_eq!(
2378 entry.schema.field_with_name(field).unwrap().data_type(),
2379 &DataType::FixedSizeBinary(16)
2380 );
2381 }
2382 (
2383 entry.record_family,
2384 entry.diff_identity_fields,
2385 entry.diff_record_uuid_field,
2386 )
2387 })
2388 .collect::<Vec<_>>();
2389
2390 assert!(identities.contains(&(
2391 "assertions",
2392 &["assertion_uuid"][..],
2393 Some("assertion_uuid")
2394 )));
2395 assert!(identities.contains(&(
2396 "assertion_graph_refs",
2397 &["assertion_uuid", "graph_uuid", "role", "ordinal"][..],
2398 None
2399 )));
2400 assert!(identities.contains(&(
2401 "confidence_inputs",
2402 &["confidence_uuid", "input_confidence_uuid"][..],
2403 None
2404 )));
2405 }
2406
2407 #[test]
2408 fn evidence_round_trips_orders_fingerprints_and_merges_idempotently() {
2409 let later = EvidenceLink::new(
2410 uuid7(31),
2411 uuid7(1),
2412 uuid7(41),
2413 EvidenceSourceKind::Observation,
2414 EvidenceRole::Contradicts,
2415 Some(0.25),
2416 uuid7(51),
2417 20,
2418 )
2419 .unwrap();
2420 let earlier = EvidenceLink::new(
2421 uuid7(30),
2422 uuid7(1),
2423 uuid7(40),
2424 EvidenceSourceKind::Document,
2425 EvidenceRole::Supports,
2426 Some(-0.0),
2427 uuid7(50),
2428 10,
2429 )
2430 .unwrap();
2431 let ledger = EvidenceLedger::new(vec![later, earlier]).unwrap();
2432 assert_eq!(ledger.links()[0].evidence_uuid, uuid7(30));
2433 assert_eq!(
2434 ledger.links()[0].weight.unwrap().to_bits(),
2435 0.0f64.to_bits()
2436 );
2437 assert_eq!(
2438 EvidenceLedger::from_batches(&[ledger.batch().unwrap()]).unwrap(),
2439 ledger
2440 );
2441 assert_eq!(ledger.merge(&ledger).unwrap(), ledger);
2442 let encoded = ledger
2443 .evidence_fingerprint(uuid7(30))
2444 .unwrap()
2445 .iter()
2446 .map(|byte| format!("{byte:02x}"))
2447 .collect::<String>();
2448 assert_eq!(
2449 encoded,
2450 "363146c1fc94d57cc02c098a7187d753d85a7b10ebd715ecbae76e85e1d3fd5e"
2451 );
2452 }
2453
2454 #[test]
2455 fn evidence_validation_and_conflicting_identity_are_fail_closed() {
2456 for weight in [f64::NAN, f64::INFINITY, -0.01, 1.01] {
2457 assert!(matches!(
2458 EvidenceLink::new(
2459 uuid7(30),
2460 uuid7(1),
2461 uuid7(40),
2462 EvidenceSourceKind::Document,
2463 EvidenceRole::Supports,
2464 Some(weight),
2465 uuid7(50),
2466 10,
2467 ),
2468 Err(KnowledgeError::Invalid {
2469 field: "weight",
2470 ..
2471 })
2472 ));
2473 }
2474 let first = EvidenceLedger::new(vec![
2475 EvidenceLink::new(
2476 uuid7(30),
2477 uuid7(1),
2478 uuid7(40),
2479 EvidenceSourceKind::Document,
2480 EvidenceRole::Supports,
2481 None,
2482 uuid7(50),
2483 10,
2484 )
2485 .unwrap(),
2486 ])
2487 .unwrap();
2488 let conflict = EvidenceLedger::new(vec![
2489 EvidenceLink::new(
2490 uuid7(30),
2491 uuid7(1),
2492 uuid7(41),
2493 EvidenceSourceKind::Observation,
2494 EvidenceRole::Context,
2495 None,
2496 uuid7(50),
2497 10,
2498 )
2499 .unwrap(),
2500 ])
2501 .unwrap();
2502 assert!(matches!(
2503 first.merge(&conflict),
2504 Err(KnowledgeError::Conflict("evidence_uuid"))
2505 ));
2506 }
2507
2508 #[test]
2509 fn evidence_ledger_rejects_each_invalid_record_invariant() {
2510 let valid = EvidenceLink::new(
2511 uuid7(30),
2512 uuid7(1),
2513 uuid7(40),
2514 EvidenceSourceKind::Document,
2515 EvidenceRole::Supports,
2516 Some(0.5),
2517 uuid7(50),
2518 10,
2519 )
2520 .unwrap();
2521
2522 let mut invalid = valid.clone();
2523 invalid.evidence_uuid = Uuid::from_u128(1);
2524 assert!(matches!(
2525 EvidenceLedger::new(vec![invalid]),
2526 Err(KnowledgeError::Invalid {
2527 field: "evidence_uuid",
2528 ..
2529 })
2530 ));
2531
2532 let mut invalid = valid.clone();
2533 invalid.assertion_uuid = Uuid::nil();
2534 assert!(matches!(
2535 EvidenceLedger::new(vec![invalid]),
2536 Err(KnowledgeError::Invalid {
2537 field: "assertion_uuid",
2538 ..
2539 })
2540 ));
2541
2542 let mut invalid = valid.clone();
2543 invalid.source_uuid = Uuid::nil();
2544 assert!(matches!(
2545 EvidenceLedger::new(vec![invalid]),
2546 Err(KnowledgeError::Invalid {
2547 field: "source_uuid",
2548 ..
2549 })
2550 ));
2551
2552 let mut invalid = valid.clone();
2553 invalid.provenance_uuid = Uuid::nil();
2554 assert!(matches!(
2555 EvidenceLedger::new(vec![invalid]),
2556 Err(KnowledgeError::Invalid {
2557 field: "provenance_uuid",
2558 ..
2559 })
2560 ));
2561
2562 let mut invalid = valid.clone();
2563 invalid.contract_version += 1;
2564 assert!(matches!(
2565 EvidenceLedger::new(vec![invalid]),
2566 Err(KnowledgeError::Invalid {
2567 field: "contract_version",
2568 ..
2569 })
2570 ));
2571
2572 for weight in [f64::NAN, f64::INFINITY, -0.01, 1.01] {
2573 let mut invalid = valid.clone();
2574 invalid.weight = Some(weight);
2575 assert!(matches!(
2576 EvidenceLedger::new(vec![invalid]),
2577 Err(KnowledgeError::Invalid {
2578 field: "weight",
2579 ..
2580 })
2581 ));
2582 }
2583
2584 assert!(matches!(
2585 EvidenceLedger::new(vec![valid.clone(), valid]),
2586 Err(KnowledgeError::Duplicate("evidence_uuid"))
2587 ));
2588 }
2589
2590 #[test]
2591 fn evidence_ledger_enforces_row_limit() {
2592 let link = EvidenceLink::new(
2593 uuid7(30),
2594 uuid7(1),
2595 uuid7(40),
2596 EvidenceSourceKind::Document,
2597 EvidenceRole::Supports,
2598 None,
2599 uuid7(50),
2600 10,
2601 )
2602 .unwrap();
2603 let links = vec![link; MAX_KNOWLEDGE_ROWS + 1];
2604 assert!(matches!(
2605 EvidenceLedger::new(links),
2606 Err(KnowledgeError::Limit {
2607 participant: "evidence",
2608 observed,
2609 limit: MAX_KNOWLEDGE_ROWS,
2610 }) if observed == MAX_KNOWLEDGE_ROWS + 1
2611 ));
2612 }
2613
2614 #[test]
2615 fn evidence_merge_enforces_combined_row_limit() {
2616 fn uuid7_from_counter(counter: u128) -> Uuid {
2617 let mut bytes = counter.to_be_bytes();
2618 bytes[6] = (bytes[6] & 0x0f) | 0x70;
2619 bytes[8] = (bytes[8] & 0x3f) | 0x80;
2620 Uuid::from_bytes(bytes)
2621 }
2622
2623 let links = (0..MAX_KNOWLEDGE_ROWS)
2624 .map(|index| {
2625 EvidenceLink::new(
2626 uuid7_from_counter(index as u128 + 1),
2627 uuid7(1),
2628 uuid7(40),
2629 EvidenceSourceKind::Document,
2630 EvidenceRole::Supports,
2631 None,
2632 uuid7(50),
2633 index as i64,
2634 )
2635 .unwrap()
2636 })
2637 .collect();
2638 let current = EvidenceLedger::new(links).unwrap();
2639 let staged = EvidenceLedger::new(vec![
2640 EvidenceLink::new(
2641 uuid7_from_counter(MAX_KNOWLEDGE_ROWS as u128 + 1),
2642 uuid7(1),
2643 uuid7(40),
2644 EvidenceSourceKind::Document,
2645 EvidenceRole::Supports,
2646 None,
2647 uuid7(50),
2648 MAX_KNOWLEDGE_ROWS as i64,
2649 )
2650 .unwrap(),
2651 ])
2652 .unwrap();
2653
2654 assert!(matches!(
2655 current.merge(&staged),
2656 Err(KnowledgeError::Limit {
2657 participant: "evidence",
2658 observed,
2659 limit: MAX_KNOWLEDGE_ROWS,
2660 }) if observed == MAX_KNOWLEDGE_ROWS + 1
2661 ));
2662 }
2663
2664 #[test]
2665 fn evidence_merge_keeps_canonical_order_and_conflict_semantics() {
2666 let make_link = |evidence_id, time| {
2667 EvidenceLink::new(
2668 uuid7(evidence_id),
2669 uuid7(1),
2670 uuid7(40),
2671 EvidenceSourceKind::Document,
2672 EvidenceRole::Supports,
2673 None,
2674 uuid7(50),
2675 time,
2676 )
2677 .unwrap()
2678 };
2679 let current = EvidenceLedger::new(vec![make_link(32, 20), make_link(30, 10)]).unwrap();
2680 let staged = EvidenceLedger::new(vec![make_link(33, 20), make_link(31, 10)]).unwrap();
2681
2682 let merged = current.merge(&staged).unwrap();
2683 assert_eq!(
2684 merged
2685 .links()
2686 .iter()
2687 .map(|row| row.evidence_uuid)
2688 .collect::<Vec<_>>(),
2689 [uuid7(30), uuid7(31), uuid7(32), uuid7(33)]
2690 );
2691 assert_eq!(merged.merge(¤t).unwrap(), merged);
2692 assert_eq!(merged.clone().into_links(), merged.links().to_vec());
2693 }
2694
2695 #[test]
2696 fn evidence_merge_does_not_revalidate_existing_rows() {
2697 let make_link = |evidence_id| {
2698 EvidenceLink::new(
2699 uuid7(evidence_id),
2700 uuid7(1),
2701 uuid7(40),
2702 EvidenceSourceKind::Document,
2703 EvidenceRole::Supports,
2704 None,
2705 uuid7(50),
2706 evidence_id as i64,
2707 )
2708 .unwrap()
2709 };
2710 let current = EvidenceLedger::new(vec![make_link(30), make_link(31)]).unwrap();
2711 let staged = EvidenceLedger::new(vec![make_link(32)]).unwrap();
2712 EVIDENCE_ROW_VALIDATION_COUNT.with(|count| count.set(0));
2713
2714 let merged = current.merge(&staged).unwrap();
2715
2716 EVIDENCE_ROW_VALIDATION_COUNT.with(|count| assert_eq!(count.get(), 0));
2717 assert_eq!(merged.links().len(), 3);
2718 }
2719
2720 #[test]
2721 #[ignore = "manual quiet-host merge cost measurement; run with --ignored --nocapture"]
2722 fn evidence_merge_cost_per_existing_row() {
2723 use std::time::Instant;
2724
2725 fn uuid7_from_counter(counter: u128) -> Uuid {
2726 let mut bytes = counter.to_be_bytes();
2727 bytes[6] = (bytes[6] & 0x0f) | 0x70;
2728 bytes[8] = (bytes[8] & 0x3f) | 0x80;
2729 Uuid::from_bytes(bytes)
2730 }
2731
2732 for row_count in [10_000usize, 100_000] {
2733 let links = (0..row_count)
2734 .map(|index| {
2735 EvidenceLink::new(
2736 uuid7_from_counter(index as u128 + 1),
2737 uuid7(1),
2738 uuid7(40),
2739 EvidenceSourceKind::Document,
2740 EvidenceRole::Supports,
2741 None,
2742 uuid7(50),
2743 index as i64,
2744 )
2745 .unwrap()
2746 })
2747 .collect();
2748 let ledger = EvidenceLedger::new(links).unwrap();
2749 let staged = EvidenceLedger::new(vec![
2750 EvidenceLink::new(
2751 uuid7_from_counter(row_count as u128 + 1),
2752 uuid7(1),
2753 uuid7(40),
2754 EvidenceSourceKind::Document,
2755 EvidenceRole::Supports,
2756 None,
2757 uuid7(50),
2758 row_count as i64,
2759 )
2760 .unwrap(),
2761 ])
2762 .unwrap();
2763 let iterations = 10;
2764 let started = Instant::now();
2765 for _ in 0..iterations {
2766 std::hint::black_box(ledger.merge(&staged).unwrap());
2767 }
2768 let per_existing_row =
2769 started.elapsed().as_nanos() as f64 / (iterations * row_count) as f64;
2770 eprintln!(
2771 "existing_rows={row_count} iterations={iterations} ns_per_existing_row={per_existing_row:.3}"
2772 );
2773 }
2774 }
2775
2776 #[test]
2777 fn closed_domain_vocabularies_round_trip_and_reject_unknown_tokens() {
2778 for value in [GraphObjectKind::Node, GraphObjectKind::Edge] {
2779 assert_eq!(GraphObjectKind::parse(value.as_str()).unwrap(), value);
2780 }
2781 assert!(GraphObjectKind::parse("vertex").is_err());
2782 for value in [
2783 AssertionGraphRole::Subject,
2784 AssertionGraphRole::Object,
2785 AssertionGraphRole::Context,
2786 ] {
2787 assert_eq!(AssertionGraphRole::parse(value.as_str()).unwrap(), value);
2788 }
2789 assert!(AssertionGraphRole::parse("target").is_err());
2790 for value in [
2791 EvidenceSourceKind::Document,
2792 EvidenceSourceKind::Observation,
2793 EvidenceSourceKind::GraphNode,
2794 EvidenceSourceKind::GraphEdge,
2795 EvidenceSourceKind::Source,
2796 EvidenceSourceKind::Artifact,
2797 ] {
2798 assert_eq!(EvidenceSourceKind::parse(value.as_str()).unwrap(), value);
2799 }
2800 assert!(EvidenceSourceKind::parse("web").is_err());
2801 for value in [
2802 EvidenceRole::Supports,
2803 EvidenceRole::Contradicts,
2804 EvidenceRole::Context,
2805 ] {
2806 assert_eq!(EvidenceRole::parse(value.as_str()).unwrap(), value);
2807 }
2808 assert!(EvidenceRole::parse("proves").is_err());
2809 for value in [
2810 ConfidencePolicy::Explicit,
2811 ConfidencePolicy::ConservativeMin,
2812 ] {
2813 assert_eq!(ConfidencePolicy::parse(value.as_str()).unwrap(), value);
2814 }
2815 assert!(ConfidencePolicy::parse("average").is_err());
2816 for value in [
2817 AlgorithmRunState::Started,
2818 AlgorithmRunState::Completed,
2819 AlgorithmRunState::Failed,
2820 AlgorithmRunState::Cancelled,
2821 AlgorithmRunState::Interrupted,
2822 ] {
2823 assert_eq!(AlgorithmRunState::parse(value.as_str()).unwrap(), value);
2824 }
2825 assert!(!AlgorithmRunState::Started.is_terminal());
2826 assert!(AlgorithmRunState::Completed.is_terminal());
2827 assert!(AlgorithmRunState::parse("running").is_err());
2828 }
2829
2830 #[test]
2831 fn knowledge_error_codes_are_closed_and_exact() {
2832 let cases = [
2833 (invalid("field", "bad"), "GF_KNOWLEDGE_INVALID"),
2834 (
2835 KnowledgeError::Limit {
2836 participant: "assertions",
2837 observed: 2,
2838 limit: 1,
2839 },
2840 "GF_RESOURCE_LIMIT",
2841 ),
2842 (KnowledgeError::Duplicate("id"), "GF_KNOWLEDGE_DUPLICATE"),
2843 (KnowledgeError::Dangling("id"), "GF_KNOWLEDGE_DANGLING"),
2844 (KnowledgeError::Conflict("id"), "GF_IDEMPOTENCY_CONFLICT"),
2845 (
2846 KnowledgeError::TransactionConflict("id"),
2847 "GF_TRANSACTION_CONFLICT",
2848 ),
2849 (
2850 KnowledgeError::Canonical(CanonicalError::Malformed("bad")),
2851 "GF_CANONICAL_INVALID",
2852 ),
2853 (
2854 KnowledgeError::Arrow(arrow::error::ArrowError::SchemaError("bad".into())),
2855 "GF_SCHEMA_MISMATCH",
2856 ),
2857 ];
2858 for (error, code) in cases {
2859 assert_eq!(error.code(), code);
2860 assert!(!error.to_string().is_empty());
2861 }
2862 }
2863
2864 #[test]
2865 fn assertion_and_algorithm_run_merges_cover_idempotent_append_and_conflict_paths() {
2866 let base = fixture();
2867 assert_eq!(base.merge(&base).unwrap(), base);
2868 let second_id = uuid7(20);
2869 let second = AssertionLedger::new(
2870 vec![Assertion::new(second_id, "second".into(), uuid7(21), 20).unwrap()],
2871 vec![
2872 AssertionGraphRef::new(
2873 second_id,
2874 uuid7(22),
2875 GraphObjectKind::Node,
2876 AssertionGraphRole::Subject,
2877 0,
2878 )
2879 .unwrap(),
2880 ],
2881 )
2882 .unwrap();
2883 let merged = base.merge(&second).unwrap();
2884 assert_eq!(merged.assertions.len(), 2);
2885 let conflicting = AssertionLedger::new(
2886 vec![Assertion::new(uuid7(1), "different".into(), uuid7(2), 10).unwrap()],
2887 base.graph_refs.clone(),
2888 )
2889 .unwrap();
2890 assert!(matches!(
2891 base.merge(&conflicting),
2892 Err(KnowledgeError::Conflict("assertion_uuid"))
2893 ));
2894
2895 let run = AlgorithmRun::new(
2896 uuid7(40),
2897 "pagerank".into(),
2898 1,
2899 1,
2900 vec![1],
2901 [2; 32],
2902 uuid7(41),
2903 100,
2904 )
2905 .unwrap();
2906 let started = AlgorithmRunEvent::new(
2907 uuid7(42),
2908 run.run_uuid,
2909 AlgorithmRunState::Started,
2910 None,
2911 None,
2912 100,
2913 run.provenance_uuid,
2914 )
2915 .unwrap();
2916 let initial = AlgorithmRunLedger::new(vec![run.clone()], vec![started.clone()]).unwrap();
2917 assert_eq!(initial.run(run.run_uuid), Some(&run));
2918 assert_eq!(initial.events_for(run.run_uuid), vec![started.clone()]);
2919 assert!(initial.terminal_event(run.run_uuid).is_none());
2920 assert_eq!(initial.merge(&initial).unwrap(), initial);
2921
2922 let completed = AlgorithmRunEvent::new(
2923 uuid7(43),
2924 run.run_uuid,
2925 AlgorithmRunState::Completed,
2926 Some([3; 32]),
2927 None,
2928 101,
2929 uuid7(44),
2930 )
2931 .unwrap();
2932 let staged =
2933 AlgorithmRunLedger::new(vec![run.clone()], vec![started, completed.clone()]).unwrap();
2934 let merged = initial.merge(&staged).unwrap();
2935 assert_eq!(merged.terminal_event(run.run_uuid), Some(&completed));
2936
2937 let conflicting_run = AlgorithmRun::new(
2938 run.run_uuid,
2939 "hits".into(),
2940 run.algorithm_version,
2941 run.descriptor_version,
2942 run.descriptor.clone(),
2943 run.projection_fingerprint,
2944 run.provenance_uuid,
2945 run.started_at_micros,
2946 )
2947 .unwrap();
2948 let conflicting_start = AlgorithmRunEvent::new(
2949 uuid7(45),
2950 conflicting_run.run_uuid,
2951 AlgorithmRunState::Started,
2952 None,
2953 None,
2954 conflicting_run.started_at_micros,
2955 conflicting_run.provenance_uuid,
2956 )
2957 .unwrap();
2958 let conflict =
2959 AlgorithmRunLedger::new(vec![conflicting_run], vec![conflicting_start]).unwrap();
2960 assert!(matches!(
2961 initial.merge(&conflict),
2962 Err(KnowledgeError::Conflict("run_uuid"))
2963 ));
2964 let conflicting_event = AlgorithmRunEvent::new(
2965 completed.event_uuid,
2966 completed.run_uuid,
2967 AlgorithmRunState::Failed,
2968 None,
2969 Some("GF_EXECUTION".into()),
2970 completed.recorded_at_micros,
2971 completed.provenance_uuid,
2972 )
2973 .unwrap();
2974 let conflict = AlgorithmRunLedger::new(
2975 vec![run],
2976 vec![initial.events()[0].clone(), conflicting_event],
2977 )
2978 .unwrap();
2979 assert!(matches!(
2980 merged.merge(&conflict),
2981 Err(KnowledgeError::Conflict("event_uuid"))
2982 ));
2983 }
2984}
2985pub use belief_projection::{
2986 ALGORITHM_INTERPRETATION_ATTACHMENT_SCHEMA, BELIEF_PROJECTION_ATTACHMENT_CONTRACT_VERSION,
2987 BeliefProjectionAttachment, BeliefProjectionAttachmentLedger,
2988};