graphforge_knowledge/
derivation.rs

1//! Immutable directed derivation edges for research lineage (#1349).
2
3use std::collections::{HashMap, HashSet};
4use std::sync::{Arc, LazyLock};
5
6use arrow::array::{FixedSizeBinaryBuilder, StringArray, TimestampMicrosecondArray, UInt32Array};
7use arrow::datatypes::{DataType, Field, Schema, SchemaRef, TimeUnit};
8use arrow::record_batch::RecordBatch;
9use graphforge_core::canonical::{
10    CANONICAL_CONTRACT_VERSION, CanonicalDomain, CanonicalWriter, fingerprint, uuid_v8,
11};
12use uuid::Uuid;
13
14use crate::{
15    KNOWLEDGE_CAPABILITY_VERSION, KnowledgeError, MAX_KNOWLEDGE_ROWS, SchemaRegistryEntry,
16    check_limit, fixed_column, invalid, require_schema, require_uuid, required_i64, required_text,
17    required_u32, string_column, timestamp_column, u32_column, uuid_at, uuid_field,
18};
19
20/// Immutable derivation-edge record contract.
21pub const ARTIFACT_DERIVATION_CONTRACT_VERSION: u32 = 1;
22/// Closed derivation-subject-kind registry version.
23pub const DERIVATION_SUBJECT_KIND_REGISTRY_VERSION: u32 = 1;
24/// Closed derivation-role registry version.
25pub const DERIVATION_ROLE_REGISTRY_VERSION: u32 = 1;
26
27/// Authoritative `knowledge/artifact_derivations.parquet` schema.
28pub static ARTIFACT_DERIVATION_SCHEMA: LazyLock<SchemaRef> = LazyLock::new(|| {
29    Arc::new(Schema::new(vec![
30        uuid_field("derivation_uuid", false),
31        uuid_field("output_uuid", false),
32        Field::new("output_kind", DataType::Utf8, false),
33        uuid_field("input_uuid", false),
34        Field::new("input_kind", DataType::Utf8, false),
35        Field::new("derivation_role", DataType::Utf8, false),
36        Field::new("ordinal", DataType::UInt32, false),
37        uuid_field("provenance_uuid", false),
38        Field::new(
39            "recorded_at",
40            DataType::Timestamp(TimeUnit::Microsecond, Some("UTC".into())),
41            false,
42        ),
43        Field::new("contract_version", DataType::UInt32, false),
44    ]))
45});
46
47static ARTIFACT_DERIVATION_SCHEMA_FINGERPRINT: LazyLock<[u8; 32]> = LazyLock::new(|| {
48    fingerprint(
49        CanonicalDomain::Schema,
50        CANONICAL_CONTRACT_VERSION,
51        b"artifact_derivation/1|derivation_uuid:fixed[16]:required|output_uuid:fixed[16]:required|output_kind:utf8:required|input_uuid:fixed[16]:required|input_kind:utf8:required|derivation_role:utf8:required|ordinal:u32:required|provenance_uuid:fixed[16]:required|recorded_at:timestamp_us_utc:required|contract_version:u32:required",
52    )
53    .expect("registered artifact derivation schema is within canonical bounds")
54});
55
56/// Closed UUID subject kind for one derivation endpoint.
57#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq, serde::Serialize, serde::Deserialize)]
58#[serde(rename_all = "snake_case")]
59pub enum DerivationSubjectKind {
60    /// Immutable research Source.
61    Source,
62    /// Immutable research Artifact.
63    Artifact,
64    /// Public graph node UUID.
65    Node,
66    /// Public graph edge UUID.
67    Edge,
68    /// Immutable assertion UUID.
69    Assertion,
70    /// Evidence-link UUID.
71    EvidenceLink,
72    /// Algorithm-run UUID.
73    AlgorithmRun,
74}
75
76impl DerivationSubjectKind {
77    /// Canonical persisted spelling.
78    #[must_use]
79    pub const fn as_str(self) -> &'static str {
80        match self {
81            Self::Source => "source",
82            Self::Artifact => "artifact",
83            Self::Node => "node",
84            Self::Edge => "edge",
85            Self::Assertion => "assertion",
86            Self::EvidenceLink => "evidence_link",
87            Self::AlgorithmRun => "algorithm_run",
88        }
89    }
90
91    pub(super) fn parse(value: &str) -> Result<Self, KnowledgeError> {
92        match value {
93            "source" => Ok(Self::Source),
94            "artifact" => Ok(Self::Artifact),
95            "node" => Ok(Self::Node),
96            "edge" => Ok(Self::Edge),
97            "assertion" => Ok(Self::Assertion),
98            "evidence_link" => Ok(Self::EvidenceLink),
99            "algorithm_run" => Ok(Self::AlgorithmRun),
100            _ => Err(invalid("derivation_subject_kind", "unknown closed value")),
101        }
102    }
103}
104
105/// Closed role for one derivation edge.
106#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq, serde::Serialize, serde::Deserialize)]
107#[serde(rename_all = "snake_case")]
108pub enum DerivationRole {
109    /// Ordered input to one output.
110    Input,
111}
112
113impl DerivationRole {
114    /// Canonical persisted spelling.
115    #[must_use]
116    pub const fn as_str(self) -> &'static str {
117        match self {
118            Self::Input => "input",
119        }
120    }
121
122    pub(super) fn parse(value: &str) -> Result<Self, KnowledgeError> {
123        match value {
124            "input" => Ok(Self::Input),
125            _ => Err(invalid("derivation_role", "unknown closed value")),
126        }
127    }
128}
129
130/// One immutable directed derivation edge.
131#[derive(Clone, Debug, Eq, PartialEq)]
132pub struct ArtifactDerivation {
133    /// Deterministic UUIDv8 identity derived from canonical edge bytes.
134    pub derivation_uuid: Uuid,
135    /// Derived subject UUID.
136    pub output_uuid: Uuid,
137    /// Closed output subject kind.
138    pub output_kind: DerivationSubjectKind,
139    /// Input subject UUID.
140    pub input_uuid: Uuid,
141    /// Closed input subject kind.
142    pub input_kind: DerivationSubjectKind,
143    /// Closed derivation role.
144    pub derivation_role: DerivationRole,
145    /// Per-output input ordering.
146    pub ordinal: u32,
147    /// Provenance event that published the edge.
148    pub provenance_uuid: Uuid,
149    /// Durable transaction time.
150    pub recorded_at_micros: i64,
151    /// Frozen record contract.
152    pub contract_version: u32,
153}
154
155impl ArtifactDerivation {
156    /// Construct one validated immutable derivation edge.
157    #[allow(clippy::too_many_arguments)]
158    pub fn new(
159        output_uuid: Uuid,
160        output_kind: DerivationSubjectKind,
161        input_uuid: Uuid,
162        input_kind: DerivationSubjectKind,
163        derivation_role: DerivationRole,
164        ordinal: u32,
165        provenance_uuid: Uuid,
166        recorded_at_micros: i64,
167    ) -> Result<Self, KnowledgeError> {
168        let derivation_uuid = derivation_identity_uuid(
169            output_uuid,
170            output_kind,
171            input_uuid,
172            input_kind,
173            derivation_role,
174            ordinal,
175        )?;
176        let row = Self {
177            derivation_uuid,
178            output_uuid,
179            output_kind,
180            input_uuid,
181            input_kind,
182            derivation_role,
183            ordinal,
184            provenance_uuid,
185            recorded_at_micros,
186            contract_version: ARTIFACT_DERIVATION_CONTRACT_VERSION,
187        };
188        validate_derivation(&row)?;
189        Ok(row)
190    }
191}
192
193/// Validated immutable derivation-edge table.
194#[derive(Clone, Debug, Default, Eq, PartialEq)]
195pub struct ArtifactDerivationLedger {
196    /// Edges ordered by `(recorded_at, derivation_uuid)`.
197    pub derivations: Vec<ArtifactDerivation>,
198}
199
200impl ArtifactDerivationLedger {
201    /// Validate, sort, and construct a derivation table.
202    pub fn new(mut derivations: Vec<ArtifactDerivation>) -> Result<Self, KnowledgeError> {
203        derivations.sort_by_key(|row| (row.recorded_at_micros, row.derivation_uuid));
204        validate_derivations(&derivations)?;
205        Ok(Self { derivations })
206    }
207
208    /// Merge staged rows into an existing ledger.
209    pub fn merge(&self, staged: &Self) -> Result<Self, KnowledgeError> {
210        let mut merged = self.derivations.clone();
211        for row in &staged.derivations {
212            if let Some(existing) = merged
213                .iter()
214                .find(|candidate| candidate.derivation_uuid == row.derivation_uuid)
215            {
216                if existing != row {
217                    return Err(KnowledgeError::Conflict("derivation_uuid"));
218                }
219                continue;
220            }
221            merged.push(row.clone());
222        }
223        Self::new(merged)
224    }
225
226    /// Encode the authoritative Arrow batch.
227    pub fn batch(&self) -> Result<RecordBatch, KnowledgeError> {
228        derivation_batch(&self.derivations)
229    }
230
231    /// Decode one or more Arrow batches.
232    pub fn from_batches(batches: &[RecordBatch]) -> Result<Self, KnowledgeError> {
233        let mut derivations = Vec::new();
234        for batch in batches {
235            require_schema(
236                batch,
237                &ARTIFACT_DERIVATION_SCHEMA,
238                "artifact_derivations.schema",
239            )?;
240            let ids = fixed_column(batch, "derivation_uuid")?;
241            let outputs = fixed_column(batch, "output_uuid")?;
242            let output_kinds = string_column(batch, "output_kind")?;
243            let inputs = fixed_column(batch, "input_uuid")?;
244            let input_kinds = string_column(batch, "input_kind")?;
245            let roles = string_column(batch, "derivation_role")?;
246            let ordinals = u32_column(batch, "ordinal")?;
247            let provenance = fixed_column(batch, "provenance_uuid")?;
248            let recorded = timestamp_column(batch, "recorded_at")?;
249            let versions = u32_column(batch, "contract_version")?;
250            for row in 0..batch.num_rows() {
251                derivations.push(ArtifactDerivation {
252                    derivation_uuid: uuid_at(ids, row, "derivation_uuid")?,
253                    output_uuid: uuid_at(outputs, row, "output_uuid")?,
254                    output_kind: DerivationSubjectKind::parse(required_text(
255                        output_kinds,
256                        row,
257                        "output_kind",
258                    )?)?,
259                    input_uuid: uuid_at(inputs, row, "input_uuid")?,
260                    input_kind: DerivationSubjectKind::parse(required_text(
261                        input_kinds,
262                        row,
263                        "input_kind",
264                    )?)?,
265                    derivation_role: DerivationRole::parse(required_text(
266                        roles,
267                        row,
268                        "derivation_role",
269                    )?)?,
270                    ordinal: required_u32(ordinals, row, "ordinal")?,
271                    provenance_uuid: uuid_at(provenance, row, "provenance_uuid")?,
272                    recorded_at_micros: required_i64(recorded, row, "recorded_at")?,
273                    contract_version: required_u32(versions, row, "contract_version")?,
274                });
275            }
276        }
277        Self::new(derivations)
278    }
279}
280
281pub(crate) fn schema_registry_entry() -> SchemaRegistryEntry {
282    SchemaRegistryEntry {
283        capability_id: "knowledge",
284        capability_version: KNOWLEDGE_CAPABILITY_VERSION,
285        record_family: "artifact_derivations",
286        record_version: ARTIFACT_DERIVATION_CONTRACT_VERSION,
287        schema: Arc::clone(&ARTIFACT_DERIVATION_SCHEMA),
288        schema_fingerprint: *ARTIFACT_DERIVATION_SCHEMA_FINGERPRINT,
289        enum_registry_versions: &[
290            ("output_kind", DERIVATION_SUBJECT_KIND_REGISTRY_VERSION),
291            ("input_kind", DERIVATION_SUBJECT_KIND_REGISTRY_VERSION),
292            ("derivation_role", DERIVATION_ROLE_REGISTRY_VERSION),
293        ],
294        sort_key: &["recorded_at", "derivation_uuid"],
295        diff_identity_fields: &["derivation_uuid"],
296        diff_record_uuid_field: Some("derivation_uuid"),
297        fingerprint_domain: CanonicalDomain::ArtifactDerivation,
298        owner: "graphforge-knowledge",
299        implementation_issue: 1349,
300        max_rows: MAX_KNOWLEDGE_ROWS,
301    }
302}
303
304fn derivation_identity_uuid(
305    output_uuid: Uuid,
306    output_kind: DerivationSubjectKind,
307    input_uuid: Uuid,
308    input_kind: DerivationSubjectKind,
309    derivation_role: DerivationRole,
310    ordinal: u32,
311) -> Result<Uuid, KnowledgeError> {
312    let mut writer = CanonicalWriter::new();
313    writer.raw(output_uuid.as_bytes())?;
314    writer.text(output_kind.as_str())?;
315    writer.raw(input_uuid.as_bytes())?;
316    writer.text(input_kind.as_str())?;
317    writer.text(derivation_role.as_str())?;
318    writer.u32(ordinal)?;
319    let digest = fingerprint(
320        CanonicalDomain::ArtifactDerivation,
321        ARTIFACT_DERIVATION_CONTRACT_VERSION,
322        &writer.finish(),
323    )?;
324    Ok(uuid_v8(digest))
325}
326
327fn validate_derivation(row: &ArtifactDerivation) -> Result<(), KnowledgeError> {
328    if row.contract_version != ARTIFACT_DERIVATION_CONTRACT_VERSION {
329        return Err(invalid(
330            "derivation.contract_version",
331            "unsupported version",
332        ));
333    }
334    require_uuid(row.derivation_uuid, "derivation_uuid")?;
335    require_uuid(row.output_uuid, "output_uuid")?;
336    require_uuid(row.input_uuid, "input_uuid")?;
337    require_uuid(row.provenance_uuid, "provenance_uuid")?;
338    if row.output_uuid == row.input_uuid && row.output_kind == row.input_kind {
339        return Err(invalid("derivation", "self-loop is forbidden"));
340    }
341    let expected = derivation_identity_uuid(
342        row.output_uuid,
343        row.output_kind,
344        row.input_uuid,
345        row.input_kind,
346        row.derivation_role,
347        row.ordinal,
348    )?;
349    if row.derivation_uuid != expected {
350        return Err(invalid("derivation_uuid", "identity mismatch"));
351    }
352    Ok(())
353}
354
355fn validate_derivations(derivations: &[ArtifactDerivation]) -> Result<(), KnowledgeError> {
356    check_limit("artifact_derivations", derivations.len())?;
357    let mut ids = HashSet::with_capacity(derivations.len());
358    let mut ordinals: HashMap<(Uuid, DerivationSubjectKind), HashSet<u32>> = HashMap::new();
359    for row in derivations {
360        validate_derivation(row)?;
361        if !ids.insert(row.derivation_uuid) {
362            return Err(KnowledgeError::Duplicate("derivation_uuid"));
363        }
364        ordinals
365            .entry((row.output_uuid, row.output_kind))
366            .or_default()
367            .insert(row.ordinal);
368    }
369    for ((output_uuid, output_kind), seen) in &ordinals {
370        if seen.len()
371            != derivations
372                .iter()
373                .filter(|row| row.output_uuid == *output_uuid && row.output_kind == *output_kind)
374                .count()
375        {
376            return Err(KnowledgeError::Duplicate("ordinal"));
377        }
378    }
379    detect_cycles(derivations)?;
380    Ok(())
381}
382
383fn detect_cycles(derivations: &[ArtifactDerivation]) -> Result<(), KnowledgeError> {
384    let mut adjacency: HashMap<(Uuid, DerivationSubjectKind), Vec<(Uuid, DerivationSubjectKind)>> =
385        HashMap::new();
386    for row in derivations {
387        adjacency
388            .entry((row.input_uuid, row.input_kind))
389            .or_default()
390            .push((row.output_uuid, row.output_kind));
391    }
392    let mut visiting = HashSet::new();
393    let mut visited = HashSet::new();
394    for start in adjacency.keys().copied().collect::<Vec<_>>() {
395        if visited.contains(&start) {
396            continue;
397        }
398        visiting.clear();
399        if dfs_cycle(start, &adjacency, &mut visiting, &mut visited) {
400            return Err(invalid("derivation", "cycle detected"));
401        }
402    }
403    Ok(())
404}
405
406fn dfs_cycle(
407    node: (Uuid, DerivationSubjectKind),
408    adjacency: &HashMap<(Uuid, DerivationSubjectKind), Vec<(Uuid, DerivationSubjectKind)>>,
409    visiting: &mut HashSet<(Uuid, DerivationSubjectKind)>,
410    visited: &mut HashSet<(Uuid, DerivationSubjectKind)>,
411) -> bool {
412    if visiting.contains(&node) {
413        return true;
414    }
415    if visited.contains(&node) {
416        return false;
417    }
418    visiting.insert(node);
419    if let Some(neighbors) = adjacency.get(&node) {
420        for neighbor in neighbors {
421            if dfs_cycle(*neighbor, adjacency, visiting, visited) {
422                return true;
423            }
424        }
425    }
426    visiting.remove(&node);
427    visited.insert(node);
428    false
429}
430
431fn derivation_batch(rows: &[ArtifactDerivation]) -> Result<RecordBatch, KnowledgeError> {
432    let mut ids = FixedSizeBinaryBuilder::with_capacity(rows.len(), 16);
433    let mut outputs = FixedSizeBinaryBuilder::with_capacity(rows.len(), 16);
434    let mut inputs = FixedSizeBinaryBuilder::with_capacity(rows.len(), 16);
435    let mut provenance = FixedSizeBinaryBuilder::with_capacity(rows.len(), 16);
436    for row in rows {
437        ids.append_value(row.derivation_uuid.as_bytes())?;
438        outputs.append_value(row.output_uuid.as_bytes())?;
439        inputs.append_value(row.input_uuid.as_bytes())?;
440        provenance.append_value(row.provenance_uuid.as_bytes())?;
441    }
442    RecordBatch::try_new(
443        Arc::clone(&ARTIFACT_DERIVATION_SCHEMA),
444        vec![
445            Arc::new(ids.finish()),
446            Arc::new(outputs.finish()),
447            Arc::new(StringArray::from_iter_values(
448                rows.iter().map(|row| row.output_kind.as_str()),
449            )),
450            Arc::new(inputs.finish()),
451            Arc::new(StringArray::from_iter_values(
452                rows.iter().map(|row| row.input_kind.as_str()),
453            )),
454            Arc::new(StringArray::from_iter_values(
455                rows.iter().map(|row| row.derivation_role.as_str()),
456            )),
457            Arc::new(UInt32Array::from_iter_values(
458                rows.iter().map(|row| row.ordinal),
459            )),
460            Arc::new(provenance.finish()),
461            Arc::new(
462                TimestampMicrosecondArray::from_iter_values(
463                    rows.iter().map(|row| row.recorded_at_micros),
464                )
465                .with_timezone("UTC"),
466            ),
467            Arc::new(UInt32Array::from_iter_values(
468                rows.iter().map(|row| row.contract_version),
469            )),
470        ],
471    )
472    .map_err(Into::into)
473}
474
475#[cfg(test)]
476mod tests {
477    use super::*;
478    use crate::tests::uuid7;
479
480    #[test]
481    fn derivation_round_trip_and_cycle_rejection() {
482        let scan = ArtifactDerivation::new(
483            uuid7(10),
484            DerivationSubjectKind::Artifact,
485            uuid7(1),
486            DerivationSubjectKind::Artifact,
487            DerivationRole::Input,
488            0,
489            uuid7(20),
490            100,
491        )
492        .unwrap();
493        let ocr = ArtifactDerivation::new(
494            uuid7(11),
495            DerivationSubjectKind::Artifact,
496            uuid7(10),
497            DerivationSubjectKind::Artifact,
498            DerivationRole::Input,
499            0,
500            uuid7(21),
501            200,
502        )
503        .unwrap();
504        let ledger = ArtifactDerivationLedger::new(vec![scan.clone(), ocr.clone()]).unwrap();
505        let reopened = ArtifactDerivationLedger::from_batches(&[ledger.batch().unwrap()]).unwrap();
506        assert_eq!(reopened, ledger);
507        let cycle = ArtifactDerivation::new(
508            uuid7(1),
509            DerivationSubjectKind::Artifact,
510            uuid7(11),
511            DerivationSubjectKind::Artifact,
512            DerivationRole::Input,
513            0,
514            uuid7(22),
515            300,
516        )
517        .unwrap();
518        assert!(matches!(
519            ArtifactDerivationLedger::new(vec![scan, ocr, cycle]),
520            Err(KnowledgeError::Invalid { .. })
521        ));
522    }
523
524    #[test]
525    fn derivation_registries_round_trip_and_reject_unknown_tokens() {
526        for kind in [
527            DerivationSubjectKind::Source,
528            DerivationSubjectKind::Artifact,
529            DerivationSubjectKind::Node,
530            DerivationSubjectKind::Edge,
531            DerivationSubjectKind::Assertion,
532            DerivationSubjectKind::EvidenceLink,
533            DerivationSubjectKind::AlgorithmRun,
534        ] {
535            assert_eq!(DerivationSubjectKind::parse(kind.as_str()).unwrap(), kind);
536        }
537        assert_eq!(
538            DerivationRole::parse(DerivationRole::Input.as_str()).unwrap(),
539            DerivationRole::Input
540        );
541        assert!(DerivationSubjectKind::parse("passage").is_err());
542        assert!(DerivationRole::parse("output").is_err());
543    }
544}
graphforge_knowledge/derivation.rs

graphforge_knowledge/
derivation.rs
