graphforge_knowledge/
artifact_preference.rs

1//! Append-only preferred-representation events per research Source (#1349).
2
3use std::collections::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,
11};
12use uuid::Uuid;
13
14use crate::{
15    KNOWLEDGE_CAPABILITY_VERSION, KnowledgeError, MAX_KNOWLEDGE_ROWS, SchemaRegistryEntry,
16    check_limit, fixed_column, invalid, optional_fixed_16, require_schema, require_uuid,
17    require_v7, required_i64, required_text, required_u32, string_column, timestamp_column,
18    u32_column, uuid_at, uuid_field,
19};
20
21/// Immutable artifact-preference event contract.
22pub const ARTIFACT_PREFERENCE_CONTRACT_VERSION: u32 = 1;
23/// Maximum UTF-8 bytes for one bounded preference reason.
24pub const MAX_ARTIFACT_PREFERENCE_REASON_BYTES: usize = 4_096;
25
26/// Authoritative `knowledge/artifact_preference_events.parquet` schema.
27pub static ARTIFACT_PREFERENCE_SCHEMA: LazyLock<SchemaRef> = LazyLock::new(|| {
28    Arc::new(Schema::new(vec![
29        uuid_field("preference_event_uuid", false),
30        uuid_field("source_uuid", false),
31        uuid_field("artifact_uuid", false),
32        uuid_field("prior_artifact_uuid", true),
33        Field::new("reason", DataType::Utf8, false),
34        uuid_field("provenance_uuid", false),
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 ARTIFACT_PREFERENCE_SCHEMA_FINGERPRINT: LazyLock<[u8; 32]> = LazyLock::new(|| {
45    fingerprint(
46        CanonicalDomain::Schema,
47        CANONICAL_CONTRACT_VERSION,
48        b"artifact_preference/1|preference_event_uuid:fixed[16]:required|source_uuid:fixed[16]:required|artifact_uuid:fixed[16]:required|prior_artifact_uuid:fixed[16]:nullable|reason:utf8:required|provenance_uuid:fixed[16]:required|recorded_at:timestamp_us_utc:required|contract_version:u32:required",
49    )
50    .expect("registered artifact preference schema is within canonical bounds")
51});
52
53/// One append-only preferred-representation event.
54#[derive(Clone, Debug, Eq, PartialEq)]
55pub struct ArtifactPreferenceEvent {
56    /// Caller-supplied UUIDv7 identity and idempotency key.
57    pub preference_event_uuid: Uuid,
58    /// Parent Source whose preferred representation changed.
59    pub source_uuid: Uuid,
60    /// Newly preferred Artifact.
61    pub artifact_uuid: Uuid,
62    /// Previously preferred Artifact, when known.
63    pub prior_artifact_uuid: Option<Uuid>,
64    /// Bounded human-readable reason; never credentials or raw content.
65    pub reason: String,
66    /// Provenance event that published the preference.
67    pub provenance_uuid: Uuid,
68    /// Durable transaction time.
69    pub recorded_at_micros: i64,
70    /// Frozen record contract.
71    pub contract_version: u32,
72}
73
74impl ArtifactPreferenceEvent {
75    /// Construct one validated preference event.
76    pub fn new(
77        preference_event_uuid: Uuid,
78        source_uuid: Uuid,
79        artifact_uuid: Uuid,
80        prior_artifact_uuid: Option<Uuid>,
81        reason: String,
82        provenance_uuid: Uuid,
83        recorded_at_micros: i64,
84    ) -> Result<Self, KnowledgeError> {
85        let row = Self {
86            preference_event_uuid,
87            source_uuid,
88            artifact_uuid,
89            prior_artifact_uuid,
90            reason,
91            provenance_uuid,
92            recorded_at_micros,
93            contract_version: ARTIFACT_PREFERENCE_CONTRACT_VERSION,
94        };
95        validate_preference_event(&row)?;
96        Ok(row)
97    }
98}
99
100/// Validated append-only artifact-preference table.
101#[derive(Clone, Debug, Default, Eq, PartialEq)]
102pub struct ArtifactPreferenceLedger {
103    /// Events ordered by `(recorded_at, preference_event_uuid)`.
104    pub events: Vec<ArtifactPreferenceEvent>,
105}
106
107impl ArtifactPreferenceLedger {
108    /// Validate, sort, and construct a preference table.
109    pub fn new(mut events: Vec<ArtifactPreferenceEvent>) -> Result<Self, KnowledgeError> {
110        events.sort_by_key(|row| (row.recorded_at_micros, row.preference_event_uuid));
111        validate_preference_events(&events)?;
112        Ok(Self { events })
113    }
114
115    /// Merge staged rows into an existing ledger.
116    pub fn merge(&self, staged: &Self) -> Result<Self, KnowledgeError> {
117        let mut merged = self.events.clone();
118        for row in &staged.events {
119            if let Some(existing) = merged
120                .iter()
121                .find(|candidate| candidate.preference_event_uuid == row.preference_event_uuid)
122            {
123                if existing != row {
124                    return Err(KnowledgeError::Conflict("preference_event_uuid"));
125                }
126                continue;
127            }
128            merged.push(row.clone());
129        }
130        Self::new(merged)
131    }
132
133    /// Canonical semantic fingerprint over caller-supplied preference-event content.
134    ///
135    /// Excludes `prior_artifact_uuid` (ledger-derived), `provenance_uuid`, and
136    /// `recorded_at_micros` (clock-derived) so that a retry with the same caller-provided
137    /// content compares equal to the committed row regardless of when it was retried.
138    pub fn preference_fingerprint(
139        &self,
140        preference_event_uuid: Uuid,
141    ) -> Result<[u8; 32], KnowledgeError> {
142        let row = self
143            .events
144            .iter()
145            .find(|row| row.preference_event_uuid == preference_event_uuid)
146            .ok_or(KnowledgeError::Dangling("preference_event_uuid"))?;
147        let mut writer = CanonicalWriter::new();
148        writer.raw(b"GFPE")?;
149        writer.u32(ARTIFACT_PREFERENCE_CONTRACT_VERSION)?;
150        writer.raw(row.preference_event_uuid.as_bytes())?;
151        writer.raw(row.source_uuid.as_bytes())?;
152        writer.raw(row.artifact_uuid.as_bytes())?;
153        // `prior_artifact_uuid` is ledger-derived, not caller-supplied; omit from fingerprint.
154        writer.text(&row.reason)?;
155        Ok(fingerprint(
156            CanonicalDomain::ArtifactPreference,
157            CANONICAL_CONTRACT_VERSION,
158            &writer.finish(),
159        )?)
160    }
161
162    /// Canonical semantic fingerprint for caller-supplied preference request fields.
163    ///
164    /// Identical to `preference_fingerprint` but computed from raw caller-provided fields
165    /// rather than requiring a committed ledger row. Used to compare an incoming request
166    /// against the already-committed row before constructing the full staged event.
167    pub fn preference_request_fingerprint(
168        preference_event_uuid: Uuid,
169        source_uuid: Uuid,
170        artifact_uuid: Uuid,
171        reason: &str,
172    ) -> Result<[u8; 32], KnowledgeError> {
173        let mut writer = CanonicalWriter::new();
174        writer.raw(b"GFPE")?;
175        writer.u32(ARTIFACT_PREFERENCE_CONTRACT_VERSION)?;
176        writer.raw(preference_event_uuid.as_bytes())?;
177        writer.raw(source_uuid.as_bytes())?;
178        writer.raw(artifact_uuid.as_bytes())?;
179        // `prior_artifact_uuid` is ledger-derived, not caller-supplied; omit.
180        writer.text(reason)?;
181        Ok(fingerprint(
182            CanonicalDomain::ArtifactPreference,
183            CANONICAL_CONTRACT_VERSION,
184            &writer.finish(),
185        )?)
186    }
187
188    /// Return the current preferred Artifact for one Source, if any.
189    #[must_use]
190    pub fn current_preferred_artifact(&self, source_uuid: Uuid) -> Option<Uuid> {
191        self.events
192            .iter()
193            .filter(|row| row.source_uuid == source_uuid)
194            .max_by_key(|row| (row.recorded_at_micros, row.preference_event_uuid))
195            .map(|row| row.artifact_uuid)
196    }
197
198    /// Encode the authoritative Arrow batch.
199    pub fn batch(&self) -> Result<RecordBatch, KnowledgeError> {
200        preference_batch(&self.events)
201    }
202
203    /// Decode one or more Arrow batches.
204    pub fn from_batches(batches: &[RecordBatch]) -> Result<Self, KnowledgeError> {
205        let mut events = Vec::new();
206        for batch in batches {
207            require_schema(
208                batch,
209                &ARTIFACT_PREFERENCE_SCHEMA,
210                "artifact_preference_events.schema",
211            )?;
212            let ids = fixed_column(batch, "preference_event_uuid")?;
213            let sources = fixed_column(batch, "source_uuid")?;
214            let artifacts = fixed_column(batch, "artifact_uuid")?;
215            let priors = fixed_column(batch, "prior_artifact_uuid")?;
216            let reasons = string_column(batch, "reason")?;
217            let provenance = fixed_column(batch, "provenance_uuid")?;
218            let recorded = timestamp_column(batch, "recorded_at")?;
219            let versions = u32_column(batch, "contract_version")?;
220            for row in 0..batch.num_rows() {
221                events.push(ArtifactPreferenceEvent {
222                    preference_event_uuid: uuid_at(ids, row, "preference_event_uuid")?,
223                    source_uuid: uuid_at(sources, row, "source_uuid")?,
224                    artifact_uuid: uuid_at(artifacts, row, "artifact_uuid")?,
225                    prior_artifact_uuid: optional_fixed_16(priors, row, "prior_artifact_uuid")?,
226                    reason: required_text(reasons, row, "reason")?.to_string(),
227                    provenance_uuid: uuid_at(provenance, row, "provenance_uuid")?,
228                    recorded_at_micros: required_i64(recorded, row, "recorded_at")?,
229                    contract_version: required_u32(versions, row, "contract_version")?,
230                });
231            }
232        }
233        Self::new(events)
234    }
235}
236
237pub(crate) fn schema_registry_entry() -> SchemaRegistryEntry {
238    SchemaRegistryEntry {
239        capability_id: "knowledge",
240        capability_version: KNOWLEDGE_CAPABILITY_VERSION,
241        record_family: "artifact_preference_events",
242        record_version: ARTIFACT_PREFERENCE_CONTRACT_VERSION,
243        schema: Arc::clone(&ARTIFACT_PREFERENCE_SCHEMA),
244        schema_fingerprint: *ARTIFACT_PREFERENCE_SCHEMA_FINGERPRINT,
245        enum_registry_versions: &[],
246        sort_key: &["recorded_at", "preference_event_uuid"],
247        diff_identity_fields: &["preference_event_uuid"],
248        diff_record_uuid_field: Some("preference_event_uuid"),
249        fingerprint_domain: CanonicalDomain::ArtifactPreference,
250        owner: "graphforge-knowledge",
251        implementation_issue: 1349,
252        max_rows: MAX_KNOWLEDGE_ROWS,
253    }
254}
255
256fn validate_preference_event(row: &ArtifactPreferenceEvent) -> Result<(), KnowledgeError> {
257    if row.contract_version != ARTIFACT_PREFERENCE_CONTRACT_VERSION {
258        return Err(invalid(
259            "preference.contract_version",
260            "unsupported version",
261        ));
262    }
263    require_v7(row.preference_event_uuid, "preference_event_uuid")?;
264    require_v7(row.source_uuid, "source_uuid")?;
265    require_v7(row.artifact_uuid, "artifact_uuid")?;
266    if let Some(prior) = row.prior_artifact_uuid {
267        require_v7(prior, "prior_artifact_uuid")?;
268        if prior == row.artifact_uuid {
269            return Err(invalid(
270                "prior_artifact_uuid",
271                "must differ from artifact_uuid",
272            ));
273        }
274    }
275    require_uuid(row.provenance_uuid, "provenance_uuid")?;
276    validate_bounded_text(
277        "preference.reason",
278        &row.reason,
279        MAX_ARTIFACT_PREFERENCE_REASON_BYTES,
280    )?;
281    Ok(())
282}
283
284fn validate_preference_events(events: &[ArtifactPreferenceEvent]) -> Result<(), KnowledgeError> {
285    check_limit("artifact_preference_events", events.len())?;
286    let mut ids = HashSet::with_capacity(events.len());
287    for row in events {
288        validate_preference_event(row)?;
289        if !ids.insert(row.preference_event_uuid) {
290            return Err(KnowledgeError::Duplicate("preference_event_uuid"));
291        }
292    }
293    Ok(())
294}
295
296fn validate_bounded_text(
297    field: &'static str,
298    value: &str,
299    limit: usize,
300) -> Result<(), KnowledgeError> {
301    if value.is_empty() {
302        return Err(invalid(field, "value must not be empty"));
303    }
304    if value.len() > limit {
305        return Err(KnowledgeError::Limit {
306            participant: field,
307            observed: value.len(),
308            limit,
309        });
310    }
311    Ok(())
312}
313
314fn preference_batch(rows: &[ArtifactPreferenceEvent]) -> Result<RecordBatch, KnowledgeError> {
315    let mut ids = FixedSizeBinaryBuilder::with_capacity(rows.len(), 16);
316    let mut sources = FixedSizeBinaryBuilder::with_capacity(rows.len(), 16);
317    let mut artifacts = FixedSizeBinaryBuilder::with_capacity(rows.len(), 16);
318    let mut priors = FixedSizeBinaryBuilder::with_capacity(rows.len(), 16);
319    let mut provenance = FixedSizeBinaryBuilder::with_capacity(rows.len(), 16);
320    for row in rows {
321        ids.append_value(row.preference_event_uuid.as_bytes())?;
322        sources.append_value(row.source_uuid.as_bytes())?;
323        artifacts.append_value(row.artifact_uuid.as_bytes())?;
324        match row.prior_artifact_uuid {
325            Some(value) => priors.append_value(value.as_bytes())?,
326            None => priors.append_null(),
327        }
328        provenance.append_value(row.provenance_uuid.as_bytes())?;
329    }
330    RecordBatch::try_new(
331        Arc::clone(&ARTIFACT_PREFERENCE_SCHEMA),
332        vec![
333            Arc::new(ids.finish()),
334            Arc::new(sources.finish()),
335            Arc::new(artifacts.finish()),
336            Arc::new(priors.finish()),
337            Arc::new(StringArray::from_iter_values(
338                rows.iter().map(|row| row.reason.as_str()),
339            )),
340            Arc::new(provenance.finish()),
341            Arc::new(
342                TimestampMicrosecondArray::from_iter_values(
343                    rows.iter().map(|row| row.recorded_at_micros),
344                )
345                .with_timezone("UTC"),
346            ),
347            Arc::new(UInt32Array::from_iter_values(
348                rows.iter().map(|row| row.contract_version),
349            )),
350        ],
351    )
352    .map_err(Into::into)
353}
354
355#[cfg(test)]
356mod tests {
357    use super::*;
358    use crate::tests::uuid7;
359
360    #[test]
361    fn preference_round_trip_and_current_selection() {
362        let first = ArtifactPreferenceEvent::new(
363            uuid7(1),
364            uuid7(2),
365            uuid7(3),
366            None,
367            "initial OCR".into(),
368            uuid7(4),
369            100,
370        )
371        .unwrap();
372        let second = ArtifactPreferenceEvent::new(
373            uuid7(5),
374            uuid7(2),
375            uuid7(6),
376            Some(uuid7(3)),
377            "better OCR".into(),
378            uuid7(7),
379            200,
380        )
381        .unwrap();
382        let ledger = ArtifactPreferenceLedger::new(vec![first, second]).unwrap();
383        let reopened = ArtifactPreferenceLedger::from_batches(&[ledger.batch().unwrap()]).unwrap();
384        assert_eq!(reopened, ledger);
385        assert_eq!(ledger.current_preferred_artifact(uuid7(2)), Some(uuid7(6)));
386    }
387}
graphforge_knowledge/artifact_preference.rs

graphforge_knowledge/
artifact_preference.rs
