//! `payments` typed-table decoder — the fact model's first FOLD-COUPLED
//! table: rows come from [`polyc_facts::verified_receipts`], the SAME
//! signature-verified fold the committed-spend budget gate, the dashboard
//! spend rollup, wallet history, and the trace projector's payment step all
//! read through (`crates/facts/src/receipts.rs`), not a second,
//! independently-written decode of `payment_receipt`/
//! `outbound_payment_receipt` payloads.
//!
//! # Why this table is different from every earlier one
//!
//! `usage`/`model_call`/`attribution`/`turn_failed` all decode a plain
//! protobuf payload via `buffa::Message::decode_from_slice`
//! (`crate::decode::decode_typed_kind_events`) — no trust decision involved,
//! any bytes that parse as the message type are a legitimate row. A payment
//! receipt is signed, self-describing JSON
//! (`polyc_crypto::approval::receipt_payload`), and "decodes cleanly" is not
//! the same question as "counts": the fact model
//! names the exact bug this table exists to prevent — reading these payloads
//! with a raw, unverified `serde_json::from_slice`, rather than verifying
//! the signature first the way every accounting path does, would let a
//! forged or tampered receipt render as if it settled. This table decodes
//! ONLY through
//! [`polyc_facts::verified_receipts`]/[`polyc_facts::verified_outbound_receipts`]
//! — the SINGLE definition of "a receipt that counts" — so a payment shown
//! by this table, the payment billing charges, and the payment the budget
//! gate enforces are the same fact by construction. [`decode_payments_events`]
//! therefore cannot reuse `crate::decode::decode_typed_kind_events` (that
//! helper is generic over `buffa::Message`, not the signed-JSON receipt
//! shape) and instead calls the shared fold directly, one event at a time —
//! see that function's docs for why it still needs a per-event loop rather
//! than handing the whole slice to the fold in one call.
//!
//! A forged/tampered receipt, or one signed by a key outside
//! `trusted_signers`, is not "a payment row with bad data" — the fold
//! returns `None` for it, and this module skips it exactly the way an
//! undecodable protobuf payload is skipped elsewhere in this crate. There is
//! no row for a receipt that doesn't verify, full stop.
//!
//! # `trusted_signers` (#1178's fold-coupled-table plumbing)
//!
//! Every other typed table in this crate needs nothing beyond the
//! caller-supplied `(position, Event)` pairs. This one additionally needs
//! the deployment's approval-signer public key, because "does this receipt
//! verify" is meaningless without an allow-list to check the embedded
//! signer against (see [`polyc_crypto::approval::verify_signed_receipt`]'s
//! own doc for why "internally consistent" and "trustworthy" are different
//! questions). `trusted_signers` reaches this module exactly the way it
//! reaches [`crate::trace::verified_payment_receipt_fields`]
//! (`crates/control-plane/src/trace.rs`) and
//! [`crate::grpc::verified_receipts`]/[`crate::grpc::verified_outbound_receipts`]
//! (`crates/control-plane/src/grpc/mod.rs`) today — the deployment's approval
//! public-key history from secret custody — but threaded one layer further
//! than any earlier typed table needed: `crate::authority::QueryAuthority`
//! holds that approval-family trust independently from session and turn-read
//! role trust; `crate::authority::ScopedQuery` carries it and passes
//! it to [`crate::engine::QueryEngine::build`], which passes it to
//! [`decode_payments_events`] at registration time. No new trust root is
//! introduced.
//!
//! # One table, a `direction` column (per the fact model doc's class table)
//!
//! `payment_receipt` (inbound: "we charged the caller") and
//! `outbound_payment_receipt` (outbound: "the control plane paid a 402
//! service on the agent's behalf") share the identical signed-JSON shape
//! ([`polyc_crypto::approval::ReceiptPayload`]) and the identical
//! [`polyc_crypto::approval::VerifiedReceipt`] output — the same "two kinds,
//! one fact" shape [`crate::decode::attribution`] already established for
//! `caller`/`participant`. `direction` carries exactly the two literal
//! strings the rest of the codebase already uses for this distinction
//! (`crates/control-plane/src/trace.rs`'s `verified_payment_receipt_fields`
//! call sites set `fields["direction"] = json!("inbound"|"outbound")` for
//! the identical two kinds) — `"inbound"` for a row folded from
//! [`polyc_proto::kinds::PAYMENT_RECEIPT`], `"outbound"` for one folded from
//! [`polyc_proto::kinds::OUTBOUND_PAYMENT_RECEIPT`] — so a query never needs
//! to know two different kind-base spellings to ask "which way did the
//! money move."
//!
//! [`VerifiedReceipt::kind`] (the signed event kind echoed inside the
//! payload itself) is deliberately NOT kept as its own column: it is the
//! same fact `direction` already carries in a form more useful to a query
//! (mirroring [`crate::decode::attribution`]'s reason for omitting
//! `kind_base` in favor of its own `role` column), and it is empty for a
//! legacy v1 receipt regardless of which kind-base it was stored under, so
//! keeping it would just be a second, sometimes-empty spelling of
//! `direction`.
//!
//! # Uniform keys (#1311)
//!
//! Same discipline as every other typed table: `partition`, `position`, and
//! `turn_id` are derived exactly the way
//! [`crate::decode::events_batch`]/[`crate::decode::attribution`] derive
//! them, via [`polyc_proto::kinds::parse`]. `event_time` is deferred to
//! #1327, unmodified from every earlier table's rationale.
//!
//! # Column selection
//!
//! Every [`VerifiedReceipt`] field is kept except `kind` (see above):
//! `reference`, `amount` (a string, never a float — but NOT one unit: an
//! inbound receipt stores a decimal figure in the settlement currency and an
//! outbound one stores the token's base units, so this column's unit is read
//! off the `direction` column beside it, exactly as
//! [`VerifiedReceipt::amount`] documents. A query that sums it across
//! directions sums two different units), `currency` (same per-direction split
//! — the settlement symbol inbound, the token contract address outbound),
//! `recipient`, `method`,
//! `timestamp` (RFC3339 string, not parsed into an Arrow timestamp type —
//! this table does not invent a parse the fold itself doesn't do),
//! `version`, `tool_call_id`, `approval_pos` (decimal string),
//! `approved_args_hash` (a sha256 hex digest, not the raw approved
//! arguments), `subject` (the opaque principal the spend is attributed to —
//! already the resolved, non-identifying id the fold hands every consumer,
//! the same status `attribution.persona_id` has), `payer_kind`/
//! `paying_account` (which account actually paid — see the "Payer
//! attribution" section below), and `signer_public_key`.
//!
//! # Payer attribution (v3, #2099)
//!
//! `payer_kind` (`"linked_wallet"`/`"deployment"`) and `paying_account` (the
//! payer's own address, empty for a deployment-signer payer) answer "did I
//! pay for that, or did the platform?" — computed at settlement and threaded
//! onto the receipt at the v2→v3 schema bump
//! ([`polyc_crypto::approval::RECEIPT_VERSION`]). Both columns are `Utf8`
//! **non-null**: a receipt that predates payer attribution (v1/v2) decodes
//! `payer_kind`/`paying_account` as `""`, the same explicit-unknown modeling
//! [`VerifiedReceipt::kind`]/the binding tuple already use for a v1 receipt —
//! never inferred, and never a nullable column standing in for "we don't
//! know" when the fold already has an unambiguous empty-string answer.
//!
//! # Redaction: `signer_public_key` is Fleet-only; payer columns are not
//!
//! The access-control boundary
//! conditions state plainly that raw signer keys stay out of
//! persona-scoped results, a redaction that applies to every non-maintainer
//! registration. `signer_public_key` is exactly that raw signer key — the
//! ed25519 public key that signed the receipt — so it gets the identical
//! raw-table/redacted-view treatment [`crate::decode::attribution`]'s
//! `identity_*` columns already established: [`crate::engine`] registers
//! this table's full decode as `payments_raw` for every scope (needed to
//! build the view), builds `payments` as a `CREATE VIEW` whose column list
//! is scope-dependent (every column for
//! [`Fleet`](crate::session::QueryScope::Fleet), every column except
//! `signer_public_key` for every other scope), then deregisters
//! `payments_raw` for every non-Fleet scope. Every OTHER column —
//! `subject` included — stays visible at every scope: `subject` is already
//! the resolved, opaque principal id the fold hands every consumer (the
//! same status `persona_id` has on `attribution`), not a raw external
//! identity, and `amount`/`recipient`/`reference`/etc. are the payment
//! ledger facts this table exists to expose to the accounting/wallet-history
//! consumers the fact model names, not identity data.
//! `paying_account` belongs in that same non-redacted class, not beside
//! `signer_public_key`: it is the reader's own linked wallet address, or the
//! deployment's — a participant's OWN account, never another participant's
//! external identity — exactly the distinction that already keeps
//! `recipient` (another payment fact, also potentially an address) visible
//! at every scope while `signer_public_key` (the deployment's own signing
//! key, not a payment fact) is Fleet-only.
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;
/// `direction`'s value for a row folded from [`kinds::PAYMENT_RECEIPT`] — the
/// control plane charged the caller. Matches the literal string
/// `crates/control-plane/src/trace.rs`'s trace projector already uses for
/// the identical distinction.
const DIRECTION_INBOUND: &str = "inbound";
/// `direction`'s value for a row folded from [`kinds::OUTBOUND_PAYMENT_RECEIPT`]
/// — the control plane paid a 402 service on the caller's behalf. Matches
/// the literal string `crates/control-plane/src/trace.rs`'s trace projector
/// already uses for the identical distinction.
const DIRECTION_OUTBOUND: &str = "outbound";
/// One decoded, SIGNATURE-VERIFIED `payments` row.
///
/// The fact model's uniform key columns (`partition`, `position`, `turn_id`)
/// plus `direction` and every [`polyc_crypto::approval::VerifiedReceipt`]
/// field except `kind` (see the module docs' "Column selection" section for
/// why `kind` itself is dropped).
#[derive(Debug, Clone)]
pub(crate) struct PaymentRow {
/// The journal partition this row's event was read from (`conv-{id}`).
pub partition: String,
/// The journal's own monotonic append position for this event.
pub position: u64,
/// The `:{turn_uuid}` suffix off the event's `kind`, canonical
/// hyphenated form, or `None` for a bare `payment_receipt`/
/// `outbound_payment_receipt` kind with no turn tagged.
pub turn_id: Option<String>,
/// `"inbound"` for a row folded from [`kinds::PAYMENT_RECEIPT`],
/// `"outbound"` for one folded from [`kinds::OUTBOUND_PAYMENT_RECEIPT`].
pub direction: String,
/// [`polyc_crypto::approval::VerifiedReceipt::reference`].
pub reference: String,
/// [`polyc_crypto::approval::VerifiedReceipt::amount`] — decimal as a
/// string.
pub amount: String,
/// [`polyc_crypto::approval::VerifiedReceipt::currency`].
pub currency: String,
/// [`polyc_crypto::approval::VerifiedReceipt::recipient`].
pub recipient: String,
/// [`polyc_crypto::approval::VerifiedReceipt::method`].
pub method: String,
/// [`polyc_crypto::approval::VerifiedReceipt::timestamp`] — RFC3339,
/// stored verbatim as `Utf8`, not parsed into an Arrow timestamp type.
pub timestamp: String,
/// [`polyc_crypto::approval::VerifiedReceipt::version`].
pub version: u64,
/// [`polyc_crypto::approval::VerifiedReceipt::tool_call_id`].
pub tool_call_id: String,
/// [`polyc_crypto::approval::VerifiedReceipt::approval_pos`] — decimal
/// string of the `approval_request` log position.
pub approval_pos: String,
/// [`polyc_crypto::approval::VerifiedReceipt::approved_args_hash`] —
/// sha256 hex of the approved arguments, not the raw arguments
/// themselves.
pub approved_args_hash: String,
/// [`polyc_crypto::approval::VerifiedReceipt::subject`] — the opaque
/// principal the spend is attributed to.
pub subject: String,
/// [`polyc_crypto::approval::VerifiedReceipt::payer_kind`] — `""` for a
/// receipt that predates payer attribution (v1/v2), never inferred.
pub payer_kind: String,
/// [`polyc_crypto::approval::VerifiedReceipt::paying_account`] — `""`
/// for a receipt that predates payer attribution, a deployment-signer
/// payer, or an unobservable payer.
pub paying_account: String,
/// [`polyc_crypto::approval::VerifiedReceipt::signer_public_key`] —
/// Fleet-only at registration time, see the module docs' "Redaction"
/// section.
pub signer_public_key: Vec<u8>,
}
/// The `payments` typed table's full Arrow schema (every column, including
/// `signer_public_key`) — what `payments_raw` registers as, for every scope.
///
/// `partition` (`Utf8`, non-null), `position` (`UInt64`, non-null),
/// `turn_id` (`Utf8`, nullable), `direction` (`Utf8`, non-null), `reference`
/// (`Utf8`, non-null), `amount` (`Utf8`, non-null), `currency` (`Utf8`,
/// non-null), `recipient` (`Utf8`, non-null), `method` (`Utf8`, non-null),
/// `timestamp` (`Utf8`, non-null), `version` (`UInt64`, non-null),
/// `tool_call_id` (`Utf8`, non-null), `approval_pos` (`Utf8`, non-null),
/// `approved_args_hash` (`Utf8`, non-null), `subject` (`Utf8`, non-null),
/// `signer_public_key` (`Binary`, non-null), `payer_kind` (`Utf8`, non-null),
/// `paying_account` (`Utf8`, non-null).
///
/// `payer_kind`/`paying_account` are appended at the end rather than beside
/// the other receipt facts: they're the newest column pair (v3), and
/// appending keeps every existing consumer's positional column index
/// (`batch.column(N)`) stable across the schema change.
///
/// See the module docs for why `kind`/`kind_base`/`conversation_id` are
/// omitted, why `event_time` is deferred to #1327, and why
/// `signer_public_key` is registered here but hidden from the redacted
/// `payments` view for every non-Fleet scope.
#[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("direction", DataType::Utf8, false),
Field::new("reference", DataType::Utf8, false),
Field::new("amount", DataType::Utf8, false),
Field::new("currency", DataType::Utf8, false),
Field::new("recipient", DataType::Utf8, false),
Field::new("method", DataType::Utf8, false),
Field::new("timestamp", DataType::Utf8, false),
Field::new("version", DataType::UInt64, false),
Field::new("tool_call_id", DataType::Utf8, false),
Field::new("approval_pos", DataType::Utf8, false),
Field::new("approved_args_hash", DataType::Utf8, false),
Field::new("subject", DataType::Utf8, false),
Field::new("signer_public_key", DataType::Binary, false),
Field::new("payer_kind", DataType::Utf8, false),
Field::new("paying_account", DataType::Utf8, false),
]))
}
/// Decode already-framed [`PaymentRow`]s into the `payments_raw` table's
/// Arrow `RecordBatch`, in [`schema`] order.
///
/// # Errors
///
/// Returns [`ArrowError`] if Arrow batch construction fails.
pub(crate) fn decode_payments_batch(rows: &[PaymentRow]) -> 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 direction_b = StringBuilder::with_capacity(rows.len(), rows.len() * 8);
let mut reference_b = StringBuilder::with_capacity(rows.len(), rows.len() * 16);
let mut amount_b = StringBuilder::with_capacity(rows.len(), rows.len() * 8);
let mut currency_b = StringBuilder::with_capacity(rows.len(), rows.len() * 4);
let mut recipient_b = StringBuilder::with_capacity(rows.len(), rows.len() * 16);
let mut method_b = StringBuilder::with_capacity(rows.len(), rows.len() * 8);
let mut timestamp_b = StringBuilder::with_capacity(rows.len(), rows.len() * 24);
let mut version_b = UInt64Builder::with_capacity(rows.len());
let mut tool_call_id_b = StringBuilder::with_capacity(rows.len(), rows.len() * 16);
let mut approval_pos_b = StringBuilder::with_capacity(rows.len(), rows.len() * 8);
let mut approved_args_hash_b = StringBuilder::with_capacity(rows.len(), rows.len() * 64);
let mut subject_b = StringBuilder::with_capacity(rows.len(), rows.len() * 16);
let mut signer_public_key_b = BinaryBuilder::with_capacity(rows.len(), rows.len() * 32);
let mut payer_kind_b = StringBuilder::with_capacity(rows.len(), rows.len() * 16);
let mut paying_account_b = StringBuilder::with_capacity(rows.len(), rows.len() * 16);
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(),
}
direction_b.append_value(&row.direction);
reference_b.append_value(&row.reference);
amount_b.append_value(&row.amount);
currency_b.append_value(&row.currency);
recipient_b.append_value(&row.recipient);
method_b.append_value(&row.method);
timestamp_b.append_value(&row.timestamp);
version_b.append_value(row.version);
tool_call_id_b.append_value(&row.tool_call_id);
approval_pos_b.append_value(&row.approval_pos);
approved_args_hash_b.append_value(&row.approved_args_hash);
subject_b.append_value(&row.subject);
signer_public_key_b.append_value(&row.signer_public_key);
payer_kind_b.append_value(&row.payer_kind);
paying_account_b.append_value(&row.paying_account);
}
let columns: Vec<ArrayRef> = vec![
Arc::new(partition_b.finish()),
Arc::new(position_b.finish()),
Arc::new(turn_id_b.finish()),
Arc::new(direction_b.finish()),
Arc::new(reference_b.finish()),
Arc::new(amount_b.finish()),
Arc::new(currency_b.finish()),
Arc::new(recipient_b.finish()),
Arc::new(method_b.finish()),
Arc::new(timestamp_b.finish()),
Arc::new(version_b.finish()),
Arc::new(tool_call_id_b.finish()),
Arc::new(approval_pos_b.finish()),
Arc::new(approved_args_hash_b.finish()),
Arc::new(subject_b.finish()),
Arc::new(signer_public_key_b.finish()),
Arc::new(payer_kind_b.finish()),
Arc::new(paying_account_b.finish()),
];
RecordBatch::try_new(schema(), columns)
}
/// Filter `partition`'s framed `events` to `payment_receipt`/
/// `outbound_payment_receipt` rows, fold each payload through the SHARED
/// verified-receipt fold (`polyc_facts::verified_receipts`), and pair a
/// successfully-verified receipt with that row's uniform key columns.
///
/// Unlike every other typed-table decoder in this crate, this one cannot
/// delegate to `crate::decode::decode_typed_kind_events` — that helper
/// decodes a plain `buffa::Message` protobuf payload, and a payment receipt
/// is signed JSON verified through a completely different fold
/// (`polyc_crypto::approval::verify_signed_receipt`, wrapped by
/// [`polyc_facts::verified_receipts`]). This function still preserves the
/// SAME "no second decode" guarantee the fold model requires: it never
/// calls `verify_signed_receipt` itself, only `polyc_facts::verified_receipts`
/// — the identical fold `crate::trace::verified_payment_receipt_fields`
/// (`crates/control-plane/src/trace.rs`) already calls for the trace
/// projector's payment step, via the identical `std::slice::from_ref`
/// singleton-slice idiom that call site established (needed here for the
/// same reason: `verified_receipts` walks a slice of `Event` with no
/// position of its own, so this loop pairs each event with its journal
/// position ONE AT A TIME rather than losing that association by handing
/// the whole partition slice to the fold in one call).
///
/// A payload that fails to verify — malformed JSON, a signature that
/// doesn't check out, or a signer outside `trusted_signers` — is silently
/// skipped, exactly like `verified_payment_receipt_fields`'s own contract:
/// there is no row for a receipt that doesn't verify, forged or otherwise.
#[must_use]
pub(crate) fn decode_payments_events(
partition: &str,
events: &[(u64, Event)],
trusted_signers: &[Vec<u8>],
) -> Vec<PaymentRow> {
events
.iter()
.filter_map(|(position, event)| {
let (base, turn_id) = kinds::parse(&event.kind);
let direction = if base == kinds::PAYMENT_RECEIPT {
DIRECTION_INBOUND
} else if base == kinds::OUTBOUND_PAYMENT_RECEIPT {
DIRECTION_OUTBOUND
} else {
return None;
};
let receipt =
polyc_facts::verified_receipts(std::slice::from_ref(event), base, trusted_signers)
.next()?;
Some(PaymentRow {
partition: partition.to_string(),
position: *position,
turn_id: turn_id.map(|id| id.to_string()),
direction: direction.to_string(),
reference: receipt.reference,
amount: receipt.amount,
currency: receipt.currency,
recipient: receipt.recipient,
method: receipt.method,
timestamp: receipt.timestamp,
version: receipt.version,
tool_call_id: receipt.tool_call_id,
approval_pos: receipt.approval_pos,
approved_args_hash: receipt.approved_args_hash,
subject: receipt.subject,
payer_kind: receipt.payer_kind,
paying_account: receipt.paying_account,
signer_public_key: receipt.signer_public_key,
})
})
.collect()
}
#[cfg(test)]
mod tests {
use arrow::array::Array as _;
use polyc_crypto::approval::{ApprovalSigner, ReceiptPayload, receipt_payload};
use uuid::Uuid;
use super::*;
/// A fully-named v2 `payment_receipt`/`outbound_payment_receipt` payload,
/// signed by `signer` — mirrors `crates/facts/src/receipts.rs`'s own test
/// fixture shape (named fields, no `..Default::default()` spread, since
/// `ReceiptPayload` has no default).
///
/// `amount`/`currency` follow the DIRECTION's production shape, not one
/// shared shape (#1739): a decimal figure in the settlement currency
/// inbound, base units in the settlement token outbound. A fixture that
/// wrote a dollar string under the outbound kind is exactly what let the
/// unit mix-up survive review.
fn signed_receipt(signer: &ApprovalSigner, kind: &'static str, tool_call_id: &str) -> Vec<u8> {
let (amount, currency) = if kind == kinds::OUTBOUND_PAYMENT_RECEIPT {
("1500000", "0xtoken")
} else {
("1.50", "USD")
};
let (payload, sig, pk) = receipt_payload(
&ReceiptPayload {
kind,
reference: "tx-1",
amount,
currency,
recipient: "0xrecipient",
method: "tempo",
timestamp: "2026-07-20T00:00:00Z",
tool_call_id,
approval_pos: "1",
approved_args_hash: "hash-abc",
subject: "persona-1",
payer_kind: "linked_wallet",
paying_account: "0xpayer",
},
signer,
);
let _ = (sig, pk);
payload
}
/// A frozen v2 `outbound_payment_receipt` — predates payer attribution.
/// Same bytes `crates/facts/src/receipts.rs`'s `signed_v2_receipt`
/// fixture uses (signed by `ApprovalSigner::from_seed(99)`).
fn signed_v2_receipt() -> Vec<u8> {
br#"{"version":2,"kind":"outbound_payment_receipt","reference":"tx-frozen-v2","amount":"10000","currency":"0xToken","recipient":"0xrecipient","method":"tempo","timestamp":"2026-06-02T00:00:00Z","tool_call_id":"call-frozen","approval_pos":"42","approved_args_hash":"abcd1234","subject":"conv-frozen","signed_by":"35ccaf567ce385fe73a3d0ef44c04c8e4f13cf02ec853ac430a8bfd40cefe008","signature_hex":"4f73999b3885188a976c4a2da45c0fc5b6917102fd5a17a10efbea7c9dc9c13216602de44b52388f6811d07ad3a9225288528828f05d436e8c1d3241e472c809"}"#.to_vec()
}
#[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",
"direction",
"reference",
"amount",
"currency",
"recipient",
"method",
"timestamp",
"version",
"tool_call_id",
"approval_pos",
"approved_args_hash",
"subject",
"signer_public_key",
"payer_kind",
"paying_account",
]
);
let expect = [
("partition", DataType::Utf8, false),
("position", DataType::UInt64, false),
("turn_id", DataType::Utf8, true),
("direction", DataType::Utf8, false),
("reference", DataType::Utf8, false),
("amount", DataType::Utf8, false),
("currency", DataType::Utf8, false),
("recipient", DataType::Utf8, false),
("method", DataType::Utf8, false),
("timestamp", DataType::Utf8, false),
("version", DataType::UInt64, false),
("tool_call_id", DataType::Utf8, false),
("approval_pos", DataType::Utf8, false),
("approved_args_hash", DataType::Utf8, false),
("subject", DataType::Utf8, false),
("signer_public_key", DataType::Binary, false),
("payer_kind", DataType::Utf8, false),
("paying_account", DataType::Utf8, false),
];
for (field, (name, ty, nullable)) in schema.fields().iter().zip(expect) {
assert_eq!(field.name(), name);
assert_eq!(field.data_type(), &ty);
assert_eq!(field.is_nullable(), nullable);
}
}
#[test]
fn decode_payments_batch_round_trips() {
let rows = vec![PaymentRow {
partition: "conv-a".to_string(),
position: 5,
turn_id: Some("turn-xyz".to_string()),
direction: "inbound".to_string(),
reference: "tx-1".to_string(),
amount: "1.50".to_string(),
currency: "USDC".to_string(),
recipient: "0xrecipient".to_string(),
method: "tempo".to_string(),
timestamp: "2026-07-20T00:00:00Z".to_string(),
version: 2,
tool_call_id: "call-1".to_string(),
approval_pos: "1".to_string(),
approved_args_hash: "hash-abc".to_string(),
subject: "persona-1".to_string(),
payer_kind: "linked_wallet".to_string(),
paying_account: "0xpayer".to_string(),
signer_public_key: vec![1, 2, 3, 4],
}];
let batch = decode_payments_batch(&rows).expect("batch build");
assert_eq!(batch.num_rows(), 1);
assert_eq!(batch.schema(), schema());
let direction = batch
.column(3)
.as_any()
.downcast_ref::<arrow::array::StringArray>()
.unwrap();
assert_eq!(direction.value(0), "inbound");
let amount = batch
.column(5)
.as_any()
.downcast_ref::<arrow::array::StringArray>()
.unwrap();
assert_eq!(amount.value(0), "1.50");
let signer_public_key = batch
.column(15)
.as_any()
.downcast_ref::<arrow::array::BinaryArray>()
.unwrap();
assert_eq!(signer_public_key.value(0), &[1, 2, 3, 4]);
let payer_kind = batch
.column(16)
.as_any()
.downcast_ref::<arrow::array::StringArray>()
.unwrap();
assert_eq!(payer_kind.value(0), "linked_wallet");
let paying_account = batch
.column(17)
.as_any()
.downcast_ref::<arrow::array::StringArray>()
.unwrap();
assert_eq!(paying_account.value(0), "0xpayer");
}
/// A real signed `payment_receipt` (inbound) decodes to one row with
/// `direction = "inbound"` and every receipt field intact, once its
/// signer is in `trusted_signers`.
#[test]
fn decode_payments_events_verifies_and_decodes_inbound_receipt() {
let turn = Uuid::from_u128(0x0195_abcd_ef01_2345_6789_abcd_ef01_2345);
let signer = ApprovalSigner::from_seed(1);
let bytes = signed_receipt(&signer, kinds::PAYMENT_RECEIPT, "call-1");
let events = vec![
(1, Event::new(kinds::TURN_START, Vec::new())),
(
2,
Event::new(kinds::tagged(kinds::PAYMENT_RECEIPT, &turn), bytes),
),
];
let trusted_signers = vec![signer.public_key_bytes()];
let decoded = decode_payments_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].direction, "inbound");
assert_eq!(decoded[0].reference, "tx-1");
assert_eq!(decoded[0].amount, "1.50", "inbound stores a decimal figure");
assert_eq!(decoded[0].currency, "USD");
assert_eq!(decoded[0].recipient, "0xrecipient");
assert_eq!(decoded[0].method, "tempo");
assert_eq!(decoded[0].tool_call_id, "call-1");
assert_eq!(decoded[0].approval_pos, "1");
assert_eq!(decoded[0].approved_args_hash, "hash-abc");
assert_eq!(decoded[0].subject, "persona-1");
assert_eq!(decoded[0].payer_kind, "linked_wallet");
assert_eq!(decoded[0].paying_account, "0xpayer");
assert_eq!(decoded[0].signer_public_key, signer.public_key_bytes());
}
/// A receipt that predates payer attribution (v2) decodes with
/// `payer_kind`/`paying_account` as explicit unknown (`""`), never
/// inferred — the same modeling `VerifiedReceipt` uses upstream.
#[test]
fn decode_payments_events_pre_payer_receipt_decodes_payer_as_unknown() {
let signer = ApprovalSigner::from_seed(99);
let events = vec![(
1,
Event::new(kinds::OUTBOUND_PAYMENT_RECEIPT, signed_v2_receipt()),
)];
let trusted_signers = vec![signer.public_key_bytes()];
let decoded = decode_payments_events("conv-pre-payer", &events, &trusted_signers);
assert_eq!(decoded.len(), 1);
assert_eq!(decoded[0].version, 2);
assert!(decoded[0].payer_kind.is_empty());
assert!(decoded[0].paying_account.is_empty());
}
/// The outbound sibling: a real signed `outbound_payment_receipt`
/// decodes with `direction = "outbound"`.
#[test]
fn decode_payments_events_verifies_and_decodes_outbound_receipt() {
let turn = Uuid::from_u128(0x0195_abcd_ef01_2345_6789_abcd_ef01_9999);
let signer = ApprovalSigner::from_seed(2);
let bytes = signed_receipt(&signer, kinds::OUTBOUND_PAYMENT_RECEIPT, "call-2");
let events = vec![(
1,
Event::new(kinds::tagged(kinds::OUTBOUND_PAYMENT_RECEIPT, &turn), bytes),
)];
let trusted_signers = vec![signer.public_key_bytes()];
let decoded = decode_payments_events("conv-out", &events, &trusted_signers);
assert_eq!(decoded.len(), 1);
assert_eq!(decoded[0].direction, "outbound");
assert_eq!(decoded[0].tool_call_id, "call-2");
// The unit this direction really stores: base units, not dollars. The
// `direction` column beside it is what tells a query which unit the
// `amount` column holds.
assert_eq!(decoded[0].amount, "1500000");
assert_eq!(decoded[0].currency, "0xtoken");
}
/// The fold-coupling's own proof: a receipt signed by a key OUTSIDE
/// `trusted_signers` — a forged or self-signed-by-an-unknown-key
/// payload, internally consistent but not trusted — never surfaces as a
/// row. This is not a decode failure being skipped; it is the shared
/// fold's own trust boundary, inherited here rather than re-implemented.
#[test]
fn decode_payments_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_receipt(&untrusted, kinds::PAYMENT_RECEIPT, "call-forged");
let events = vec![(
1,
Event::new(kinds::tagged(kinds::PAYMENT_RECEIPT, &turn), forged),
)];
let trusted_signers = vec![trusted.public_key_bytes()];
let decoded = decode_payments_events("conv-forged", &events, &trusted_signers);
assert_eq!(
decoded.len(),
0,
"a receipt signed by a key outside trusted_signers must never appear as a row"
);
}
/// A structurally malformed payload (not even valid JSON) is skipped the
/// same way an unverified signature is — no row, no panic.
#[test]
fn decode_payments_events_drops_a_malformed_payload() {
let events = vec![(
1,
Event::new(kinds::PAYMENT_RECEIPT, vec![0xFF, 0xFE, 0xFD]),
)];
let decoded = decode_payments_events("conv-corrupt", &events, &[]);
assert_eq!(decoded.len(), 0);
}
#[test]
fn decode_payments_events_bare_kind_has_no_turn_id() {
let signer = ApprovalSigner::from_seed(5);
let bytes = signed_receipt(&signer, kinds::PAYMENT_RECEIPT, "call-bare");
let events = vec![(7, Event::new(kinds::PAYMENT_RECEIPT, bytes))];
let trusted_signers = vec![signer.public_key_bytes()];
let decoded = decode_payments_events("conv-bare", &events, &trusted_signers);
assert_eq!(decoded.len(), 1);
assert_eq!(decoded[0].turn_id, None);
}
/// An event kind outside `{payment_receipt, outbound_payment_receipt}`
/// is not payment data at all and must never surface as a row here, even
/// if its payload happens to be a validly-signed receipt.
#[test]
fn unrelated_kind_is_not_decoded_as_a_payment() {
let signer = ApprovalSigner::from_seed(6);
let bytes = signed_receipt(&signer, kinds::PAYMENT_RECEIPT, "call-x");
let events = vec![(1, Event::new(kinds::USAGE, bytes))];
let trusted_signers = vec![signer.public_key_bytes()];
let decoded = decode_payments_events("conv-unrelated", &events, &trusted_signers);
assert_eq!(
decoded.len(),
0,
"a usage-kind event must never decode as a payment"
);
}
#[test]
fn payment_event_round_trip_through_decode_and_batch() {
let turn = Uuid::from_u128(0x0195_abcd_ef01_2345_6789_abcd_ef01_3333);
let signer = ApprovalSigner::from_seed(7);
let bytes = signed_receipt(&signer, kinds::PAYMENT_RECEIPT, "call-rt");
let events = vec![(
1,
Event::new(kinds::tagged(kinds::PAYMENT_RECEIPT, &turn), bytes),
)];
let trusted_signers = vec![signer.public_key_bytes()];
let decoded = decode_payments_events("conv-rt", &events, &trusted_signers);
let batch = decode_payments_batch(&decoded).expect("batch build");
let turn_id = batch
.column(2)
.as_any()
.downcast_ref::<arrow::array::StringArray>()
.unwrap();
assert_eq!(turn_id.value(0), turn.to_string());
let subject = batch
.column(14)
.as_any()
.downcast_ref::<arrow::array::StringArray>()
.unwrap();
assert_eq!(subject.value(0), "persona-1");
}
}