polyc-query 2026.9.0

//! `refusals` typed-table decoder (`#2090`, INV-W5) — the `payments`
//! sibling for durable payment-refusal events: rows come from
//! [`polyc_facts::verified_refusals`], the SAME signature-verified fold
//! `crates/facts/src/refusals.rs` owns, not a second, independently-written
//! decode of `payment_refusal` payloads. See
//! [`crate::decode::payments`]'s module docs for the full rationale this
//! table follows verbatim (fold-coupled decode, why it cannot reuse
//! `decode_typed_kind_events`, the `trusted_signers` threading) — this
//! module states only what's different for refusals.
//!
//! # Uniform keys (`#1311`)
//!
//! Same discipline as every other typed table: `partition`, `position`, and
//! `turn_id` are derived via [`polyc_proto::kinds::parse`]. Unlike a settled
//! receipt (persisted inside the end-of-turn `persist_turn` batch, tagged
//! with the SAME real `turn_id`), a `payment_refusal` event is appended in
//! its OWN atomic batch at the reject site itself
//! (`crates/control-plane/src/harness_dialer.rs`'s `record_refusal_if_rejected`)
//! — but it is STILL tagged with the refused turn's own real `turn_id` when
//! one is known, so `turn_id` here joins back to the same conversation/turn a
//! receipt or usage row would. `event_time` is deferred to #1327, unmodified
//! from every earlier table's rationale.
//!
//! # Column selection
//!
//! Every [`polyc_crypto::approval::VerifiedRefusal`] field is kept:
//! `reason` (a stable, machine-readable tag — never the free-text `Display`
//! rendering a future reason variant might carry), `reason_detail` (a
//! non-secret diagnostic string), `merchant_host`, `requested_base_units`/
//! `permitted_base_units` (decimal strings, empty for a reason that carries
//! no cap), `tool_call_id`, `subject` (the opaque principal, same status
//! `payments.subject` has), `timestamp` (decimal string of the unix-seconds
//! clock value the reject site recorded at, the `payments.timestamp`
//! sibling — without it a refused row carries no time and cannot be placed
//! on a ledger beside settled receipts), and `signer_public_key`.
//!
//! # Redaction: `signer_public_key` is Fleet-only
//!
//! The identical raw-table/redacted-view treatment
//! [`crate::decode::payments`]'s module docs describe for its own
//! `signer_public_key` column: [`crate::engine`] registers this table's full
//! decode as `refusals_raw` for every scope, builds `refusals` 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
//! `refusals_raw` for every non-Fleet scope. Every other column — including
//! `reason_detail` and `merchant_host` — stays visible at every scope: they
//! are refusal facts about the caller's OWN attempted payment, not another
//! participant's identity or key material.
//!
//! # Decode totality (INV-W5b)
//!
//! [`decode_refusals_events`] never panics and never drops a structurally
//! valid, verified refusal event, regardless of `reason`'s value: `reason`
//! is carried through verbatim as whatever string the signed payload names,
//! with no validation against a known tag set at decode time. A future
//! reason variant this build has no tag for still decodes to a normal row —
//! the "unknown" case is a PROJECTION concern for a consumer that wants to
//! render a friendly label (e.g. "unknown reason"), not something this fact
//! table itself decides, so no known-tag allowlist gates this module's
//! production decode path at all (see this module's tests for the tag set
//! `crates/control-plane/src/harness_dialer.rs`'s `classify_reject_reason`
//! mints today, pinned there only to prove the "unknown tag still decodes"
//! property, INV-W5b).
//!
//! # Committed-turn filter: not applied — see [`crate::engine::tables::REFUSALS_TABLE`]
//!
//! The view built on top of this table's `refusals_raw` registration
//! deliberately omits the committed-turn `WHERE EXISTS` filter every other
//! scope-dependent view in this crate carries — see that constant's own doc
//! for the reasoning.

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;

/// The stable, machine-readable reason tags this build recognizes — the
/// read-side mirror of `classify_reject_reason`'s and (`#2090` follow-up)
/// `classify_delegated_key_resolution`'s exhaustive matches, plus three
/// non-enum-driven `fulfill_payment` early-return sites
/// (`crates/control-plane/src/harness_dialer.rs`; this crate has no
/// dependency on that one, so the lists are kept in lockstep by the tests in
/// each rather than by a shared type). `unlinked`/`linked_but_unusable`/
/// `temporarily_unavailable` come from the pre-proxy `DelegatedKeyResolution`
/// seam; `payments_not_configured`/`no_approved_call`/`invalid_approved_args`
/// (`#2090` review round 2) come from three `fulfill_payment` early returns
/// that refuse before either exhaustive match ever runs — same event kind,
/// same `reason` column throughout, no schema change needed for any of them.
/// A tag NOT in this list is still a valid row (see the module docs' "Decode
/// totality" section) — this is test-only support for pinning that INV-W5b
/// property, not a filter any production decode path consults.
#[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",
];

/// Whether `reason` is one of [`KNOWN_REASONS`] — test-only, see that
/// constant's own doc.
#[cfg(test)]
fn is_known_reason(reason: &str) -> bool {
    KNOWN_REASONS.contains(&reason)
}

/// One decoded, SIGNATURE-VERIFIED `refusals` row.
#[derive(Debug, Clone)]
pub(crate) struct RefusalRow {
    /// 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_refusal` kind with no
    /// turn tagged.
    pub turn_id: Option<String>,
    /// [`polyc_crypto::approval::VerifiedRefusal::reason`].
    pub reason: String,
    /// [`polyc_crypto::approval::VerifiedRefusal::reason_detail`].
    pub reason_detail: String,
    /// [`polyc_crypto::approval::VerifiedRefusal::merchant_host`].
    pub merchant_host: String,
    /// [`polyc_crypto::approval::VerifiedRefusal::requested_base_units`].
    pub requested_base_units: String,
    /// [`polyc_crypto::approval::VerifiedRefusal::permitted_base_units`].
    pub permitted_base_units: String,
    /// [`polyc_crypto::approval::VerifiedRefusal::tool_call_id`].
    pub tool_call_id: String,
    /// [`polyc_crypto::approval::VerifiedRefusal::subject`].
    pub subject: String,
    /// [`polyc_crypto::approval::VerifiedRefusal::timestamp`] — decimal
    /// string of the unix-seconds clock value the reject site recorded at.
    pub timestamp: String,
    /// [`polyc_crypto::approval::VerifiedRefusal::signer_public_key`] —
    /// Fleet-only at registration time, see the module docs' "Redaction"
    /// section.
    pub signer_public_key: Vec<u8>,
}

/// The `refusals` typed table's full Arrow schema (every column, including
/// `signer_public_key`) — what `refusals_raw` registers as, for every 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("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),
    ]))
}

/// Decode already-framed [`RefusalRow`]s into the `refusals_raw` table's
/// Arrow `RecordBatch`, in [`schema`] order.
///
/// # Errors
///
/// Returns [`ArrowError`] if Arrow batch construction fails.
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)
}

/// Filter `partition`'s framed `events` to `payment_refusal` rows, fold each
/// payload through the SHARED verified-refusal fold
/// (`polyc_facts::verified_refusals`), and pair a successfully-verified
/// refusal with that row's uniform key columns.
///
/// Unlike every other typed-table decoder that reads a plain protobuf
/// payload, this one cannot delegate to `crate::decode::decode_typed_kind_events`
/// — a `payment_refusal` is signed JSON, verified through
/// [`polyc_crypto::approval::verify_signed_refusal`], wrapped by
/// [`polyc_facts::verified_refusals`]. Mirrors
/// [`crate::decode::payments::decode_payments_events`]'s own one-event-at-a-time
/// loop for the identical reason: `verified_refusals` walks a slice of
/// `Event` with no position of its own, so pairing happens one event at a
/// time rather than losing the association across the whole partition slice.
///
/// A payload that fails to verify — malformed JSON, a signature that
/// doesn't check out, or a signer outside `trusted_signers` — is silently
/// skipped: there is no row for a refusal that doesn't verify, forged or
/// otherwise. This is INV-W5b's OTHER half (an unverifiable payload is
/// dropped) — a verified payload's `reason`, whatever string it names, is
/// NEVER dropped; see the module docs' "Decode totality" section.
#[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());
    }

    /// INV-W5b: an unrecognized reason tag (a variant a future binary added,
    /// which this build has no [`KNOWN_REASONS`] entry for) still decodes to
    /// a normal row — never a panic, never a dropped event.
    #[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"));
    }
}