polyc-query 2026.8.3

//! `grant_replays` typed-table decoder — the fact model's ninth typed table,
//! and the THIRD FOLD-COUPLED one (after [`crate::decode::payments`] and
//! [`crate::decode::approvals`]): rows come from
//! [`polyc_facts::fold_grant_replay_event`], the SAME fold
//! `forensics::parse_grant_replay_entry` and `trace::decode_grant_replay_fields`
//! now read through, not a second, independently-written decode of a signed
//! `grant_replay` payload.
//!
//! # One table, no discriminator (unlike `payments`/`approvals`/`handoffs`)
//!
//! `grant_replay` is a single signed-JSON shape — there is no request/response
//! pairing or multi-phase kind family the way `approval_request`/
//! `approval_response` or `handoff`/`handoff_denied` are, so
//! this table needs no `direction`/`phase` discriminator column: every row
//! carries every column, none of them `NULL` by construction (see
//! [`crate::decode::approvals`]'s module docs for the discriminated shape
//! this table deliberately does NOT need).
//!
//! # `trusted_signers` (threaded exactly like `payments`/`approvals`)
//!
//! Reaches this module the identical way it reaches
//! [`crate::decode::approvals::decode_approvals_events`]: `trusted_signers`
//! becomes a per-event input at [`crate::engine::QueryEngine::build`],
//! sourced from `crate::authority::ScopedQuery`'s own `trusted_signers`
//! field — the deployment's approval-signer public key. No new trust root.
//!
//! # CRITICAL POSTURE: unlike `payments`, an unverified record is KEPT
//!
//! [`crate::decode::payments::decode_payments_events`] drops any payment
//! receipt whose signature does not verify. `grant_replays` does NOT drop a
//! `grant_replay` that fails verification — `crates/facts/src/grant_replays.rs`'s
//! module docs state the reason plainly: the `/grant-replays` audit surface
//! (`crates/control-plane/src/forensics.rs`'s `collect_grant_replays`) exists
//! to show exactly this as an audit signal, a tampered or forged record
//! tagged `invalid` rather than hidden. [`decode_grant_replays_events`]
//! therefore keeps a row for every structurally-decodable `grant_replay`
//! regardless of `signature_status`; only a payload too malformed to yield
//! even `tool`/`grant_ref`/`covered_capabilities`/`coverage_hash` (the
//! [`polyc_facts::fold_grant_replay_event`] contract) drops the row — see
//! that function's own tests for the contrast with `payments`' drop test.
//!
//! # Column selection and redaction
//!
//! NON-NULL for every row: `partition`, `position`, `tool`, `grant_ref`,
//! `covered_capabilities`, `coverage_hash`, `signature_status`. Nullable:
//! `turn_id` (absent for a bare, un-tagged event — never happens in practice,
//! since `crate::approval::grant_replay_event` always tags with the turn, but
//! the column stays nullable for the same reason every other typed table's
//! `turn_id` does) and `signer_public_key` (absent only if the payload's own
//! `signed_by` were empty, which the encoder never produces). The payload's
//! own embedded `conversation_id`/`turn_id` fields are deliberately omitted —
//! redundant with the `partition`/`turn_id` key columns already on every row,
//! the identical rationale [`crate::decode::approvals`]'s module docs give
//! for dropping `conversation_id`. `covered_capabilities` is stored as `Utf8`
//! JSON-array text, matching every other JSON-shaped string column this
//! crate exposes (`handoffs.allowed`, `tool_calls.arguments`/`.result`)
//! rather than a native Arrow list type.
//!
//! Fleet-only (redacted out of the `grant_replays` view for every other
//! scope, `crate::views::GRANT_REPLAYS_REDACTED_VIEW_SQL`): `signer_public_key`
//! — docs/reference/datafusion-data-layer.md's boundary conditions put raw
//! signer keys out of persona scope unconditionally. Every other column is
//! exactly `forensics::collect_grant_replays`' own participant-visible set
//! (`GrantReplayEntry`'s `tool`/`grant_ref`/`covered_capabilities`/
//! `coverage_hash`/`signature` fields), so persona-visible:
//! `partition`, `position`, `turn_id`, `tool`, `grant_ref`,
//! `covered_capabilities`, `coverage_hash`, `signature_status`.

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;

/// [`polyc_facts::GrantReplaySignatureStatus::Verified`]'s column string —
/// mirrors `trace.rs`'s own `"verified"`/`"invalid"` literals.
const SIGNATURE_VERIFIED: &str = "verified";
/// [`polyc_facts::GrantReplaySignatureStatus::Invalid`]'s column string.
const SIGNATURE_INVALID: &str = "invalid";

/// One decoded `grant_replays` row — every field
/// [`polyc_facts::GrantReplayFact`] carries, plus the fact model's uniform
/// key columns (`partition`, `position`, `turn_id`). No column is `NULL` by
/// construction (see the module docs' "One table, no discriminator"
/// section), `turn_id`/`signer_public_key` stay `Option` only because
/// nothing enforces the encoder's own invariants at this layer.
#[derive(Debug, Clone)]
pub(crate) struct GrantReplayRow {
    /// 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`, or `None` for a
    /// bare, un-tagged event.
    pub turn_id: Option<String>,
    /// The tool whose call the grant cleared.
    pub tool: String,
    /// The grant that cleared it: sha256-hex of its full signed payload.
    pub grant_ref: String,
    /// The capability names the grant kept against taint, as JSON-array
    /// text.
    pub covered_capabilities: String,
    /// The template-coverage hash the grant matched (`#618`).
    pub coverage_hash: String,
    /// [`SIGNATURE_VERIFIED`] or [`SIGNATURE_INVALID`].
    pub signature_status: String,
    /// Fleet-only at registration time: the embedded signer public key.
    pub signer_public_key: Option<Vec<u8>>,
}

/// The `grant_replays` typed table's full Arrow schema (every column,
/// including the Fleet-only `signer_public_key`) — what `grant_replays_raw`
/// registers as, for every scope. See the module docs' "Column selection and
/// redaction" section for nullability and the Fleet-only subset.
#[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("tool", DataType::Utf8, false),
        Field::new("grant_ref", DataType::Utf8, false),
        Field::new("covered_capabilities", DataType::Utf8, false),
        Field::new("coverage_hash", DataType::Utf8, false),
        Field::new("signature_status", DataType::Utf8, false),
        Field::new("signer_public_key", DataType::Binary, true),
    ]))
}

/// Decode already-framed [`GrantReplayRow`]s into the `grant_replays_raw`
/// table's Arrow `RecordBatch`, in [`schema`] order.
///
/// # Errors
///
/// Returns [`ArrowError`] if Arrow batch construction fails.
pub(crate) fn decode_grant_replays_batch(
    rows: &[GrantReplayRow],
) -> 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 tool_b = StringBuilder::with_capacity(rows.len(), rows.len() * 16);
    let mut grant_ref_b = StringBuilder::with_capacity(rows.len(), rows.len() * 16);
    let mut covered_capabilities_b = StringBuilder::with_capacity(rows.len(), rows.len() * 32);
    let mut coverage_hash_b = StringBuilder::with_capacity(rows.len(), rows.len() * 16);
    let mut signature_status_b = StringBuilder::with_capacity(rows.len(), rows.len() * 8);
    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(),
        }
        tool_b.append_value(&row.tool);
        grant_ref_b.append_value(&row.grant_ref);
        covered_capabilities_b.append_value(&row.covered_capabilities);
        coverage_hash_b.append_value(&row.coverage_hash);
        signature_status_b.append_value(&row.signature_status);
        match &row.signer_public_key {
            Some(v) => signer_public_key_b.append_value(v),
            None => signer_public_key_b.append_null(),
        }
    }

    let columns: Vec<ArrayRef> = vec![
        Arc::new(partition_b.finish()),
        Arc::new(position_b.finish()),
        Arc::new(turn_id_b.finish()),
        Arc::new(tool_b.finish()),
        Arc::new(grant_ref_b.finish()),
        Arc::new(covered_capabilities_b.finish()),
        Arc::new(coverage_hash_b.finish()),
        Arc::new(signature_status_b.finish()),
        Arc::new(signer_public_key_b.finish()),
    ];
    RecordBatch::try_new(schema(), columns)
}

/// Map [`polyc_facts::GrantReplaySignatureStatus`] onto this table's column
/// string — see the module docs for why the spelling matches `trace.rs`'s
/// own JSON literals.
const fn signature_status_str(status: polyc_facts::GrantReplaySignatureStatus) -> &'static str {
    match status {
        polyc_facts::GrantReplaySignatureStatus::Verified => SIGNATURE_VERIFIED,
        polyc_facts::GrantReplaySignatureStatus::Invalid => SIGNATURE_INVALID,
    }
}

/// Filter `partition`'s framed `events` to `grant_replay` rows, fold each
/// payload through the SHARED grant-replay fold
/// ([`polyc_facts::fold_grant_replay_event`]), and pair a successfully-folded
/// fact with that row's uniform key columns.
///
/// A payload too malformed to yield even the fold's minimal identity
/// (`tool`/`grant_ref`/`covered_capabilities`/`coverage_hash`) is skipped —
/// the module docs' "CRITICAL POSTURE" section is the important case NOT
/// covered by this sentence: a record that verification actively REJECTS is
/// still returned as a row, tagged
/// [`polyc_facts::GrantReplaySignatureStatus::Invalid`], not dropped.
#[must_use]
pub(crate) fn decode_grant_replays_events(
    partition: &str,
    events: &[(u64, Event)],
    trusted_signers: &[Vec<u8>],
) -> Vec<GrantReplayRow> {
    events
        .iter()
        .filter_map(|(position, event)| {
            let (_base, turn_id) = polyc_proto::kinds::parse(&event.kind);
            let turn_id = turn_id.map(|id| id.to_string());
            let fact = polyc_facts::fold_grant_replay_event(event, trusted_signers)?;
            let covered_capabilities =
                serde_json::to_string(&fact.covered_capabilities).unwrap_or_default();
            Some(GrantReplayRow {
                partition: partition.to_string(),
                position: *position,
                turn_id,
                tool: fact.tool,
                grant_ref: fact.grant_ref,
                covered_capabilities,
                coverage_hash: fact.coverage_hash,
                signature_status: signature_status_str(fact.signature_status).to_string(),
                signer_public_key: fact.signer_public_key,
            })
        })
        .collect()
}

#[cfg(test)]
mod tests {
    use arrow::array::Array as _;
    use polyc_crypto::approval::{ApprovalSigner, test_util::grant_replay_payload};
    use polyc_proto::kinds;
    use uuid::Uuid;

    use super::*;

    #[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",
                "tool",
                "grant_ref",
                "covered_capabilities",
                "coverage_hash",
                "signature_status",
                "signer_public_key",
            ]
        );

        let expect = [
            ("partition", DataType::Utf8, false),
            ("position", DataType::UInt64, false),
            ("turn_id", DataType::Utf8, true),
            ("tool", DataType::Utf8, false),
            ("grant_ref", DataType::Utf8, false),
            ("covered_capabilities", DataType::Utf8, false),
            ("coverage_hash", DataType::Utf8, false),
            ("signature_status", DataType::Utf8, false),
            ("signer_public_key", DataType::Binary, true),
        ];
        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);
        }
    }

    /// A real signed `grant_replay` decodes to one row (round trip through
    /// decode + Arrow batch).
    #[test]
    fn decode_round_trips_a_real_signed_grant_replay() {
        let turn = Uuid::from_u128(0x0195_abcd_ef01_2345_6789_abcd_ef01_5555);
        let signer = ApprovalSigner::from_seed(1);
        let (payload, _sig, _pk) = grant_replay_payload(
            "conv-rt",
            &turn.to_string(),
            "read_file",
            "grant-ref-rt",
            &["fs.read".to_owned(), "fs.list".to_owned()],
            "sha256:coverage-rt",
            &signer,
        );
        let events = vec![(
            1,
            Event::new(kinds::tagged(kinds::GRANT_REPLAY, &turn), payload),
        )];
        let trusted_signers = vec![signer.public_key_bytes()];

        let decoded = decode_grant_replays_events("conv-rt", &events, &trusted_signers);
        assert_eq!(decoded.len(), 1);

        let row = &decoded[0];
        assert_eq!(row.turn_id, Some(turn.to_string()));
        assert_eq!(row.tool, "read_file");
        assert_eq!(row.grant_ref, "grant-ref-rt");
        assert_eq!(row.covered_capabilities, r#"["fs.read","fs.list"]"#);
        assert_eq!(row.coverage_hash, "sha256:coverage-rt");
        assert_eq!(row.signature_status, "verified");
        assert_eq!(row.signer_public_key, Some(signer.public_key_bytes()));

        let batch = decode_grant_replays_batch(&decoded).expect("batch build");
        assert_eq!(batch.num_rows(), 1);
        assert_eq!(batch.schema(), schema());

        let tool = batch
            .column(3)
            .as_any()
            .downcast_ref::<arrow::array::StringArray>()
            .unwrap();
        assert_eq!(tool.value(0), "read_file");
    }

    /// Signature-status coverage — the contrast with `payments`' drop test:
    /// a trusted signer verifies, an untrusted (but internally consistent)
    /// signer reads `invalid` with the ROW STILL PRESENT, and only a
    /// structurally-malformed payload drops the row entirely.
    #[test]
    fn signature_status_trusted_signer_verifies_and_row_is_present() {
        let signer = ApprovalSigner::from_seed(2);
        let (payload, _sig, _pk) = grant_replay_payload(
            "conv-a",
            "turn-a",
            "read_file",
            "grant-ref-a",
            &["fs.read".to_owned()],
            "sha256:coverage-a",
            &signer,
        );
        let events = vec![(1, Event::new(kinds::GRANT_REPLAY.to_owned(), payload))];
        let trusted_signers = vec![signer.public_key_bytes()];
        let decoded = decode_grant_replays_events("conv-a", &events, &trusted_signers);
        assert_eq!(decoded.len(), 1);
        assert_eq!(decoded[0].signature_status, "verified");
    }

    /// An untrusted self-signed record is NOT dropped — unlike `payments`,
    /// the row stays present, tagged `invalid`, as the audit signal the
    /// `/grant-replays` endpoint exists to show.
    #[test]
    fn signature_status_untrusted_signer_is_invalid_but_row_stays_present() {
        let trusted = ApprovalSigner::from_seed(3);
        let untrusted = ApprovalSigner::from_seed(4);
        let (payload, _sig, _pk) = grant_replay_payload(
            "conv-b",
            "turn-b",
            "read_file",
            "grant-ref-b",
            &["fs.read".to_owned()],
            "sha256:coverage-b",
            &untrusted,
        );
        let events = vec![(1, Event::new(kinds::GRANT_REPLAY.to_owned(), payload))];
        let trusted_signers = vec![trusted.public_key_bytes()];
        let decoded = decode_grant_replays_events("conv-b", &events, &trusted_signers);
        assert_eq!(
            decoded.len(),
            1,
            "an untrusted-signer record must still surface as a row, unlike payments"
        );
        assert_eq!(decoded[0].signature_status, "invalid");
        assert_eq!(decoded[0].tool, "read_file", "claimed fields still shown");
    }

    /// Only a structurally-malformed payload (can't even yield `tool`/
    /// `grant_ref`/`covered_capabilities`/`coverage_hash`) drops the row.
    #[test]
    fn structurally_malformed_record_drops_the_row() {
        let events = vec![(
            1,
            Event::new(kinds::GRANT_REPLAY.to_owned(), vec![0xFF, 0xFE, 0xFD]),
        )];
        let decoded = decode_grant_replays_events("conv-c", &events, &[]);
        assert_eq!(decoded.len(), 0);
    }

    #[test]
    fn unrelated_kind_is_not_decoded_as_a_grant_replay() {
        let events = vec![(1, Event::new(kinds::USAGE.to_owned(), Vec::new()))];
        let decoded = decode_grant_replays_events("conv-d", &events, &[]);
        assert_eq!(decoded.len(), 0);
    }
}