//! Append-only epistemic attachments connecting resolved interpretation to completed knowledge runs.
use std::collections::{HashMap, HashSet};
use std::sync::{Arc, LazyLock};
#[cfg(test)]
use std::{cell::Cell, thread_local};
use arrow::array::{
Array, BinaryArray, BinaryBuilder, FixedSizeBinaryArray, FixedSizeBinaryBuilder, ListArray,
ListBuilder, TimestampMicrosecondArray, TimestampMicrosecondBuilder, UInt32Array,
UInt32Builder,
};
use arrow::datatypes::{DataType, Field, Schema, SchemaRef, TimeUnit};
use arrow::record_batch::RecordBatch;
use graphforge_core::canonical::{
CANONICAL_CONTRACT_VERSION, CanonicalDomain, CanonicalWriter, MAX_CANONICAL_BINARY_BYTES,
fingerprint,
};
use uuid::{Uuid, Version};
use crate::{
EPISTEMIC_CAPABILITY_VERSION, KnowledgeError, MAX_KNOWLEDGE_ROWS, SchemaRegistryEntry,
};
/// Frozen `belief_projection_attachment/1` record contract.
pub const BELIEF_PROJECTION_ATTACHMENT_CONTRACT_VERSION: u32 = 1;
/// Authoritative `algorithm_interpretation_attachments` schema.
pub static ALGORITHM_INTERPRETATION_ATTACHMENT_SCHEMA: LazyLock<SchemaRef> = LazyLock::new(|| {
let uuid_list = DataType::List(Arc::new(Field::new(
"item",
DataType::FixedSizeBinary(16),
false,
)));
Arc::new(Schema::new(vec![
uuid_field("attachment_uuid", false),
uuid_field("run_uuid", false),
uuid_field("source_generation_uuid", false),
timestamp_field("transaction_cutoff", false),
timestamp_field("valid_time", true),
Field::new("policy_version", DataType::UInt32, false),
Field::new("policy_bytes", DataType::Binary, false),
fingerprint_field("policy_fingerprint", false),
fingerprint_field("snapshot_fingerprint", false),
fingerprint_field("valid_time_fingerprint", true),
fingerprint_field("graph_content_fingerprint", false),
fingerprint_field("descriptor_fingerprint", false),
Field::new("source_record_uuids", uuid_list, false),
uuid_field("provenance_uuid", false),
timestamp_field("recorded_at", false),
Field::new("contract_version", DataType::UInt32, false),
]))
});
static SCHEMA_FINGERPRINT: LazyLock<[u8; 32]> = LazyLock::new(|| {
fingerprint(
CanonicalDomain::Schema,
CANONICAL_CONTRACT_VERSION,
b"belief_projection_attachment/1|attachment_uuid:fixed[16]:required|run_uuid:fixed[16]:required|source_generation_uuid:fixed[16]:required|transaction_cutoff:timestamp_us_utc:required|valid_time:timestamp_us_utc:nullable|policy_version:u32:required|policy_bytes:binary:required|policy_fingerprint:fixed[32]:required|snapshot_fingerprint:fixed[32]:required|valid_time_fingerprint:fixed[32]:nullable|graph_content_fingerprint:fixed[32]:required|descriptor_fingerprint:fixed[32]:required|source_record_uuids:list<fixed[16]>:required|provenance_uuid:fixed[16]:required|recorded_at:timestamp_us_utc:required|contract_version:u32:required",
)
.expect("registered belief-projection attachment schema is bounded")
});
/// One immutable interpretation attachment for a completed algorithm run.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct BeliefProjectionAttachment {
/// Stable attachment identity and idempotency key.
pub attachment_uuid: Uuid,
/// Completed knowledge run identity.
pub run_uuid: Uuid,
/// Source project generation pinned during resolution.
pub source_generation_uuid: Uuid,
/// Mandatory transaction-time cutoff.
pub transaction_cutoff_micros: i64,
/// Optional valid time applied after transaction reconstruction.
pub valid_time_micros: Option<i64>,
/// Explicit resolved-belief policy version.
pub policy_version: u32,
/// Canonical language-neutral policy bytes.
pub policy_bytes: Vec<u8>,
/// Canonical fingerprint of `policy_bytes`.
pub policy_fingerprint: [u8; 32],
/// Fingerprint of the transaction-time epistemic snapshot.
pub snapshot_fingerprint: [u8; 32],
/// Fingerprint of the optional valid-time interpretation.
pub valid_time_fingerprint: Option<[u8; 32]>,
/// Canonical logical graph-content fingerprint.
pub graph_content_fingerprint: [u8; 32],
/// Neutral algorithm invocation-descriptor fingerprint.
pub descriptor_fingerprint: [u8; 32],
/// Sorted and deduplicated decision-relevant epistemic records.
pub source_record_uuids: Vec<Uuid>,
/// Producing provenance event.
pub provenance_uuid: Uuid,
/// Mandatory attachment transaction time.
pub recorded_at_micros: i64,
/// Frozen record contract.
pub contract_version: u32,
}
impl BeliefProjectionAttachment {
/// Validate and construct an attachment, canonicalizing source UUIDs.
#[allow(clippy::too_many_arguments)]
pub fn new(
attachment_uuid: Uuid,
run_uuid: Uuid,
source_generation_uuid: Uuid,
transaction_cutoff_micros: i64,
valid_time_micros: Option<i64>,
policy_version: u32,
policy_bytes: Vec<u8>,
snapshot_fingerprint: [u8; 32],
valid_time_fingerprint: Option<[u8; 32]>,
graph_content_fingerprint: [u8; 32],
descriptor_fingerprint: [u8; 32],
mut source_record_uuids: Vec<Uuid>,
provenance_uuid: Uuid,
recorded_at_micros: i64,
) -> Result<Self, KnowledgeError> {
source_record_uuids.sort_unstable();
source_record_uuids.dedup();
let policy_fingerprint = fingerprint(
CanonicalDomain::BeliefProjectionPolicy,
CANONICAL_CONTRACT_VERSION,
&policy_bytes,
)?;
let row = Self {
attachment_uuid,
run_uuid,
source_generation_uuid,
transaction_cutoff_micros,
valid_time_micros,
policy_version,
policy_bytes,
policy_fingerprint,
snapshot_fingerprint,
valid_time_fingerprint,
graph_content_fingerprint,
descriptor_fingerprint,
source_record_uuids,
provenance_uuid,
recorded_at_micros,
contract_version: BELIEF_PROJECTION_ATTACHMENT_CONTRACT_VERSION,
};
validate(&row)?;
Ok(row)
}
}
/// Validated append-only interpretation-attachment participant.
///
/// The validated collection is private so callers cannot invalidate it after
/// construction. Read through [`Self::attachments`] or consume with
/// [`Self::into_attachments`].
/// This is a source compatibility change for callers that directly used the
/// former public `attachments` field; construct ledgers with [`Self::new`].
///
/// ```compile_fail
/// use graphforge_knowledge::BeliefProjectionAttachmentLedger;
/// let mut ledger = BeliefProjectionAttachmentLedger::default();
/// ledger.attachments.clear();
/// ```
///
/// ```compile_fail
/// use graphforge_knowledge::BeliefProjectionAttachmentLedger;
/// let ledger = BeliefProjectionAttachmentLedger::default();
/// ledger.attachments()[0].policy_version = 0;
/// ```
#[derive(Clone, Debug, Default, Eq, PartialEq)]
pub struct BeliefProjectionAttachmentLedger {
/// Attachments ordered by `(recorded_at, attachment_uuid)`.
attachments: Vec<BeliefProjectionAttachment>,
}
impl BeliefProjectionAttachmentLedger {
/// Borrow the validated attachments in deterministic order.
#[must_use]
pub fn attachments(&self) -> &[BeliefProjectionAttachment] {
&self.attachments
}
/// Consume the ledger and return its validated attachments.
#[must_use]
pub fn into_attachments(self) -> Vec<BeliefProjectionAttachment> {
self.attachments
}
/// Validate, sort, and construct a complete participant.
pub fn new(attachments: Vec<BeliefProjectionAttachment>) -> Result<Self, KnowledgeError> {
Self::new_with_limit(attachments, MAX_KNOWLEDGE_ROWS)
}
fn new_with_limit(
mut attachments: Vec<BeliefProjectionAttachment>,
row_limit: usize,
) -> Result<Self, KnowledgeError> {
validate_row_count(attachments.len(), row_limit)?;
let mut ids = HashSet::with_capacity(attachments.len());
for row in &attachments {
validate(row)?;
if !ids.insert(row.attachment_uuid) {
return Err(KnowledgeError::Duplicate("attachment_uuid"));
}
}
attachments.sort_by_key(|row| (row.recorded_at_micros, row.attachment_uuid));
Ok(Self { attachments })
}
/// Merge with exact replay and transaction-conflict semantics.
pub fn merge(&self, staged: &Self) -> Result<Self, KnowledgeError> {
self.merge_with_limit(staged, MAX_KNOWLEDGE_ROWS)
}
fn merge_with_limit(&self, staged: &Self, row_limit: usize) -> Result<Self, KnowledgeError> {
// Both operands crossed `new()` or `from_batches()`, and callers cannot
// mutate the private row vector. Revalidate only at those boundaries.
let mut rows = Vec::with_capacity(self.attachments.len() + staged.attachments.len());
rows.extend(self.attachments.iter().cloned());
let mut by_id = HashMap::with_capacity(self.attachments.len() + staged.attachments.len());
by_id.extend(
self.attachments
.iter()
.map(|row| (row.attachment_uuid, row)),
);
for row in &staged.attachments {
if let Some(existing) = by_id.get(&row.attachment_uuid) {
if *existing != row {
return Err(KnowledgeError::TransactionConflict("attachment_uuid"));
}
} else {
rows.push(row.clone());
by_id.insert(row.attachment_uuid, row);
}
}
// Preserve the previous error precedence: scan every staged identity
// for conflicts before reporting a combined row-limit violation.
validate_row_count(rows.len(), row_limit)?;
rows.sort_by_key(|row| (row.recorded_at_micros, row.attachment_uuid));
Ok(Self { attachments: rows })
}
/// Canonical fingerprint over one exact immutable attachment.
pub fn attachment_fingerprint(&self, id: Uuid) -> Result<[u8; 32], KnowledgeError> {
let row = self
.attachments
.iter()
.find(|row| row.attachment_uuid == id)
.ok_or(KnowledgeError::Dangling("attachment_uuid"))?;
let mut writer = CanonicalWriter::new();
writer.raw(row.attachment_uuid.as_bytes())?;
writer.raw(row.run_uuid.as_bytes())?;
writer.raw(row.source_generation_uuid.as_bytes())?;
writer.i64(row.transaction_cutoff_micros)?;
optional_i64(&mut writer, row.valid_time_micros)?;
writer.u32(row.policy_version)?;
writer.binary(&row.policy_bytes)?;
writer.raw(&row.policy_fingerprint)?;
writer.raw(&row.snapshot_fingerprint)?;
optional_fingerprint(&mut writer, row.valid_time_fingerprint)?;
writer.raw(&row.graph_content_fingerprint)?;
writer.raw(&row.descriptor_fingerprint)?;
writer.u64(u64::try_from(row.source_record_uuids.len()).unwrap_or(u64::MAX))?;
for source in &row.source_record_uuids {
writer.raw(source.as_bytes())?;
}
writer.raw(row.provenance_uuid.as_bytes())?;
writer.i64(row.recorded_at_micros)?;
writer.u32(row.contract_version)?;
fingerprint(
CanonicalDomain::BeliefProjectionAttachment,
CANONICAL_CONTRACT_VERSION,
&writer.finish(),
)
.map_err(Into::into)
}
/// Build the authoritative Arrow batch.
pub fn batch(&self) -> Result<RecordBatch, KnowledgeError> {
let len = self.attachments.len();
let mut attachment = FixedSizeBinaryBuilder::with_capacity(len, 16);
let mut run = FixedSizeBinaryBuilder::with_capacity(len, 16);
let mut generation = FixedSizeBinaryBuilder::with_capacity(len, 16);
let mut cutoff = TimestampMicrosecondBuilder::new().with_timezone("UTC");
let mut valid_time = TimestampMicrosecondBuilder::new().with_timezone("UTC");
let mut policy_version = UInt32Builder::new();
let mut policy_bytes = BinaryBuilder::new();
let mut policy_fp = FixedSizeBinaryBuilder::with_capacity(len, 32);
let mut snapshot_fp = FixedSizeBinaryBuilder::with_capacity(len, 32);
let mut valid_fp = FixedSizeBinaryBuilder::with_capacity(len, 32);
let mut graph_fp = FixedSizeBinaryBuilder::with_capacity(len, 32);
let mut descriptor_fp = FixedSizeBinaryBuilder::with_capacity(len, 32);
let mut sources = ListBuilder::new(FixedSizeBinaryBuilder::new(16)).with_field(Arc::new(
Field::new("item", DataType::FixedSizeBinary(16), false),
));
let mut provenance = FixedSizeBinaryBuilder::with_capacity(len, 16);
let mut recorded = TimestampMicrosecondBuilder::new().with_timezone("UTC");
let mut version = UInt32Builder::new();
for row in &self.attachments {
append(&mut attachment, row.attachment_uuid.as_bytes())?;
append(&mut run, row.run_uuid.as_bytes())?;
append(&mut generation, row.source_generation_uuid.as_bytes())?;
cutoff.append_value(row.transaction_cutoff_micros);
valid_time.append_option(row.valid_time_micros);
policy_version.append_value(row.policy_version);
policy_bytes.append_value(&row.policy_bytes);
append(&mut policy_fp, &row.policy_fingerprint)?;
append(&mut snapshot_fp, &row.snapshot_fingerprint)?;
append_optional(&mut valid_fp, row.valid_time_fingerprint.as_ref())?;
append(&mut graph_fp, &row.graph_content_fingerprint)?;
append(&mut descriptor_fp, &row.descriptor_fingerprint)?;
for source in &row.source_record_uuids {
append(sources.values(), source.as_bytes())?;
}
sources.append(true);
append(&mut provenance, row.provenance_uuid.as_bytes())?;
recorded.append_value(row.recorded_at_micros);
version.append_value(row.contract_version);
}
Ok(RecordBatch::try_new(
Arc::clone(&ALGORITHM_INTERPRETATION_ATTACHMENT_SCHEMA),
vec![
Arc::new(attachment.finish()),
Arc::new(run.finish()),
Arc::new(generation.finish()),
Arc::new(cutoff.finish()),
Arc::new(valid_time.finish()),
Arc::new(policy_version.finish()),
Arc::new(policy_bytes.finish()),
Arc::new(policy_fp.finish()),
Arc::new(snapshot_fp.finish()),
Arc::new(valid_fp.finish()),
Arc::new(graph_fp.finish()),
Arc::new(descriptor_fp.finish()),
Arc::new(sources.finish()),
Arc::new(provenance.finish()),
Arc::new(recorded.finish()),
Arc::new(version.finish()),
],
)?)
}
/// Decode exact Arrow batches and re-run every invariant.
pub fn from_batches(batches: &[RecordBatch]) -> Result<Self, KnowledgeError> {
let mut rows = Vec::new();
for batch in batches {
if batch.schema().as_ref() != ALGORITHM_INTERPRETATION_ATTACHMENT_SCHEMA.as_ref() {
return Err(invalid(
"belief_projection_attachment.schema",
"schema mismatch",
));
}
let attachment = fixed(batch, "attachment_uuid")?;
let run = fixed(batch, "run_uuid")?;
let generation = fixed(batch, "source_generation_uuid")?;
let cutoff = timestamp(batch, "transaction_cutoff")?;
let valid_time = timestamp(batch, "valid_time")?;
let policy_version = uint32(batch, "policy_version")?;
let policy_bytes = binary(batch, "policy_bytes")?;
let policy_fp = fixed(batch, "policy_fingerprint")?;
let snapshot_fp = fixed(batch, "snapshot_fingerprint")?;
let valid_fp = fixed(batch, "valid_time_fingerprint")?;
let graph_fp = fixed(batch, "graph_content_fingerprint")?;
let descriptor_fp = fixed(batch, "descriptor_fingerprint")?;
let sources = list(batch, "source_record_uuids")?;
let provenance = fixed(batch, "provenance_uuid")?;
let recorded = timestamp(batch, "recorded_at")?;
let versions = uint32(batch, "contract_version")?;
for row in 0..batch.num_rows() {
rows.push(BeliefProjectionAttachment {
attachment_uuid: uuid_at(attachment, row, "attachment_uuid")?,
run_uuid: uuid_at(run, row, "run_uuid")?,
source_generation_uuid: uuid_at(generation, row, "source_generation_uuid")?,
transaction_cutoff_micros: cutoff.value(row),
valid_time_micros: optional_timestamp(valid_time, row),
policy_version: policy_version.value(row),
policy_bytes: policy_bytes.value(row).to_vec(),
policy_fingerprint: bytes32(policy_fp, row, "policy_fingerprint")?,
snapshot_fingerprint: bytes32(snapshot_fp, row, "snapshot_fingerprint")?,
valid_time_fingerprint: optional_bytes32(
valid_fp,
row,
"valid_time_fingerprint",
)?,
graph_content_fingerprint: bytes32(graph_fp, row, "graph_content_fingerprint")?,
descriptor_fingerprint: bytes32(descriptor_fp, row, "descriptor_fingerprint")?,
source_record_uuids: uuid_list_at(sources, row)?,
provenance_uuid: uuid_at(provenance, row, "provenance_uuid")?,
recorded_at_micros: recorded.value(row),
contract_version: versions.value(row),
});
}
}
Self::new(rows)
}
}
pub(crate) fn schema_registry_entry() -> SchemaRegistryEntry {
SchemaRegistryEntry {
capability_id: "epistemic",
capability_version: EPISTEMIC_CAPABILITY_VERSION,
record_family: "algorithm_interpretation_attachments",
record_version: BELIEF_PROJECTION_ATTACHMENT_CONTRACT_VERSION,
schema: Arc::clone(&ALGORITHM_INTERPRETATION_ATTACHMENT_SCHEMA),
schema_fingerprint: *SCHEMA_FINGERPRINT,
enum_registry_versions: &[],
sort_key: &["recorded_at", "attachment_uuid"],
diff_identity_fields: &["attachment_uuid"],
diff_record_uuid_field: Some("attachment_uuid"),
fingerprint_domain: CanonicalDomain::BeliefProjectionAttachment,
owner: "graphforge-knowledge",
implementation_issue: 2004,
max_rows: MAX_KNOWLEDGE_ROWS,
}
}
fn validate(row: &BeliefProjectionAttachment) -> Result<(), KnowledgeError> {
#[cfg(test)]
VALIDATION_CALLS.with(|calls| calls.set(calls.get() + 1));
if row.contract_version != BELIEF_PROJECTION_ATTACHMENT_CONTRACT_VERSION {
return Err(invalid(
"belief_projection_attachment.contract_version",
"unsupported version",
));
}
require_v7(row.attachment_uuid, "attachment_uuid")?;
require_v7(row.run_uuid, "run_uuid")?;
require_uuid(row.source_generation_uuid, "source_generation_uuid")?;
require_uuid(row.provenance_uuid, "provenance_uuid")?;
if row.policy_version == 0 {
return Err(invalid(
"belief_projection_attachment.policy_version",
"must be positive",
));
}
validate_policy_len(row.policy_bytes.len())?;
let expected = fingerprint(
CanonicalDomain::BeliefProjectionPolicy,
CANONICAL_CONTRACT_VERSION,
&row.policy_bytes,
)?;
if row.policy_fingerprint != expected {
return Err(invalid(
"belief_projection_attachment.policy_fingerprint",
"does not match policy bytes",
));
}
if row
.source_record_uuids
.windows(2)
.any(|pair| pair[0] >= pair[1])
{
return Err(invalid(
"belief_projection_attachment.source_record_uuids",
"must be sorted and deduplicated",
));
}
for source in &row.source_record_uuids {
require_uuid(*source, "source_record_uuid")?;
}
Ok(())
}
#[cfg(test)]
thread_local! {
static VALIDATION_CALLS: Cell<usize> = const { Cell::new(0) };
}
#[cfg(test)]
fn reset_validation_calls() {
VALIDATION_CALLS.with(|calls| calls.set(0));
}
#[cfg(test)]
fn validation_call_count() -> usize {
VALIDATION_CALLS.with(Cell::get)
}
fn validate_row_count(count: usize, limit: usize) -> Result<(), KnowledgeError> {
if count > limit {
Err(KnowledgeError::Limit {
participant: "algorithm_interpretation_attachments",
observed: count,
limit,
})
} else {
Ok(())
}
}
fn validate_policy_len(length: usize) -> Result<(), KnowledgeError> {
if u64::try_from(length).unwrap_or(u64::MAX) > MAX_CANONICAL_BINARY_BYTES {
Err(KnowledgeError::Limit {
participant: "policy_bytes",
observed: length,
limit: usize::try_from(MAX_CANONICAL_BINARY_BYTES).unwrap_or(usize::MAX),
})
} else {
Ok(())
}
}
fn optional_i64(writer: &mut CanonicalWriter, value: Option<i64>) -> Result<(), KnowledgeError> {
match value {
Some(value) => {
writer.u8(1)?;
writer.i64(value)?;
}
None => writer.u8(0)?,
}
Ok(())
}
fn optional_fingerprint(
writer: &mut CanonicalWriter,
value: Option<[u8; 32]>,
) -> Result<(), KnowledgeError> {
match value {
Some(value) => {
writer.u8(1)?;
writer.raw(&value)?;
}
None => writer.u8(0)?,
}
Ok(())
}
fn append(builder: &mut FixedSizeBinaryBuilder, value: &[u8]) -> Result<(), KnowledgeError> {
builder
.append_value(value)
.map_err(|_| invalid("belief_projection_attachment", "invalid fixed-width value"))
}
fn append_optional(
builder: &mut FixedSizeBinaryBuilder,
value: Option<&[u8; 32]>,
) -> Result<(), KnowledgeError> {
if let Some(value) = value {
append(builder, value)
} else {
builder.append_null();
Ok(())
}
}
fn uuid_field(name: &str, nullable: bool) -> Field {
Field::new(name, DataType::FixedSizeBinary(16), nullable)
}
fn fingerprint_field(name: &str, nullable: bool) -> Field {
Field::new(name, DataType::FixedSizeBinary(32), nullable)
}
fn timestamp_field(name: &str, nullable: bool) -> Field {
Field::new(
name,
DataType::Timestamp(TimeUnit::Microsecond, Some("UTC".into())),
nullable,
)
}
fn fixed<'a>(
batch: &'a RecordBatch,
name: &'static str,
) -> Result<&'a FixedSizeBinaryArray, KnowledgeError> {
batch
.column_by_name(name)
.and_then(|v| v.as_any().downcast_ref())
.ok_or_else(|| invalid(name, "column type mismatch"))
}
fn timestamp<'a>(
batch: &'a RecordBatch,
name: &'static str,
) -> Result<&'a TimestampMicrosecondArray, KnowledgeError> {
batch
.column_by_name(name)
.and_then(|v| v.as_any().downcast_ref())
.ok_or_else(|| invalid(name, "column type mismatch"))
}
fn uint32<'a>(
batch: &'a RecordBatch,
name: &'static str,
) -> Result<&'a UInt32Array, KnowledgeError> {
batch
.column_by_name(name)
.and_then(|v| v.as_any().downcast_ref())
.ok_or_else(|| invalid(name, "column type mismatch"))
}
fn binary<'a>(
batch: &'a RecordBatch,
name: &'static str,
) -> Result<&'a BinaryArray, KnowledgeError> {
batch
.column_by_name(name)
.and_then(|v| v.as_any().downcast_ref())
.ok_or_else(|| invalid(name, "column type mismatch"))
}
fn list<'a>(batch: &'a RecordBatch, name: &'static str) -> Result<&'a ListArray, KnowledgeError> {
batch
.column_by_name(name)
.and_then(|v| v.as_any().downcast_ref())
.ok_or_else(|| invalid(name, "column type mismatch"))
}
fn uuid_at(
values: &FixedSizeBinaryArray,
row: usize,
field: &'static str,
) -> Result<Uuid, KnowledgeError> {
Uuid::from_slice(values.value(row)).map_err(|_| invalid(field, "invalid UUID bytes"))
}
fn bytes32(
values: &FixedSizeBinaryArray,
row: usize,
field: &'static str,
) -> Result<[u8; 32], KnowledgeError> {
values
.value(row)
.try_into()
.map_err(|_| invalid(field, "invalid fingerprint width"))
}
fn optional_bytes32(
values: &FixedSizeBinaryArray,
row: usize,
field: &'static str,
) -> Result<Option<[u8; 32]>, KnowledgeError> {
(!values.is_null(row))
.then(|| bytes32(values, row, field))
.transpose()
}
fn optional_timestamp(values: &TimestampMicrosecondArray, row: usize) -> Option<i64> {
(!values.is_null(row)).then(|| values.value(row))
}
fn uuid_list_at(values: &ListArray, row: usize) -> Result<Vec<Uuid>, KnowledgeError> {
let value = values.value(row);
let fixed = value
.as_any()
.downcast_ref::<FixedSizeBinaryArray>()
.ok_or_else(|| invalid("source_record_uuids", "item type mismatch"))?;
(0..fixed.len())
.map(|index| uuid_at(fixed, index, "source_record_uuid"))
.collect()
}
fn require_v7(value: Uuid, field: &'static str) -> Result<(), KnowledgeError> {
if value.get_version() != Some(Version::SortRand) {
return Err(invalid(field, "must be UUIDv7"));
}
require_uuid(value, field)
}
fn require_uuid(value: Uuid, field: &'static str) -> Result<(), KnowledgeError> {
if value.is_nil() {
Err(invalid(field, "must not be nil"))
} else {
Ok(())
}
}
const fn invalid(field: &'static str, message: &'static str) -> KnowledgeError {
KnowledgeError::Invalid { field, message }
}
#[cfg(test)]
mod tests {
use super::*;
use std::time::{Duration, Instant};
fn uuid7(seed: u8) -> Uuid {
let mut bytes = [seed; 16];
bytes[6] = (bytes[6] & 0x0f) | 0x70;
bytes[8] = (bytes[8] & 0x3f) | 0x80;
Uuid::from_bytes(bytes)
}
fn attachment(id: u8, sources: Vec<Uuid>) -> BeliefProjectionAttachment {
BeliefProjectionAttachment::new(
uuid7(id),
uuid7(id + 1),
uuid7(id + 2),
10,
Some(11),
1,
b"policy-v1".to_vec(),
[3; 32],
Some([4; 32]),
[5; 32],
[6; 32],
sources,
uuid7(id + 3),
20,
)
.unwrap()
}
fn attachment_at(id: u8, recorded_at_micros: i64) -> BeliefProjectionAttachment {
let mut row = attachment(id, vec![]);
row.recorded_at_micros = recorded_at_micros;
row
}
fn non_v7_uuid(seed: u8) -> Uuid {
let mut bytes = [seed; 16];
bytes[6] = (bytes[6] & 0x0f) | 0x40;
bytes[8] = (bytes[8] & 0x3f) | 0x80;
Uuid::from_bytes(bytes)
}
fn benchmark_uuid(seed: u64) -> Uuid {
let mut bytes = [0; 16];
bytes[8..].copy_from_slice(&seed.to_be_bytes());
bytes[6] = 0x70;
bytes[8] = (bytes[8] & 0x3f) | 0x80;
Uuid::from_bytes(bytes)
}
fn benchmark_attachment(seed: u64) -> BeliefProjectionAttachment {
let policy_bytes = b"benchmark-policy".to_vec();
BeliefProjectionAttachment {
attachment_uuid: benchmark_uuid(seed + 1),
run_uuid: benchmark_uuid(seed + 2),
source_generation_uuid: benchmark_uuid(seed + 3),
transaction_cutoff_micros: 10,
valid_time_micros: Some(11),
policy_version: 1,
policy_fingerprint: fingerprint(
CanonicalDomain::BeliefProjectionPolicy,
CANONICAL_CONTRACT_VERSION,
&policy_bytes,
)
.unwrap(),
policy_bytes,
snapshot_fingerprint: [3; 32],
valid_time_fingerprint: Some([4; 32]),
graph_content_fingerprint: [5; 32],
descriptor_fingerprint: [6; 32],
source_record_uuids: vec![benchmark_uuid(seed + 4)],
provenance_uuid: benchmark_uuid(seed + 5),
recorded_at_micros: i64::try_from(seed).unwrap(),
contract_version: BELIEF_PROJECTION_ATTACHMENT_CONTRACT_VERSION,
}
}
fn merge_with_full_revalidation(
existing: &BeliefProjectionAttachmentLedger,
staged: &BeliefProjectionAttachmentLedger,
) -> Result<BeliefProjectionAttachmentLedger, KnowledgeError> {
let mut rows = existing.attachments.clone();
let mut by_id = rows
.iter()
.cloned()
.map(|row| (row.attachment_uuid, row))
.collect::<HashMap<_, _>>();
for row in &staged.attachments {
if let Some(previous) = by_id.get(&row.attachment_uuid) {
if previous != row {
return Err(KnowledgeError::TransactionConflict("attachment_uuid"));
}
} else {
rows.push(row.clone());
by_id.insert(row.attachment_uuid, row.clone());
}
}
BeliefProjectionAttachmentLedger::new(rows)
}
fn median(mut values: Vec<Duration>) -> Duration {
values.sort_unstable();
values[values.len() / 2]
}
fn input_digest(
existing: &BeliefProjectionAttachmentLedger,
staged: &BeliefProjectionAttachmentLedger,
) -> String {
let mut writer = CanonicalWriter::new();
for ledger in [existing, staged] {
writer.u64(ledger.attachments.len() as u64).unwrap();
for row in &ledger.attachments {
writer.raw(row.attachment_uuid.as_bytes()).unwrap();
writer.raw(row.run_uuid.as_bytes()).unwrap();
writer.raw(row.source_generation_uuid.as_bytes()).unwrap();
writer.i64(row.transaction_cutoff_micros).unwrap();
writer
.u8(u8::from(row.valid_time_micros.is_some()))
.unwrap();
if let Some(value) = row.valid_time_micros {
writer.i64(value).unwrap();
}
writer.u32(row.policy_version).unwrap();
writer.binary(&row.policy_bytes).unwrap();
writer.raw(&row.policy_fingerprint).unwrap();
writer.raw(&row.snapshot_fingerprint).unwrap();
writer
.u8(u8::from(row.valid_time_fingerprint.is_some()))
.unwrap();
if let Some(value) = row.valid_time_fingerprint {
writer.raw(&value).unwrap();
}
writer.raw(&row.graph_content_fingerprint).unwrap();
writer.raw(&row.descriptor_fingerprint).unwrap();
writer.u64(row.source_record_uuids.len() as u64).unwrap();
for source in &row.source_record_uuids {
writer.raw(source.as_bytes()).unwrap();
}
writer.raw(row.provenance_uuid.as_bytes()).unwrap();
writer.i64(row.recorded_at_micros).unwrap();
writer.u32(row.contract_version).unwrap();
}
}
fingerprint(
CanonicalDomain::BeliefProjectionAttachment,
CANONICAL_CONTRACT_VERSION,
&writer.finish(),
)
.unwrap()
.iter()
.map(|byte| format!("{byte:02x}"))
.collect()
}
#[test]
fn sources_are_canonical_and_arrow_round_trip_is_exact() {
let row = attachment(1, vec![uuid7(9), uuid7(8), uuid7(9)]);
assert_eq!(row.source_record_uuids, vec![uuid7(8), uuid7(9)]);
let ledger = BeliefProjectionAttachmentLedger::new(vec![row]).unwrap();
let decoded =
BeliefProjectionAttachmentLedger::from_batches(&[ledger.batch().unwrap()]).unwrap();
assert_eq!(decoded, ledger);
assert_eq!(
decoded.attachment_fingerprint(uuid7(1)).unwrap(),
ledger.attachment_fingerprint(uuid7(1)).unwrap()
);
}
#[test]
fn replay_is_idempotent_and_conflict_has_transaction_code() {
let row = attachment(10, vec![]);
let ledger = BeliefProjectionAttachmentLedger::new(vec![row.clone()]).unwrap();
assert_eq!(
ledger
.merge(&BeliefProjectionAttachmentLedger::new(vec![row.clone()]).unwrap())
.unwrap(),
ledger
);
let mut different = row;
different.graph_content_fingerprint = [99; 32];
let error = ledger
.merge(&BeliefProjectionAttachmentLedger::new(vec![different]).unwrap())
.unwrap_err();
assert_eq!(error.code(), "GF_TRANSACTION_CONFLICT");
}
#[test]
fn registry_is_epistemic_owned_and_frozen() {
let entry = schema_registry_entry();
assert_eq!(entry.capability_id, "epistemic");
assert_eq!(entry.record_family, "algorithm_interpretation_attachments");
assert_eq!(entry.sort_key, &["recorded_at", "attachment_uuid"]);
assert_eq!(entry.implementation_issue, 2004);
}
#[test]
fn defensive_attachment_validation_and_optional_fingerprints_are_exact() {
let row = attachment(30, vec![uuid7(40)]);
assert!(matches!(
BeliefProjectionAttachmentLedger::new(vec![row.clone(), row.clone()]),
Err(KnowledgeError::Duplicate("attachment_uuid"))
));
let mut invalid = row.clone();
invalid.contract_version += 1;
assert!(
validate(&invalid)
.unwrap_err()
.to_string()
.contains("unsupported version")
);
invalid = row.clone();
invalid.policy_version = 0;
assert!(
validate(&invalid)
.unwrap_err()
.to_string()
.contains("must be positive")
);
invalid = row.clone();
invalid.policy_fingerprint = [0; 32];
assert!(
validate(&invalid)
.unwrap_err()
.to_string()
.contains("does not match policy bytes")
);
invalid = row.clone();
invalid.source_record_uuids = vec![uuid7(42), uuid7(41)];
assert!(
validate(&invalid)
.unwrap_err()
.to_string()
.contains("sorted and deduplicated")
);
invalid = row.clone();
invalid.source_record_uuids = vec![uuid7(41), uuid7(41)];
assert!(
validate(&invalid)
.unwrap_err()
.to_string()
.contains("sorted and deduplicated")
);
invalid = row;
invalid.source_record_uuids = vec![Uuid::nil()];
assert!(
validate(&invalid)
.unwrap_err()
.to_string()
.contains("must not be nil")
);
let without_optional = BeliefProjectionAttachment::new(
uuid7(50),
uuid7(51),
uuid7(52),
10,
None,
1,
b"policy-v1".to_vec(),
[3; 32],
None,
[5; 32],
[6; 32],
vec![],
uuid7(53),
20,
)
.unwrap();
BeliefProjectionAttachmentLedger::new(vec![without_optional])
.unwrap()
.attachment_fingerprint(uuid7(50))
.unwrap();
let wrong = RecordBatch::new_empty(Arc::new(Schema::empty()));
assert!(
BeliefProjectionAttachmentLedger::from_batches(&[wrong])
.unwrap_err()
.to_string()
.contains("schema mismatch")
);
}
#[test]
fn attachment_and_run_uuids_must_be_v7() {
let row = attachment(31, vec![]);
let mut invalid = row.clone();
invalid.attachment_uuid = non_v7_uuid(1);
assert!(
validate(&invalid)
.unwrap_err()
.to_string()
.contains("must be UUIDv7")
);
invalid = row;
invalid.run_uuid = non_v7_uuid(2);
assert!(
validate(&invalid)
.unwrap_err()
.to_string()
.contains("must be UUIDv7")
);
}
#[test]
fn attachment_contract_version_must_be_supported() {
let mut invalid = attachment(40, vec![]);
invalid.contract_version = BELIEF_PROJECTION_ATTACHMENT_CONTRACT_VERSION + 1;
assert!(
validate(&invalid)
.unwrap_err()
.to_string()
.contains("unsupported version")
);
}
#[test]
fn duplicate_attachment_uuid_is_rejected() {
let first = attachment(41, vec![]);
let mut duplicate = first.clone();
duplicate.recorded_at_micros += 1;
assert!(matches!(
BeliefProjectionAttachmentLedger::new(vec![first, duplicate]),
Err(KnowledgeError::Duplicate("attachment_uuid"))
));
}
#[test]
fn source_generation_and_provenance_uuids_must_be_non_nil() {
let row = attachment(32, vec![]);
let mut invalid = row.clone();
invalid.source_generation_uuid = Uuid::nil();
assert!(
validate(&invalid)
.unwrap_err()
.to_string()
.contains("must not be nil")
);
invalid = row;
invalid.provenance_uuid = Uuid::nil();
assert!(
validate(&invalid)
.unwrap_err()
.to_string()
.contains("must not be nil")
);
}
#[test]
fn policy_version_must_be_positive() {
let mut invalid = attachment(33, vec![]);
invalid.policy_version = 0;
assert!(
validate(&invalid)
.unwrap_err()
.to_string()
.contains("must be positive")
);
}
#[test]
fn policy_byte_limit_rejects_the_first_excess_byte() {
let maximum = usize::try_from(MAX_CANONICAL_BINARY_BYTES).unwrap();
assert!(validate_policy_len(maximum).is_ok());
assert!(matches!(
validate_policy_len(maximum + 1),
Err(KnowledgeError::Limit {
participant: "policy_bytes",
observed,
limit,
}) if observed == maximum + 1 && limit == maximum
));
let mut invalid = attachment(39, vec![]);
invalid.policy_bytes = vec![0; maximum + 1];
assert!(
validate(&invalid)
.unwrap_err()
.to_string()
.contains("policy_bytes")
);
}
#[test]
fn policy_fingerprint_must_match_policy_bytes() {
let mut invalid = attachment(34, vec![]);
invalid.policy_fingerprint = [0; 32];
assert!(
validate(&invalid)
.unwrap_err()
.to_string()
.contains("does not match policy bytes")
);
}
#[test]
fn source_record_uuids_must_be_sorted_unique_and_non_nil() {
let row = attachment(35, vec![uuid7(40), uuid7(41)]);
let mut invalid = row.clone();
invalid.source_record_uuids = vec![uuid7(41), uuid7(40)];
assert!(
validate(&invalid)
.unwrap_err()
.to_string()
.contains("sorted and deduplicated")
);
invalid = row.clone();
invalid.source_record_uuids = vec![uuid7(40), uuid7(40)];
assert!(
validate(&invalid)
.unwrap_err()
.to_string()
.contains("sorted and deduplicated")
);
invalid = row;
invalid.source_record_uuids = vec![Uuid::nil()];
assert!(
validate(&invalid)
.unwrap_err()
.to_string()
.contains("must not be nil")
);
}
#[test]
fn row_limit_constructor_and_merge_reject_first_excess() {
let rows = vec![attachment(36, vec![]), attachment(37, vec![])];
assert!(BeliefProjectionAttachmentLedger::new_with_limit(rows.clone(), 2).is_ok());
assert!(matches!(
BeliefProjectionAttachmentLedger::new_with_limit(rows.clone(), 1),
Err(KnowledgeError::Limit {
observed: 2,
limit: 1,
..
})
));
let existing = BeliefProjectionAttachmentLedger::new_with_limit(rows, 2).unwrap();
let staged = BeliefProjectionAttachmentLedger::new(vec![attachment(38, vec![])]).unwrap();
assert!(matches!(
existing.merge_with_limit(&staged, 2),
Err(KnowledgeError::Limit {
observed: 3,
limit: 2,
..
})
));
let replay =
BeliefProjectionAttachmentLedger::new(vec![existing.attachments()[0].clone()]).unwrap();
assert_eq!(existing.merge_with_limit(&replay, 2).unwrap(), existing);
let novel = attachment_at(39, 10);
let mut conflicting = attachment_at(36, 30);
conflicting.graph_content_fingerprint = [99; 32];
let staged = BeliefProjectionAttachmentLedger::new(vec![novel, conflicting]).unwrap();
assert!(matches!(
existing.merge_with_limit(&staged, 2),
Err(KnowledgeError::TransactionConflict("attachment_uuid"))
));
}
#[test]
fn incremental_merge_checks_staged_rows_once_and_never_existing_rows() {
let first = attachment(60, vec![]);
let replay = first.clone();
let later = attachment_at(70, 30);
let earlier = attachment_at(80, 15);
reset_validation_calls();
let existing = BeliefProjectionAttachmentLedger::new(vec![first]).unwrap();
assert_eq!(validation_call_count(), 1);
reset_validation_calls();
let staged = BeliefProjectionAttachmentLedger::new(vec![later, replay, earlier]).unwrap();
assert_eq!(validation_call_count(), 3);
reset_validation_calls();
let merged = existing.merge(&staged).unwrap();
assert_eq!(validation_call_count(), 0);
assert_eq!(
merged
.attachments()
.iter()
.map(|r| r.attachment_uuid)
.collect::<Vec<_>>(),
vec![uuid7(80), uuid7(60), uuid7(70)]
);
assert_eq!(merged.clone().into_attachments().len(), 3);
}
#[test]
fn constructor_validates_each_new_row_once() {
let rows = vec![attachment(100, vec![]), attachment(101, vec![])];
reset_validation_calls();
let ledger = BeliefProjectionAttachmentLedger::new(rows).unwrap();
assert_eq!(ledger.attachments().len(), 2);
assert_eq!(validation_call_count(), 2);
}
#[test]
fn constructor_sorts_rows_by_recorded_time_then_uuid() {
let late = attachment_at(110, 30);
let same_time_lower_uuid = attachment_at(100, 30);
let same_time_higher_uuid = attachment_at(120, 30);
let ledger = BeliefProjectionAttachmentLedger::new(vec![
same_time_higher_uuid,
late,
same_time_lower_uuid,
])
.unwrap();
assert_eq!(
ledger
.attachments()
.iter()
.map(|row| row.attachment_uuid)
.collect::<Vec<_>>(),
vec![uuid7(100), uuid7(110), uuid7(120)]
);
}
#[test]
#[ignore = "manual quiet-host before/after merge-cost measurement"]
fn quiet_host_merge_cost_measurement() {
for existing_count in [1_000_u64, 10_000] {
let existing = BeliefProjectionAttachmentLedger::new(
(0..existing_count).map(benchmark_attachment).collect(),
)
.unwrap();
let mut staged_row = benchmark_attachment(existing_count + 10);
staged_row.recorded_at_micros = i64::try_from(existing_count / 2).unwrap();
let staged = BeliefProjectionAttachmentLedger::new(vec![staged_row]).unwrap();
let baseline = merge_with_full_revalidation(&existing, &staged).unwrap();
let incremental = existing.merge(&staged).unwrap();
assert_eq!(baseline, incremental);
let repetitions = 9;
let mut before = Vec::with_capacity(repetitions);
let mut after = Vec::with_capacity(repetitions);
for _ in 0..repetitions {
let start = Instant::now();
std::hint::black_box(merge_with_full_revalidation(&existing, &staged).unwrap());
before.push(start.elapsed());
let start = Instant::now();
std::hint::black_box(existing.merge(&staged).unwrap());
after.push(start.elapsed());
}
let before_median = median(before);
let after_median = median(after);
let divisor = existing_count as f64;
println!(
"existing_rows={existing_count} staged_rows=1 repetitions={repetitions} input_sha256={} before_ns={} before_ns_per_existing_row={:.3} after_ns={} after_ns_per_existing_row={:.3}",
input_digest(&existing, &staged),
before_median.as_nanos(),
before_median.as_nanos() as f64 / divisor,
after_median.as_nanos(),
after_median.as_nanos() as f64 / divisor,
);
}
}
}