graphforge_knowledge/
supersession.rs

1//! Append-only epistemic assertion supersession relations.
2
3use std::collections::{HashMap, HashSet};
4use std::sync::{Arc, LazyLock};
5
6use arrow::array::{
7    Array, FixedSizeBinaryArray, FixedSizeBinaryBuilder, TimestampMicrosecondArray,
8    TimestampMicrosecondBuilder, UInt32Array, UInt32Builder,
9};
10use arrow::datatypes::{DataType, Field, Schema, SchemaRef, TimeUnit};
11use arrow::record_batch::RecordBatch;
12use graphforge_core::canonical::{
13    CANONICAL_CONTRACT_VERSION, CanonicalDomain, CanonicalWriter, fingerprint,
14};
15use uuid::{Uuid, Version};
16
17use crate::{
18    EPISTEMIC_CAPABILITY_VERSION, KnowledgeError, MAX_KNOWLEDGE_ROWS, SchemaRegistryEntry,
19};
20
21/// Assertion-supersession record contract.
22pub const ASSERTION_SUPERSESSION_CONTRACT_VERSION: u32 = 1;
23/// Branch-preserving, non-selecting supersession policy.
24pub const ASSERTION_SUPERSESSION_POLICY_VERSION: u32 = 1;
25
26/// Authoritative `knowledge/assertion_supersessions.parquet` schema.
27pub static ASSERTION_SUPERSESSION_SCHEMA: LazyLock<SchemaRef> = LazyLock::new(|| {
28    Arc::new(Schema::new(vec![
29        uuid_field("supersession_uuid"),
30        uuid_field("prior_assertion_uuid"),
31        uuid_field("replacement_assertion_uuid"),
32        uuid_field("status_event_uuid"),
33        uuid_field("reasoning_uuid"),
34        uuid_field("provenance_uuid"),
35        Field::new(
36            "recorded_at",
37            DataType::Timestamp(TimeUnit::Microsecond, Some("UTC".into())),
38            false,
39        ),
40        Field::new("contract_version", DataType::UInt32, false),
41    ]))
42});
43
44static ASSERTION_SUPERSESSION_SCHEMA_FINGERPRINT: LazyLock<[u8; 32]> = LazyLock::new(|| {
45    fingerprint(
46        CanonicalDomain::AssertionSupersession,
47        CANONICAL_CONTRACT_VERSION,
48        b"assertion_supersession/1|supersession_uuid:fixed[16]:required|prior_assertion_uuid:fixed[16]:required|replacement_assertion_uuid:fixed[16]:required|status_event_uuid:fixed[16]:required|reasoning_uuid:fixed[16]:required|provenance_uuid:fixed[16]:required|recorded_at:timestamp_us_utc:required|contract_version:u32:required",
49    )
50    .expect("registered assertion-supersession schema is within canonical bounds")
51});
52
53/// One immutable assertion-supersession relation.
54#[derive(Clone, Debug, Eq, PartialEq)]
55pub struct AssertionSupersession {
56    /// Caller-supplied UUIDv7 identity/idempotency key.
57    pub supersession_uuid: Uuid,
58    /// Existing assertion that becomes explicitly superseded.
59    pub prior_assertion_uuid: Uuid,
60    /// Existing replacement assertion.
61    pub replacement_assertion_uuid: Uuid,
62    /// Exact paired `superseded` status-event UUID.
63    pub status_event_uuid: Uuid,
64    /// Existing reasoning record attached to the prior assertion.
65    pub reasoning_uuid: Uuid,
66    /// Producing provenance event.
67    pub provenance_uuid: Uuid,
68    /// Transaction time shared with the paired status event.
69    pub recorded_at_micros: i64,
70    /// Frozen record contract.
71    pub contract_version: u32,
72}
73
74impl AssertionSupersession {
75    /// Validate and construct one immutable relation.
76    pub fn new(
77        supersession_uuid: Uuid,
78        prior_assertion_uuid: Uuid,
79        replacement_assertion_uuid: Uuid,
80        status_event_uuid: Uuid,
81        reasoning_uuid: Uuid,
82        provenance_uuid: Uuid,
83        recorded_at_micros: i64,
84    ) -> Result<Self, KnowledgeError> {
85        for (value, field) in [
86            (supersession_uuid, "supersession_uuid"),
87            (prior_assertion_uuid, "prior_assertion_uuid"),
88            (replacement_assertion_uuid, "replacement_assertion_uuid"),
89            (status_event_uuid, "status_event_uuid"),
90            (reasoning_uuid, "reasoning_uuid"),
91        ] {
92            require_v7(value, field)?;
93        }
94        require_uuid(provenance_uuid, "provenance_uuid")?;
95        if prior_assertion_uuid == replacement_assertion_uuid {
96            return Err(invalid(
97                "replacement_assertion_uuid",
98                "must differ from prior_assertion_uuid",
99            ));
100        }
101        Ok(Self {
102            supersession_uuid,
103            prior_assertion_uuid,
104            replacement_assertion_uuid,
105            status_event_uuid,
106            reasoning_uuid,
107            provenance_uuid,
108            recorded_at_micros,
109            contract_version: ASSERTION_SUPERSESSION_CONTRACT_VERSION,
110        })
111    }
112}
113
114/// Validated branch-preserving supersession participant.
115#[derive(Clone, Debug, Default, Eq, PartialEq)]
116pub struct AssertionSupersessionLedger {
117    /// Relations ordered by `(recorded_at, supersession_uuid)`.
118    relations: Vec<AssertionSupersession>,
119}
120
121impl AssertionSupersessionLedger {
122    /// Return validated relations in canonical order.
123    #[must_use]
124    pub fn relations(&self) -> &[AssertionSupersession] {
125        &self.relations
126    }
127
128    /// Validate, sort, and construct one complete participant.
129    pub fn new(mut relations: Vec<AssertionSupersession>) -> Result<Self, KnowledgeError> {
130        if relations.len() > MAX_KNOWLEDGE_ROWS {
131            return Err(KnowledgeError::Limit {
132                participant: "assertion_supersessions",
133                observed: relations.len(),
134                limit: MAX_KNOWLEDGE_ROWS,
135            });
136        }
137        let mut ids = HashSet::with_capacity(relations.len());
138        let mut status_ids = HashSet::with_capacity(relations.len());
139        for relation in &relations {
140            validate_relation(relation)?;
141            if !ids.insert(relation.supersession_uuid) {
142                return Err(KnowledgeError::Duplicate("supersession_uuid"));
143            }
144            if !status_ids.insert(relation.status_event_uuid) {
145                return Err(KnowledgeError::Duplicate("status_event_uuid"));
146            }
147        }
148        validate_acyclic(&relations)?;
149        relations.sort_by_key(|row| (row.recorded_at_micros, row.supersession_uuid));
150        Ok(Self { relations })
151    }
152
153    /// Merge append-only relations with exact replay semantics.
154    pub fn merge(&self, staged: &Self) -> Result<Self, KnowledgeError> {
155        let mut relations = self.relations.clone();
156        let mut by_id = relations
157            .iter()
158            .cloned()
159            .map(|row| (row.supersession_uuid, row))
160            .collect::<HashMap<_, _>>();
161        for relation in &staged.relations {
162            if let Some(existing) = by_id.get(&relation.supersession_uuid) {
163                if existing != relation {
164                    return Err(KnowledgeError::Conflict("supersession_uuid"));
165                }
166            } else {
167                relations.push(relation.clone());
168                by_id.insert(relation.supersession_uuid, relation.clone());
169            }
170        }
171        Self::new(relations)
172    }
173
174    /// Canonical fingerprint over one exact immutable relation.
175    pub fn relation_fingerprint(
176        &self,
177        supersession_uuid: Uuid,
178    ) -> Result<[u8; 32], KnowledgeError> {
179        let row = self
180            .relations
181            .iter()
182            .find(|row| row.supersession_uuid == supersession_uuid)
183            .ok_or(KnowledgeError::Dangling("supersession_uuid"))?;
184        relation_fingerprint_for_row(row)
185    }
186
187    /// Compute canonical relation fingerprints in one pass without UUID searches.
188    pub fn relation_fingerprints(&self) -> Result<Vec<(Uuid, [u8; 32])>, KnowledgeError> {
189        self.relations
190            .iter()
191            .map(|row| Ok((row.supersession_uuid, relation_fingerprint_for_row(row)?)))
192            .collect()
193    }
194
195    /// Build the authoritative Arrow batch.
196    pub fn batch(&self) -> Result<RecordBatch, KnowledgeError> {
197        let len = self.relations.len();
198        let mut supersessions = FixedSizeBinaryBuilder::with_capacity(len, 16);
199        let mut priors = FixedSizeBinaryBuilder::with_capacity(len, 16);
200        let mut replacements = FixedSizeBinaryBuilder::with_capacity(len, 16);
201        let mut statuses = FixedSizeBinaryBuilder::with_capacity(len, 16);
202        let mut reasoning = FixedSizeBinaryBuilder::with_capacity(len, 16);
203        let mut provenance = FixedSizeBinaryBuilder::with_capacity(len, 16);
204        let mut times = TimestampMicrosecondBuilder::with_capacity(len).with_timezone("UTC");
205        let mut versions = UInt32Builder::with_capacity(len);
206        for row in &self.relations {
207            for (builder, value, field) in [
208                (
209                    &mut supersessions,
210                    row.supersession_uuid,
211                    "supersession_uuid",
212                ),
213                (
214                    &mut priors,
215                    row.prior_assertion_uuid,
216                    "prior_assertion_uuid",
217                ),
218                (
219                    &mut replacements,
220                    row.replacement_assertion_uuid,
221                    "replacement_assertion_uuid",
222                ),
223                (&mut statuses, row.status_event_uuid, "status_event_uuid"),
224                (&mut reasoning, row.reasoning_uuid, "reasoning_uuid"),
225                (&mut provenance, row.provenance_uuid, "provenance_uuid"),
226            ] {
227                builder
228                    .append_value(value.as_bytes())
229                    .map_err(|_| invalid(field, "invalid UUID width"))?;
230            }
231            times.append_value(row.recorded_at_micros);
232            versions.append_value(row.contract_version);
233        }
234        RecordBatch::try_new(
235            Arc::clone(&ASSERTION_SUPERSESSION_SCHEMA),
236            vec![
237                Arc::new(supersessions.finish()),
238                Arc::new(priors.finish()),
239                Arc::new(replacements.finish()),
240                Arc::new(statuses.finish()),
241                Arc::new(reasoning.finish()),
242                Arc::new(provenance.finish()),
243                Arc::new(times.finish()),
244                Arc::new(versions.finish()),
245            ],
246        )
247        .map_err(|_| invalid("assertion_supersession", "Arrow batch construction failed"))
248    }
249
250    /// Decode and validate exact Arrow batches.
251    pub fn from_batches(batches: &[RecordBatch]) -> Result<Self, KnowledgeError> {
252        let mut relations = Vec::new();
253        for batch in batches {
254            if batch.schema().as_ref() != ASSERTION_SUPERSESSION_SCHEMA.as_ref() {
255                return Err(invalid("assertion_supersession.schema", "schema mismatch"));
256            }
257            let supersessions = fixed(batch, "supersession_uuid")?;
258            let priors = fixed(batch, "prior_assertion_uuid")?;
259            let replacements = fixed(batch, "replacement_assertion_uuid")?;
260            let statuses = fixed(batch, "status_event_uuid")?;
261            let reasoning = fixed(batch, "reasoning_uuid")?;
262            let provenance = fixed(batch, "provenance_uuid")?;
263            let times = timestamp(batch, "recorded_at")?;
264            let versions = uint32(batch, "contract_version")?;
265            for row in 0..batch.num_rows() {
266                relations.push(AssertionSupersession {
267                    supersession_uuid: uuid_at(supersessions, row, "supersession_uuid")?,
268                    prior_assertion_uuid: uuid_at(priors, row, "prior_assertion_uuid")?,
269                    replacement_assertion_uuid: uuid_at(
270                        replacements,
271                        row,
272                        "replacement_assertion_uuid",
273                    )?,
274                    status_event_uuid: uuid_at(statuses, row, "status_event_uuid")?,
275                    reasoning_uuid: uuid_at(reasoning, row, "reasoning_uuid")?,
276                    provenance_uuid: uuid_at(provenance, row, "provenance_uuid")?,
277                    recorded_at_micros: times.value(row),
278                    contract_version: versions.value(row),
279                });
280            }
281        }
282        Self::new(relations)
283    }
284}
285
286fn relation_fingerprint_for_row(row: &AssertionSupersession) -> Result<[u8; 32], KnowledgeError> {
287    let mut writer = CanonicalWriter::new();
288    for value in [
289        row.supersession_uuid,
290        row.prior_assertion_uuid,
291        row.replacement_assertion_uuid,
292        row.status_event_uuid,
293        row.reasoning_uuid,
294        row.provenance_uuid,
295    ] {
296        writer.raw(value.as_bytes())?;
297    }
298    writer.i64(row.recorded_at_micros)?;
299    writer.u32(row.contract_version)?;
300    fingerprint(
301        CanonicalDomain::AssertionSupersession,
302        CANONICAL_CONTRACT_VERSION,
303        &writer.finish(),
304    )
305    .map_err(Into::into)
306}
307
308pub(crate) fn schema_registry_entry() -> SchemaRegistryEntry {
309    SchemaRegistryEntry {
310        capability_id: "epistemic",
311        capability_version: EPISTEMIC_CAPABILITY_VERSION,
312        record_family: "assertion_supersessions",
313        record_version: ASSERTION_SUPERSESSION_CONTRACT_VERSION,
314        schema: Arc::clone(&ASSERTION_SUPERSESSION_SCHEMA),
315        schema_fingerprint: *ASSERTION_SUPERSESSION_SCHEMA_FINGERPRINT,
316        enum_registry_versions: &[(
317            "assertion_supersession_policy",
318            ASSERTION_SUPERSESSION_POLICY_VERSION,
319        )],
320        sort_key: &["recorded_at", "supersession_uuid"],
321        diff_identity_fields: &["supersession_uuid"],
322        diff_record_uuid_field: Some("supersession_uuid"),
323        fingerprint_domain: CanonicalDomain::AssertionSupersession,
324        owner: "graphforge-knowledge",
325        implementation_issue: 778,
326        max_rows: MAX_KNOWLEDGE_ROWS,
327    }
328}
329
330fn validate_acyclic(relations: &[AssertionSupersession]) -> Result<(), KnowledgeError> {
331    let mut adjacency = HashMap::<Uuid, Vec<Uuid>>::new();
332    for row in relations {
333        adjacency
334            .entry(row.prior_assertion_uuid)
335            .or_default()
336            .push(row.replacement_assertion_uuid);
337    }
338    let mut visiting = HashSet::new();
339    let mut acyclic = HashSet::new();
340    for node in adjacency.keys().copied() {
341        if !visit(node, &adjacency, &mut visiting, &mut acyclic) {
342            return Err(invalid("assertion_supersession", "cycle detected"));
343        }
344    }
345    Ok(())
346}
347
348fn visit(
349    node: Uuid,
350    adjacency: &HashMap<Uuid, Vec<Uuid>>,
351    visiting: &mut HashSet<Uuid>,
352    acyclic: &mut HashSet<Uuid>,
353) -> bool {
354    if acyclic.contains(&node) {
355        return true;
356    }
357    if !visiting.insert(node) {
358        return false;
359    }
360    if adjacency.get(&node).is_some_and(|next| {
361        next.iter()
362            .any(|child| !visit(*child, adjacency, visiting, acyclic))
363    }) {
364        return false;
365    }
366    visiting.remove(&node);
367    acyclic.insert(node);
368    true
369}
370
371fn validate_relation(row: &AssertionSupersession) -> Result<(), KnowledgeError> {
372    if row.contract_version != ASSERTION_SUPERSESSION_CONTRACT_VERSION {
373        return Err(invalid(
374            "assertion_supersession.contract_version",
375            "unsupported version",
376        ));
377    }
378    AssertionSupersession::new(
379        row.supersession_uuid,
380        row.prior_assertion_uuid,
381        row.replacement_assertion_uuid,
382        row.status_event_uuid,
383        row.reasoning_uuid,
384        row.provenance_uuid,
385        row.recorded_at_micros,
386    )
387    .map(|_| ())
388}
389
390fn require_v7(value: Uuid, field: &'static str) -> Result<(), KnowledgeError> {
391    if value.get_version() != Some(Version::SortRand) {
392        return Err(invalid(field, "must be UUIDv7"));
393    }
394    require_uuid(value, field)
395}
396
397fn require_uuid(value: Uuid, field: &'static str) -> Result<(), KnowledgeError> {
398    if value.is_nil() {
399        return Err(invalid(field, "must not be nil"));
400    }
401    Ok(())
402}
403
404const fn invalid(field: &'static str, message: &'static str) -> KnowledgeError {
405    KnowledgeError::Invalid { field, message }
406}
407
408fn uuid_field(name: &str) -> Field {
409    Field::new(name, DataType::FixedSizeBinary(16), false)
410}
411
412fn fixed<'a>(
413    batch: &'a RecordBatch,
414    name: &'static str,
415) -> Result<&'a FixedSizeBinaryArray, KnowledgeError> {
416    batch
417        .column_by_name(name)
418        .and_then(|value| value.as_any().downcast_ref())
419        .ok_or_else(|| invalid(name, "column type mismatch"))
420}
421
422fn timestamp<'a>(
423    batch: &'a RecordBatch,
424    name: &'static str,
425) -> Result<&'a TimestampMicrosecondArray, KnowledgeError> {
426    batch
427        .column_by_name(name)
428        .and_then(|value| value.as_any().downcast_ref())
429        .ok_or_else(|| invalid(name, "column type mismatch"))
430}
431
432fn uint32<'a>(
433    batch: &'a RecordBatch,
434    name: &'static str,
435) -> Result<&'a UInt32Array, KnowledgeError> {
436    batch
437        .column_by_name(name)
438        .and_then(|value| value.as_any().downcast_ref())
439        .ok_or_else(|| invalid(name, "column type mismatch"))
440}
441
442fn uuid_at(
443    values: &FixedSizeBinaryArray,
444    row: usize,
445    field: &'static str,
446) -> Result<Uuid, KnowledgeError> {
447    Uuid::from_slice(values.value(row)).map_err(|_| invalid(field, "invalid UUID bytes"))
448}
449
450#[cfg(test)]
451mod tests {
452    use super::*;
453
454    fn uuid7(seed: u8) -> Uuid {
455        let mut bytes = [seed; 16];
456        bytes[6] = (bytes[6] & 0x0f) | 0x70;
457        bytes[8] = (bytes[8] & 0x3f) | 0x80;
458        Uuid::from_bytes(bytes)
459    }
460
461    fn relation(id: u8, prior: u8, replacement: u8, time: i64) -> AssertionSupersession {
462        AssertionSupersession::new(
463            uuid7(id),
464            uuid7(prior),
465            uuid7(replacement),
466            uuid7(id.wrapping_add(40)),
467            uuid7(id.wrapping_add(80)),
468            uuid7(id.wrapping_add(120)),
469            time,
470        )
471        .unwrap()
472    }
473
474    #[test]
475    fn chains_branches_order_round_trip_and_fingerprint_are_stable() {
476        let ledger = AssertionSupersessionLedger::new(vec![
477            relation(3, 10, 12, 2),
478            relation(2, 10, 11, 1),
479            relation(4, 11, 13, 2),
480        ])
481        .unwrap();
482        assert_eq!(ledger.relations()[0].supersession_uuid, uuid7(2));
483        let decoded =
484            AssertionSupersessionLedger::from_batches(&[ledger.batch().unwrap()]).unwrap();
485        assert_eq!(decoded, ledger);
486        assert_eq!(
487            decoded.relation_fingerprint(uuid7(3)).unwrap(),
488            ledger.relation_fingerprint(uuid7(3)).unwrap()
489        );
490    }
491
492    #[test]
493    fn self_links_cycles_duplicate_statuses_and_conflicts_fail() {
494        assert!(
495            AssertionSupersession::new(
496                uuid7(1),
497                uuid7(2),
498                uuid7(2),
499                uuid7(3),
500                uuid7(4),
501                uuid7(5),
502                1,
503            )
504            .is_err()
505        );
506        assert!(
507            AssertionSupersessionLedger::new(vec![relation(1, 10, 11, 1), relation(2, 11, 10, 2),])
508                .is_err()
509        );
510        let mut duplicate_status = relation(2, 11, 12, 2);
511        duplicate_status.status_event_uuid = uuid7(41);
512        assert!(
513            AssertionSupersessionLedger::new(vec![relation(1, 10, 11, 1), duplicate_status,])
514                .is_err()
515        );
516        let base = AssertionSupersessionLedger::new(vec![relation(1, 10, 11, 1)]).unwrap();
517        assert_eq!(base.merge(&base).unwrap(), base);
518        let conflict = AssertionSupersessionLedger::new(vec![relation(1, 10, 12, 1)]).unwrap();
519        assert!(matches!(
520            base.merge(&conflict),
521            Err(KnowledgeError::Conflict("supersession_uuid"))
522        ));
523    }
524}
graphforge_knowledge/supersession.rs

graphforge_knowledge/
supersession.rs
