use std::sync::Arc;
use arrow::array::{ArrayRef, BinaryBuilder, StringBuilder, UInt64Builder};
use arrow::datatypes::{DataType, Field, Schema, SchemaRef};
use arrow::error::ArrowError;
use arrow::record_batch::RecordBatch;
use polyc_eventlog::Event;
use polyc_proto::kinds;
#[cfg(test)]
const KNOWN_REASONS: [&str; 22] = [
"approval_mismatch",
"blocked_destination",
"client_build",
"host_not_allowed",
"persona_host_not_allowed",
"over_spend_cap",
"mandate_refused",
"mandate_host_unknown",
"over_budget",
"too_many_payments_in_flight",
"payment_already_in_flight",
"missing_currency",
"invalid_max_spend",
"backend_unavailable",
"fetch_failed",
"payment",
"unlinked",
"linked_but_unusable",
"temporarily_unavailable",
"payments_not_configured",
"no_approved_call",
"invalid_approved_args",
];
#[cfg(test)]
fn is_known_reason(reason: &str) -> bool {
KNOWN_REASONS.contains(&reason)
}
#[derive(Debug, Clone)]
pub(crate) struct RefusalRow {
pub partition: String,
pub position: u64,
pub turn_id: Option<String>,
pub reason: String,
pub reason_detail: String,
pub merchant_host: String,
pub requested_base_units: String,
pub permitted_base_units: String,
pub tool_call_id: String,
pub subject: String,
pub timestamp: String,
pub signer_public_key: Vec<u8>,
}
#[must_use]
pub(crate) fn schema() -> SchemaRef {
Arc::new(Schema::new(vec![
Field::new("partition", DataType::Utf8, false),
Field::new("position", DataType::UInt64, false),
Field::new("turn_id", DataType::Utf8, true),
Field::new("reason", DataType::Utf8, false),
Field::new("reason_detail", DataType::Utf8, false),
Field::new("merchant_host", DataType::Utf8, false),
Field::new("requested_base_units", DataType::Utf8, false),
Field::new("permitted_base_units", DataType::Utf8, false),
Field::new("tool_call_id", DataType::Utf8, false),
Field::new("subject", DataType::Utf8, false),
Field::new("timestamp", DataType::Utf8, false),
Field::new("signer_public_key", DataType::Binary, false),
]))
}
pub(crate) fn decode_refusals_batch(rows: &[RefusalRow]) -> Result<RecordBatch, ArrowError> {
let mut partition_b = StringBuilder::with_capacity(rows.len(), rows.len() * 8);
let mut position_b = UInt64Builder::with_capacity(rows.len());
let mut turn_id_b = StringBuilder::with_capacity(rows.len(), rows.len() * 36);
let mut reason_b = StringBuilder::with_capacity(rows.len(), rows.len() * 16);
let mut reason_detail_b = StringBuilder::with_capacity(rows.len(), rows.len() * 32);
let mut merchant_host_b = StringBuilder::with_capacity(rows.len(), rows.len() * 16);
let mut requested_b = StringBuilder::with_capacity(rows.len(), rows.len() * 8);
let mut permitted_b = StringBuilder::with_capacity(rows.len(), rows.len() * 8);
let mut tool_call_id_b = StringBuilder::with_capacity(rows.len(), rows.len() * 16);
let mut subject_b = StringBuilder::with_capacity(rows.len(), rows.len() * 16);
let mut timestamp_b = StringBuilder::with_capacity(rows.len(), rows.len() * 12);
let mut signer_public_key_b = BinaryBuilder::with_capacity(rows.len(), rows.len() * 32);
for row in rows {
partition_b.append_value(&row.partition);
position_b.append_value(row.position);
match &row.turn_id {
Some(id) => turn_id_b.append_value(id),
None => turn_id_b.append_null(),
}
reason_b.append_value(&row.reason);
reason_detail_b.append_value(&row.reason_detail);
merchant_host_b.append_value(&row.merchant_host);
requested_b.append_value(&row.requested_base_units);
permitted_b.append_value(&row.permitted_base_units);
tool_call_id_b.append_value(&row.tool_call_id);
subject_b.append_value(&row.subject);
timestamp_b.append_value(&row.timestamp);
signer_public_key_b.append_value(&row.signer_public_key);
}
let columns: Vec<ArrayRef> = vec![
Arc::new(partition_b.finish()),
Arc::new(position_b.finish()),
Arc::new(turn_id_b.finish()),
Arc::new(reason_b.finish()),
Arc::new(reason_detail_b.finish()),
Arc::new(merchant_host_b.finish()),
Arc::new(requested_b.finish()),
Arc::new(permitted_b.finish()),
Arc::new(tool_call_id_b.finish()),
Arc::new(subject_b.finish()),
Arc::new(timestamp_b.finish()),
Arc::new(signer_public_key_b.finish()),
];
RecordBatch::try_new(schema(), columns)
}
#[must_use]
pub(crate) fn decode_refusals_events(
partition: &str,
events: &[(u64, Event)],
trusted_signers: &[Vec<u8>],
) -> Vec<RefusalRow> {
events
.iter()
.filter_map(|(position, event)| {
let (base, turn_id) = kinds::parse(&event.kind);
if base != kinds::PAYMENT_REFUSAL {
return None;
}
let refusal =
polyc_facts::verified_refusals(std::slice::from_ref(event), trusted_signers)
.next()?;
Some(RefusalRow {
partition: partition.to_string(),
position: *position,
turn_id: turn_id.map(|id| id.to_string()),
reason: refusal.reason,
reason_detail: refusal.reason_detail,
merchant_host: refusal.merchant_host,
requested_base_units: refusal.requested_base_units,
permitted_base_units: refusal.permitted_base_units,
tool_call_id: refusal.tool_call_id,
subject: refusal.subject,
timestamp: refusal.timestamp,
signer_public_key: refusal.signer_public_key,
})
})
.collect()
}
#[cfg(test)]
mod tests {
use arrow::array::Array as _;
use polyc_crypto::approval::{ApprovalSigner, RefusalPayload, refusal_payload};
use uuid::Uuid;
use super::*;
fn signed_refusal(signer: &ApprovalSigner, reason: &str, tool_call_id: &str) -> Vec<u8> {
let (payload, sig, pk) = refusal_payload(
&RefusalPayload {
kind: kinds::PAYMENT_REFUSAL,
reason,
reason_detail: "detail text",
merchant_host: "merchant.example",
requested_base_units: "500",
permitted_base_units: "100",
tool_call_id,
subject: "persona-1",
timestamp: "1780000000",
},
signer,
);
let _ = (sig, pk);
payload
}
#[test]
fn schema_shape() {
let schema = schema();
let names: Vec<&str> = schema.fields().iter().map(|f| f.name().as_str()).collect();
assert_eq!(
names,
vec![
"partition",
"position",
"turn_id",
"reason",
"reason_detail",
"merchant_host",
"requested_base_units",
"permitted_base_units",
"tool_call_id",
"subject",
"timestamp",
"signer_public_key",
]
);
}
#[test]
fn decode_refusals_batch_round_trips() {
let rows = vec![RefusalRow {
partition: "conv-a".to_string(),
position: 5,
turn_id: Some("turn-xyz".to_string()),
reason: "over_spend_cap".to_string(),
reason_detail: "over cap".to_string(),
merchant_host: "merchant.example".to_string(),
requested_base_units: "500".to_string(),
permitted_base_units: "100".to_string(),
tool_call_id: "call-1".to_string(),
subject: "persona-1".to_string(),
timestamp: "1780000000".to_string(),
signer_public_key: vec![1, 2, 3, 4],
}];
let batch = decode_refusals_batch(&rows).expect("batch build");
assert_eq!(batch.num_rows(), 1);
assert_eq!(batch.schema(), schema());
let reason = batch
.column(3)
.as_any()
.downcast_ref::<arrow::array::StringArray>()
.unwrap();
assert_eq!(reason.value(0), "over_spend_cap");
}
#[test]
fn decode_refusals_events_verifies_and_decodes() {
let turn = Uuid::from_u128(0x0195_abcd_ef01_2345_6789_abcd_ef01_2345);
let signer = ApprovalSigner::from_seed(1);
let bytes = signed_refusal(&signer, "over_spend_cap", "call-1");
let events = vec![
(1, Event::new(kinds::TURN_START, Vec::new())),
(
2,
Event::new(kinds::tagged(kinds::PAYMENT_REFUSAL, &turn), bytes),
),
];
let trusted_signers = vec![signer.public_key_bytes()];
let decoded = decode_refusals_events("conv-real", &events, &trusted_signers);
assert_eq!(decoded.len(), 1);
assert_eq!(decoded[0].partition, "conv-real");
assert_eq!(decoded[0].position, 2);
assert_eq!(decoded[0].turn_id, Some(turn.to_string()));
assert_eq!(decoded[0].reason, "over_spend_cap");
assert_eq!(decoded[0].requested_base_units, "500");
assert_eq!(decoded[0].permitted_base_units, "100");
assert_eq!(decoded[0].tool_call_id, "call-1");
assert_eq!(decoded[0].subject, "persona-1");
assert_eq!(decoded[0].timestamp, "1780000000");
assert_eq!(decoded[0].signer_public_key, signer.public_key_bytes());
}
#[test]
fn decode_refusals_events_keeps_an_unknown_reason_tag() {
let signer = ApprovalSigner::from_seed(2);
let bytes = signed_refusal(
&signer,
"some_future_reason_this_build_does_not_know",
"call-2",
);
let events = vec![(1, Event::new(kinds::PAYMENT_REFUSAL, bytes))];
let trusted_signers = vec![signer.public_key_bytes()];
let decoded = decode_refusals_events("conv-unknown", &events, &trusted_signers);
assert_eq!(
decoded.len(),
1,
"an unknown reason tag must not be dropped"
);
assert_eq!(
decoded[0].reason,
"some_future_reason_this_build_does_not_know"
);
assert!(!is_known_reason(&decoded[0].reason));
}
#[test]
fn decode_refusals_events_drops_a_receipt_from_an_untrusted_signer() {
let turn = Uuid::from_u128(0x0195_abcd_ef01_2345_6789_abcd_ef01_1111);
let trusted = ApprovalSigner::from_seed(3);
let untrusted = ApprovalSigner::from_seed(4);
let forged = signed_refusal(&untrusted, "over_spend_cap", "call-forged");
let events = vec![(
1,
Event::new(kinds::tagged(kinds::PAYMENT_REFUSAL, &turn), forged),
)];
let trusted_signers = vec![trusted.public_key_bytes()];
let decoded = decode_refusals_events("conv-forged", &events, &trusted_signers);
assert_eq!(
decoded.len(),
0,
"a refusal signed by a key outside trusted_signers must never appear as a row"
);
}
#[test]
fn decode_refusals_events_drops_a_malformed_payload() {
let events = vec![(
1,
Event::new(kinds::PAYMENT_REFUSAL, vec![0xFF, 0xFE, 0xFD]),
)];
let decoded = decode_refusals_events("conv-corrupt", &events, &[]);
assert_eq!(decoded.len(), 0);
}
#[test]
fn decode_refusals_events_bare_kind_has_no_turn_id() {
let signer = ApprovalSigner::from_seed(5);
let bytes = signed_refusal(&signer, "fetch_failed", "call-bare");
let events = vec![(7, Event::new(kinds::PAYMENT_REFUSAL, bytes))];
let trusted_signers = vec![signer.public_key_bytes()];
let decoded = decode_refusals_events("conv-bare", &events, &trusted_signers);
assert_eq!(decoded.len(), 1);
assert_eq!(decoded[0].turn_id, None);
}
#[test]
fn unrelated_kind_is_not_decoded_as_a_refusal() {
let signer = ApprovalSigner::from_seed(6);
let bytes = signed_refusal(&signer, "over_spend_cap", "call-x");
let events = vec![(1, Event::new(kinds::USAGE, bytes))];
let trusted_signers = vec![signer.public_key_bytes()];
let decoded = decode_refusals_events("conv-unrelated", &events, &trusted_signers);
assert_eq!(
decoded.len(),
0,
"a usage-kind event must never decode as a refusal"
);
}
#[test]
fn known_reasons_cover_every_recording_seam_tag() {
for tag in [
"approval_mismatch",
"blocked_destination",
"client_build",
"host_not_allowed",
"persona_host_not_allowed",
"over_spend_cap",
"mandate_refused",
"mandate_host_unknown",
"over_budget",
"too_many_payments_in_flight",
"payment_already_in_flight",
"missing_currency",
"invalid_max_spend",
"backend_unavailable",
"fetch_failed",
"payment",
"unlinked",
"linked_but_unusable",
"temporarily_unavailable",
"payments_not_configured",
"no_approved_call",
"invalid_approved_args",
] {
assert!(is_known_reason(tag), "{tag} must be a known reason tag");
}
assert!(!is_known_reason("not_a_real_tag"));
}
}