polyc-query 2026.8.3

//! `approvals` typed-table decoder — the fact model's second FOLD-COUPLED
//! table (after [`crate::decode::payments`]): rows come from
//! [`polyc_facts::fold_approval_event`], the SAME fold
//! `forensics::classify_response_signature`/`parse_request_entry`/
//! `parse_response_entry` and `trace::response_signature_evidence` now read
//! through (`crates/facts/src/approvals.rs`), not a second,
//! independently-written decode of `approval_request`/`approval_response`
//! payloads.
//!
//! # One table, a `phase` column (mirrors `payments`' `direction`)
//!
//! `approval_request` and `approval_response` are two distinct signed-JSON
//! shapes that answer the same question — "what happened to this tool
//! call's approval" — so, like [`crate::decode::payments`]'s `direction`
//! column over `payment_receipt`/`outbound_payment_receipt`, they share one
//! table with a `phase` discriminator: `"request"` for a row folded from
//! [`polyc_facts::ApprovalFact::Request`], `"response"` for one folded from
//! [`polyc_facts::ApprovalFact::Response`]. Every request-only column is
//! `NULL` on a response row and vice versa — the shape [`crate::decode::message_content`]'s
//! `tool_calls.block_type` already establishes for a per-kind discriminator
//! with disjoint per-variant columns.
//!
//! # CRITICAL POSTURE: unlike `payments`, an unverified response is KEPT
//!
//! [`crate::decode::payments::decode_payments_events`] drops any payment
//! receipt whose signature does not verify — a receipt that doesn't verify
//! never counts. `approvals` does NOT drop an `approval_response` that fails
//! verification: `crates/facts/src/approvals.rs`'s module docs state the
//! reason plainly — the `/approvals` audit surface
//! (`crates/control-plane/src/forensics.rs`'s `collect_approvals`) exists to
//! show exactly this as an audit signal, a tampered or forged decision
//! tagged `INVALID` rather than hidden. [`decode_approvals_events`] therefore
//! keeps a row for every structurally-decodable `approval_response`
//! regardless of `signature_status`; only a payload too malformed to yield
//! even `request_id`+`approved` (the [`polyc_facts::fold_approval_event`]
//! contract) drops the row — see that function's own tests for the
//! contrast with `payments`' drop test, named explicitly to make the
//! difference visible.
//!
//! # `trusted_signers` (threaded exactly like `payments`)
//!
//! Reaches this module the identical way it reaches
//! [`crate::decode::payments::decode_payments_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.
//!
//! # Column selection and redaction
//!
//! NON-NULL for every row: `partition`, `position`, `phase`, `request_id`.
//! Nullable, populated by `phase`: `turn_id` (either phase, absent for a
//! bare, un-tagged event); `tool_name` (on BOTH phases — a request's own
//! gated tool, or a v2 response's own signed `tool_name` binding);
//! request side, otherwise — `args_json`, `request_reason`,
//! `request_sandbox_mode`; response side, otherwise — `approved`,
//! `response_reason`, `signature_status`, `signer_public_key`,
//! `modified_args_json`, `approved_for_session`, `caller`, `approver`,
//! `response_sandbox_mode`, `injected_context`, `routine_grant`,
//! `tool_descriptor_hash`, `grant_scope`. `conversation_id` (== the
//! `partition` column already) and `nonce` (the approval-response
//! single-use capability-token nonce, meaningless outside that one
//! consumption check) are deliberately omitted, matching `payments`'
//! rationale for dropping columns that duplicate the uniform key or carry
//! no query-facing meaning.
//!
//! Fleet-only (redacted out of the `approvals` view for every other scope,
//! `crate::views::APPROVALS_REDACTED_VIEW_SQL`): `request_reason`,
//! `request_sandbox_mode`, `signer_public_key`, `modified_args_json`,
//! `approved_for_session`, `caller`, `approver`, `response_sandbox_mode`,
//! `injected_context` — docs/reference/datafusion-data-layer.md's boundary
//! conditions put raw signer keys out of persona scope unconditionally, and
//! the remaining response fields in this list exceed `collect_approvals`'
//! own participant-visible set (it shows a participant only `tool_name`/
//! `args_json` on the request and `approved`/`reason`/signature status on
//! the response). Persona-visible: `partition`, `position`, `turn_id`,
//! `phase`, `request_id`, `tool_name`, `args_json`, `approved`,
//! `response_reason`, `signature_status`, `routine_grant`,
//! `tool_descriptor_hash`, `grant_scope` — the participant-visible set, plus
//! the three routine-grant marker columns: none of the three
//! carries an external identity or signer key, and an owner reading their
//! own routine's grant/revocation ledger needs exactly these to tell a
//! grant from a revocation and a per-tool grant from a blanket one — see
//! `crate::views::ROUTINE_GRANTS_VIEW_SQL`.

use std::sync::Arc;

use arrow::array::{ArrayRef, BinaryBuilder, BooleanBuilder, 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;

/// `phase`'s value for a row folded from [`polyc_facts::ApprovalFact::Request`].
const PHASE_REQUEST: &str = "request";

/// `phase`'s value for a row folded from [`polyc_facts::ApprovalFact::Response`].
const PHASE_RESPONSE: &str = "response";

/// [`polyc_facts::ApprovalSignatureStatus::Verified`]'s column string —
/// mirrors `trace.rs`'s own `"verified"`/`"invalid"`/`"legacy_unverifiable"`
/// literals, the machine-facing spelling of the same three outcomes
/// `forensics::SigStatus`'s `Display` impl renders for a human
/// (`"valid"`/`"INVALID"`/`"legacy (unverifiable)"`).
const SIGNATURE_VERIFIED: &str = "verified";
/// [`polyc_facts::ApprovalSignatureStatus::Invalid`]'s column string.
const SIGNATURE_INVALID: &str = "invalid";
/// [`polyc_facts::ApprovalSignatureStatus::LegacyUnverifiable`]'s column
/// string.
const SIGNATURE_LEGACY_UNVERIFIABLE: &str = "legacy_unverifiable";

/// One decoded `approvals` row — either half of the request/response pair,
/// discriminated by [`Self::phase`]. The fact model's uniform key columns
/// (`partition`, `position`, `turn_id`) plus every field
/// [`polyc_facts::ApprovalRequestFact`]/[`polyc_facts::ApprovalResponseFact`]
/// carries, `NULL` on whichever half does not apply to this row's phase.
#[derive(Debug, Clone)]
pub(crate) struct ApprovalRow {
    /// 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>,
    /// [`PHASE_REQUEST`] or [`PHASE_RESPONSE`].
    pub phase: String,
    /// The tool-call id this row's request/response is about.
    pub request_id: String,
    /// Request-only: the tool being gated.
    pub tool_name: Option<String>,
    /// Request-only: the proposed tool arguments, as JSON text.
    pub args_json: Option<String>,
    /// Request-only: the OVERRIDE explanation for why the call is gated.
    pub request_reason: Option<String>,
    /// Request-only: the sandbox/permission mode the harness ran under.
    pub request_sandbox_mode: Option<String>,
    /// Response-only: whether the request was approved.
    pub approved: Option<bool>,
    /// Response-only: free-form human-supplied reason.
    pub response_reason: Option<String>,
    /// Response-only: [`SIGNATURE_VERIFIED`]/[`SIGNATURE_INVALID`]/
    /// [`SIGNATURE_LEGACY_UNVERIFIABLE`].
    pub signature_status: Option<String>,
    /// Response-only, Fleet-only at registration time: the embedded signer
    /// public key.
    pub signer_public_key: Option<Vec<u8>>,
    /// Response-only, Fleet-only: the approver's edit to the proposed args.
    pub modified_args_json: Option<String>,
    /// Response-only, Fleet-only: whether the approval is remembered for
    /// the rest of the session.
    pub approved_for_session: Option<bool>,
    /// Response-only, Fleet-only: the caller identity the approval is
    /// scoped to.
    pub caller: Option<String>,
    /// Response-only, Fleet-only: the identity that actually resolved this
    /// decision (`#1025`).
    pub approver: Option<String>,
    /// Response-only, Fleet-only: the sandbox/permission mode the paused
    /// turn ran under.
    pub response_sandbox_mode: Option<String>,
    /// Response-only, Fleet-only: context the approver attached to inject
    /// before the tool runs.
    pub injected_context: Option<String>,
    /// Response-only, persona-visible: whether this response
    /// mints or revokes a routine tool grant.
    pub routine_grant: Option<bool>,
    /// Response-only, persona-visible: the granted tool's
    /// descriptor hash — empty when [`Self::routine_grant`] is not
    /// `Some(true)`.
    pub tool_descriptor_hash: Option<String>,
    /// Response-only, persona-visible: the grant's scope
    /// (`"tool"`/`"blanket_below_high"`/`"blanket_all"`) — empty when
    /// [`Self::routine_grant`] is not `Some(true)`.
    pub grant_scope: Option<String>,
}

/// The `approvals` typed table's full Arrow schema (every column, including
/// the Fleet-only ones) — what `approvals_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("phase", DataType::Utf8, false),
        Field::new("request_id", DataType::Utf8, false),
        Field::new("tool_name", DataType::Utf8, true),
        Field::new("args_json", DataType::Utf8, true),
        Field::new("request_reason", DataType::Utf8, true),
        Field::new("request_sandbox_mode", DataType::Utf8, true),
        Field::new("approved", DataType::Boolean, true),
        Field::new("response_reason", DataType::Utf8, true),
        Field::new("signature_status", DataType::Utf8, true),
        Field::new("signer_public_key", DataType::Binary, true),
        Field::new("modified_args_json", DataType::Utf8, true),
        Field::new("approved_for_session", DataType::Boolean, true),
        Field::new("caller", DataType::Utf8, true),
        Field::new("approver", DataType::Utf8, true),
        Field::new("response_sandbox_mode", DataType::Utf8, true),
        Field::new("injected_context", DataType::Utf8, true),
        Field::new("routine_grant", DataType::Boolean, true),
        Field::new("tool_descriptor_hash", DataType::Utf8, true),
        Field::new("grant_scope", DataType::Utf8, true),
    ]))
}

/// Decode already-framed [`ApprovalRow`]s into the `approvals_raw` table's
/// Arrow `RecordBatch`, in [`schema`] order.
///
/// # Errors
///
/// Returns [`ArrowError`] if Arrow batch construction fails.
// 19 columns' worth of builder wiring, the widest typed table this crate
// has decoded so far — one mechanical `match ... { Some(v) => append_value,
// None => append_null }` arm per column, not a shape a helper extraction
// would meaningfully shorten (mirrors the same call this crate's other
// wide decoders make, e.g. `payments`' own 16-column version, which sits
// just under the threshold this one crosses).
// `approved_b`/`approver_b` are both real, necessarily similar column names
// (`approved`: the decision; `approver`: who resolved it, #1025) — renaming
// either to silence the lint would make the pair harder to tell apart, not
// easier.
#[allow(clippy::too_many_lines, clippy::similar_names)]
pub(crate) fn decode_approvals_batch(rows: &[ApprovalRow]) -> 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 phase_b = StringBuilder::with_capacity(rows.len(), rows.len() * 8);
    let mut request_id_b = StringBuilder::with_capacity(rows.len(), rows.len() * 16);
    let mut tool_name_b = StringBuilder::with_capacity(rows.len(), rows.len() * 16);
    let mut args_json_b = StringBuilder::with_capacity(rows.len(), rows.len() * 32);
    let mut request_reason_b = StringBuilder::with_capacity(rows.len(), rows.len() * 16);
    let mut request_sandbox_mode_b = StringBuilder::with_capacity(rows.len(), rows.len() * 8);
    let mut approved_b = BooleanBuilder::with_capacity(rows.len());
    let mut response_reason_b = StringBuilder::with_capacity(rows.len(), rows.len() * 16);
    let mut signature_status_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 modified_args_json_b = StringBuilder::with_capacity(rows.len(), rows.len() * 32);
    let mut approved_for_session_b = BooleanBuilder::with_capacity(rows.len());
    let mut caller_b = StringBuilder::with_capacity(rows.len(), rows.len() * 16);
    let mut approver_b = StringBuilder::with_capacity(rows.len(), rows.len() * 16);
    let mut response_sandbox_mode_b = StringBuilder::with_capacity(rows.len(), rows.len() * 8);
    let mut injected_context_b = StringBuilder::with_capacity(rows.len(), rows.len() * 16);
    let mut routine_grant_b = BooleanBuilder::with_capacity(rows.len());
    let mut tool_descriptor_hash_b = StringBuilder::with_capacity(rows.len(), rows.len() * 16);
    let mut grant_scope_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(),
        }
        phase_b.append_value(&row.phase);
        request_id_b.append_value(&row.request_id);
        match &row.tool_name {
            Some(v) => tool_name_b.append_value(v),
            None => tool_name_b.append_null(),
        }
        match &row.args_json {
            Some(v) => args_json_b.append_value(v),
            None => args_json_b.append_null(),
        }
        match &row.request_reason {
            Some(v) => request_reason_b.append_value(v),
            None => request_reason_b.append_null(),
        }
        match &row.request_sandbox_mode {
            Some(v) => request_sandbox_mode_b.append_value(v),
            None => request_sandbox_mode_b.append_null(),
        }
        match row.approved {
            Some(v) => approved_b.append_value(v),
            None => approved_b.append_null(),
        }
        match &row.response_reason {
            Some(v) => response_reason_b.append_value(v),
            None => response_reason_b.append_null(),
        }
        match &row.signature_status {
            Some(v) => signature_status_b.append_value(v),
            None => signature_status_b.append_null(),
        }
        match &row.signer_public_key {
            Some(v) => signer_public_key_b.append_value(v),
            None => signer_public_key_b.append_null(),
        }
        match &row.modified_args_json {
            Some(v) => modified_args_json_b.append_value(v),
            None => modified_args_json_b.append_null(),
        }
        match row.approved_for_session {
            Some(v) => approved_for_session_b.append_value(v),
            None => approved_for_session_b.append_null(),
        }
        match &row.caller {
            Some(v) => caller_b.append_value(v),
            None => caller_b.append_null(),
        }
        match &row.approver {
            Some(v) => approver_b.append_value(v),
            None => approver_b.append_null(),
        }
        match &row.response_sandbox_mode {
            Some(v) => response_sandbox_mode_b.append_value(v),
            None => response_sandbox_mode_b.append_null(),
        }
        match &row.injected_context {
            Some(v) => injected_context_b.append_value(v),
            None => injected_context_b.append_null(),
        }
        match row.routine_grant {
            Some(v) => routine_grant_b.append_value(v),
            None => routine_grant_b.append_null(),
        }
        match &row.tool_descriptor_hash {
            Some(v) => tool_descriptor_hash_b.append_value(v),
            None => tool_descriptor_hash_b.append_null(),
        }
        match &row.grant_scope {
            Some(v) => grant_scope_b.append_value(v),
            None => grant_scope_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(phase_b.finish()),
        Arc::new(request_id_b.finish()),
        Arc::new(tool_name_b.finish()),
        Arc::new(args_json_b.finish()),
        Arc::new(request_reason_b.finish()),
        Arc::new(request_sandbox_mode_b.finish()),
        Arc::new(approved_b.finish()),
        Arc::new(response_reason_b.finish()),
        Arc::new(signature_status_b.finish()),
        Arc::new(signer_public_key_b.finish()),
        Arc::new(modified_args_json_b.finish()),
        Arc::new(approved_for_session_b.finish()),
        Arc::new(caller_b.finish()),
        Arc::new(approver_b.finish()),
        Arc::new(response_sandbox_mode_b.finish()),
        Arc::new(injected_context_b.finish()),
        Arc::new(routine_grant_b.finish()),
        Arc::new(tool_descriptor_hash_b.finish()),
        Arc::new(grant_scope_b.finish()),
    ];
    RecordBatch::try_new(schema(), columns)
}

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

/// Filter `partition`'s framed `events` to `approval_request`/
/// `approval_response` rows, fold each payload through the SHARED approval
/// fold ([`polyc_facts::fold_approval_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
/// (`request_id` for either phase, plus `approved` for a response) is
/// skipped — the module docs' "CRITICAL POSTURE" section is the important
/// case NOT covered by this sentence: a response that verification actively
/// REJECTS is still returned as a row, tagged
/// [`polyc_facts::ApprovalSignatureStatus::Invalid`], not dropped.
#[must_use]
pub(crate) fn decode_approvals_events(
    partition: &str,
    events: &[(u64, Event)],
    trusted_signers: &[Vec<u8>],
) -> Vec<ApprovalRow> {
    events
        .iter()
        .filter_map(|(position, event)| {
            let (_base, turn_id) = kinds::parse(&event.kind);
            let turn_id = turn_id.map(|id| id.to_string());
            match polyc_facts::fold_approval_event(event, trusted_signers)? {
                polyc_facts::ApprovalFact::Request(req) => Some(ApprovalRow {
                    partition: partition.to_string(),
                    position: *position,
                    turn_id,
                    phase: PHASE_REQUEST.to_string(),
                    request_id: req.request_id,
                    tool_name: Some(req.tool_name),
                    args_json: Some(req.args_json),
                    request_reason: Some(req.reason),
                    request_sandbox_mode: Some(req.sandbox_mode),
                    approved: None,
                    response_reason: None,
                    signature_status: None,
                    signer_public_key: None,
                    modified_args_json: None,
                    approved_for_session: None,
                    caller: None,
                    approver: None,
                    response_sandbox_mode: None,
                    injected_context: None,
                    routine_grant: None,
                    tool_descriptor_hash: None,
                    grant_scope: None,
                }),
                polyc_facts::ApprovalFact::Response(resp) => Some(ApprovalRow {
                    partition: partition.to_string(),
                    position: *position,
                    turn_id,
                    phase: PHASE_RESPONSE.to_string(),
                    request_id: resp.request_id,
                    tool_name: resp.tool_name,
                    args_json: None,
                    request_reason: None,
                    request_sandbox_mode: None,
                    approved: Some(resp.approved),
                    response_reason: resp.response_reason,
                    signature_status: Some(signature_status_str(resp.signature_status).to_string()),
                    signer_public_key: resp.signer_public_key,
                    modified_args_json: resp.modified_args_json,
                    approved_for_session: resp.approved_for_session,
                    caller: resp.caller,
                    approver: resp.approver,
                    response_sandbox_mode: resp.sandbox_mode,
                    injected_context: resp.injected_context,
                    routine_grant: resp.routine_grant,
                    tool_descriptor_hash: resp.tool_descriptor_hash,
                    grant_scope: resp.grant_scope,
                }),
            }
        })
        .collect()
}

#[cfg(test)]
mod tests {
    use arrow::array::Array as _;
    use polyc_crypto::approval::{ApprovalSigner, request_payload, response_payload};
    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",
                "phase",
                "request_id",
                "tool_name",
                "args_json",
                "request_reason",
                "request_sandbox_mode",
                "approved",
                "response_reason",
                "signature_status",
                "signer_public_key",
                "modified_args_json",
                "approved_for_session",
                "caller",
                "approver",
                "response_sandbox_mode",
                "injected_context",
                "routine_grant",
                "tool_descriptor_hash",
                "grant_scope",
            ]
        );

        let expect = [
            ("partition", DataType::Utf8, false),
            ("position", DataType::UInt64, false),
            ("turn_id", DataType::Utf8, true),
            ("phase", DataType::Utf8, false),
            ("request_id", DataType::Utf8, false),
            ("tool_name", DataType::Utf8, true),
            ("args_json", DataType::Utf8, true),
            ("request_reason", DataType::Utf8, true),
            ("request_sandbox_mode", DataType::Utf8, true),
            ("approved", DataType::Boolean, true),
            ("response_reason", DataType::Utf8, true),
            ("signature_status", DataType::Utf8, true),
            ("signer_public_key", DataType::Binary, true),
            ("modified_args_json", DataType::Utf8, true),
            ("approved_for_session", DataType::Boolean, true),
            ("caller", DataType::Utf8, true),
            ("approver", DataType::Utf8, true),
            ("response_sandbox_mode", DataType::Utf8, true),
            ("injected_context", DataType::Utf8, true),
            ("routine_grant", DataType::Boolean, true),
            ("tool_descriptor_hash", DataType::Utf8, true),
            ("grant_scope", DataType::Utf8, 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 request+response pair decodes to two rows (round trip
    /// through decode + Arrow batch), the request row carrying its own
    /// fields and every response-only column `NULL`, the response row the
    /// mirror image.
    #[test]
    fn decode_round_trips_a_real_signed_request_and_response_pair() {
        let turn = Uuid::from_u128(0x0195_abcd_ef01_2345_6789_abcd_ef01_4444);
        let signer = ApprovalSigner::from_seed(1);
        let request_bytes = request_payload(
            "call-1",
            "rm",
            r#"{"path":"/tmp"}"#,
            "default",
            "",
            &[],
            "",
            "",
            "",
            &[],
            false,
        );
        let response_bytes = response_payload(
            "call-1",
            "rm",
            r#"{"path":"/tmp"}"#,
            "",
            true,
            false,
            &[],
            "caller-1",
            "",
            "default",
            "ok",
            "",
            "conv-rt",
            "nonce-1",
            &turn.to_string(),
            &signer,
        )
        .0;
        let events = vec![
            (
                1,
                Event::new(kinds::tagged(kinds::APPROVAL_REQUEST, &turn), request_bytes),
            ),
            (
                2,
                Event::new(
                    kinds::tagged(kinds::APPROVAL_RESPONSE, &turn),
                    response_bytes,
                ),
            ),
        ];
        let trusted_signers = vec![signer.public_key_bytes()];

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

        let req = &decoded[0];
        assert_eq!(req.phase, "request");
        assert_eq!(req.request_id, "call-1");
        assert_eq!(req.turn_id, Some(turn.to_string()));
        assert_eq!(req.tool_name.as_deref(), Some("rm"));
        assert_eq!(req.args_json.as_deref(), Some(r#"{"path":"/tmp"}"#));
        assert_eq!(req.approved, None);
        assert_eq!(req.signature_status, None);

        let resp = &decoded[1];
        assert_eq!(resp.phase, "response");
        assert_eq!(resp.request_id, "call-1");
        assert_eq!(resp.approved, Some(true));
        assert_eq!(resp.response_reason.as_deref(), Some("ok"));
        assert_eq!(resp.signature_status.as_deref(), Some("verified"));
        assert_eq!(resp.signer_public_key, Some(signer.public_key_bytes()));
        assert_eq!(resp.caller.as_deref(), Some("caller-1"));
        assert_eq!(
            resp.tool_name.as_deref(),
            Some("rm"),
            "a v2 response's own signed tool_name binding populates the shared column"
        );
        assert_eq!(resp.routine_grant, Some(false));
        assert_eq!(resp.tool_descriptor_hash.as_deref(), Some(""));
        assert_eq!(resp.grant_scope.as_deref(), Some(""));

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

        let phase = batch
            .column(3)
            .as_any()
            .downcast_ref::<arrow::array::StringArray>()
            .unwrap();
        assert_eq!(phase.value(0), "request");
        assert_eq!(phase.value(1), "response");

        let tool_name = batch
            .column(5)
            .as_any()
            .downcast_ref::<arrow::array::StringArray>()
            .unwrap();
        assert!(tool_name.is_valid(0));
        assert!(
            tool_name.is_valid(1),
            "a v2 response's own signed tool_name binding populates the shared column"
        );
        assert_eq!(tool_name.value(1), "rm");
    }

    /// Signature-status coverage — the contrast with `payments`' drop test
    /// (`decode_payments_events_drops_a_receipt_from_an_untrusted_signer`):
    /// a trusted signer verifies, an untrusted (but internally consistent)
    /// signer reads `Invalid` with the ROW STILL PRESENT, a legacy v1 shape
    /// reads `LegacyUnverifiable`, 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 bytes = response_payload(
            "call-2",
            "rm",
            "{}",
            "",
            true,
            false,
            &[],
            "caller-2",
            "",
            "",
            "ok",
            "",
            "conv-a",
            "n1",
            "",
            &signer,
        )
        .0;
        let events = vec![(1, Event::new(kinds::APPROVAL_RESPONSE.to_owned(), bytes))];
        let trusted_signers = vec![signer.public_key_bytes()];
        let decoded = decode_approvals_events("conv-a", &events, &trusted_signers);
        assert_eq!(decoded.len(), 1);
        assert_eq!(decoded[0].signature_status.as_deref(), Some("verified"));
    }

    /// An untrusted self-signed response is NOT dropped — unlike `payments`,
    /// the row stays present, tagged `invalid`, as the audit signal the
    /// `/approvals` 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 bytes = response_payload(
            "call-3",
            "rm",
            "{}",
            "",
            true,
            false,
            &[],
            "caller-3",
            "",
            "",
            "ok",
            "",
            "conv-b",
            "n1",
            "",
            &untrusted,
        )
        .0;
        let events = vec![(1, Event::new(kinds::APPROVAL_RESPONSE.to_owned(), bytes))];
        let trusted_signers = vec![trusted.public_key_bytes()];
        let decoded = decode_approvals_events("conv-b", &events, &trusted_signers);
        assert_eq!(
            decoded.len(),
            1,
            "an untrusted-signer response must still surface as a row, unlike payments"
        );
        assert_eq!(decoded[0].signature_status.as_deref(), Some("invalid"));
        assert_eq!(
            decoded[0].approved,
            Some(true),
            "claimed fields still shown"
        );
    }

    /// A legacy (pre-binding, call-id-only) response reads
    /// `legacy_unverifiable`, not `invalid` — not a tamper signal.
    #[test]
    fn signature_status_legacy_shape_is_unverifiable_not_invalid() {
        let legacy = br#"{"request_id":"call-4","approved":true,"reason":"ok"}"#.to_vec();
        let events = vec![(1, Event::new(kinds::APPROVAL_RESPONSE.to_owned(), legacy))];
        let decoded = decode_approvals_events("conv-c", &events, &[]);
        assert_eq!(decoded.len(), 1);
        assert_eq!(
            decoded[0].signature_status.as_deref(),
            Some("legacy_unverifiable")
        );
        assert_eq!(decoded[0].response_reason, None);
    }

    /// Only a structurally-malformed payload (can't even yield a
    /// `request_id`) drops the row.
    #[test]
    fn structurally_malformed_response_drops_the_row() {
        let events = vec![(
            1,
            Event::new(kinds::APPROVAL_RESPONSE.to_owned(), vec![0xFF, 0xFE, 0xFD]),
        )];
        let decoded = decode_approvals_events("conv-d", &events, &[]);
        assert_eq!(decoded.len(), 0);
    }

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

    /// A `routine_grant`-marked response (POLY-25) decodes its
    /// marker, descriptor hash, and scope into the shared `approvals` table
    /// — the read surface this issue adds, over the write shape POLY-25/
    /// POLY-32 already ship.
    #[test]
    fn decode_reads_a_routine_grant_response() {
        let signer = ApprovalSigner::from_seed(42);
        let (payload, ..) = polyc_crypto::approval::routine_grant_payload(
            "call-9",
            "fs_write",
            "{}",
            "",
            true,
            "owner-1",
            "",
            "default",
            "",
            &[],
            "conv-fire-9",
            "nonce-9",
            "",
            "hash-xyz",
            "blanket_below_high",
            &signer,
        );
        let events = vec![(1, Event::new(kinds::APPROVAL_RESPONSE.to_owned(), payload))];
        let trusted_signers = vec![signer.public_key_bytes()];

        let decoded = decode_approvals_events("conv-fire-9", &events, &trusted_signers);
        assert_eq!(decoded.len(), 1);
        assert_eq!(decoded[0].tool_name.as_deref(), Some("fs_write"));
        assert_eq!(decoded[0].routine_grant, Some(true));
        assert_eq!(decoded[0].tool_descriptor_hash.as_deref(), Some("hash-xyz"));
        assert_eq!(
            decoded[0].grant_scope.as_deref(),
            Some("blanket_below_high")
        );

        let batch = decode_approvals_batch(&decoded).expect("batch build");
        let routine_grant = batch
            .column(19)
            .as_any()
            .downcast_ref::<arrow::array::BooleanArray>()
            .unwrap();
        assert!(routine_grant.value(0));
        let grant_scope = batch
            .column(21)
            .as_any()
            .downcast_ref::<arrow::array::StringArray>()
            .unwrap();
        assert_eq!(grant_scope.value(0), "blanket_below_high");
    }
}