polyc-query 2026.8.3

//! `handoffs` typed-table decoder — the fact model's eighth typed table.
//!
//! # Shared decode (#1579)
//!
//! Query and trace now share `polyc_facts::fold_handoff_event`. This module
//! keeps only query concerns. These include `signed_by` redaction and Arrow
//! column shaping.
//!
//! # `signature_status` is a deployment-pinned verdict (`#1124`)
//!
//! The fold uses this deployment's current and retired handoff keys.
//! [`decode_handoffs_events`] receives that trust set. The verdict is
//! `verified`, `invalid`, or `untrusted`. A valid signature from an unknown key
//! is untrusted.
//!
//! # Tampered and untrusted events stay visible
//!
//! This matches [`crate::decode::approvals`]. Forged, tampered, and
//! foreign-signed events remain visible. Only malformed payloads drop a row.
//!
//! # One table with two phases
//!
//! Both wire shapes describe a requested transfer. They share one table with a
//! `phase` column, like approvals and payments. Phase-specific columns are
//! `NULL` on the other phase.
//!
//! # Column selection
//!
//! `reason` is shared by the `handoff` phase (`Handoff::reason`) and the
//! `handoff_denied` phase (`HandoffDenied::reason` — the freeform reason the
//! model gave for the requested transfer, carried over from the request onto
//! the denial) — the same "one column, two kinds" shape `direction`/`phase`
//! already establish elsewhere in this crate, not a coincidence of naming:
//! both proto fields answer the identical question, "why was this transfer
//! requested". `child_agent_id` is likewise shared by `handoff`
//! (`Handoff::child_agent_id`, the agent that ran) and `handoff_denied`
//! (`HandoffDenied::child_agent_id`, the agent the model asked for and was
//! refused). `HandoffDenied::parent_conversation_id` is deliberately omitted
//! — a denial is always recorded on the PARENT's own partition, so that
//! field is already the `partition` column under another name, the identical
//! rationale [`crate::decode::approvals`]'s module docs give for omitting
//! `conversation_id`. `Handoff::carried_context` (the full carried
//! `Message` list) is also omitted: `collect_handoffs` never surfaces it,
//! only `carried_count` — the same "expose what the existing consumer
//! actually reads, not everything the wire type carries" discipline
//! [`crate::decode::payments`] applies to `VerifiedReceipt::kind`.
//! `allowed` (`HandoffDenied::allowed`, the parent's declared
//! `canHandoffTo` allowlist at request time) is stored as `Utf8` JSON-array
//! text, matching every other JSON-shaped string column this crate exposes
//! (`tool_calls.arguments`/`.result`) rather than a native Arrow list type.
//!
//! # Redaction: only `signed_by` is Fleet-only
//!
//! `signed_by` is a raw public key, so only Fleet queries expose it. Every
//! scope sees `signature_status` and the event fields. [`crate::engine`]
//! builds the scope-specific view from `handoffs_raw`. It then removes the raw
//! table from non-Fleet scopes.

use std::sync::Arc;

use arrow::array::{ArrayRef, BinaryBuilder, StringBuilder, UInt32Builder, UInt64Builder};
use arrow::datatypes::{DataType, Field, Schema, SchemaRef};
use arrow::error::ArrowError;
use arrow::record_batch::RecordBatch;
use polyc_crypto::signing_role::{HandoffRole, RoleTrustSet};
use polyc_eventlog::Event;
use polyc_proto::kinds;

/// `phase`'s value for a row folded from [`kinds::HANDOFF`].
const PHASE_HANDOFF: &str = "handoff";

/// `phase`'s value for a row folded from [`kinds::HANDOFF_DENIED`].
const PHASE_HANDOFF_DENIED: &str = "handoff_denied";

/// One decoded `handoffs` row, discriminated by [`Self::phase`]. It contains
/// the uniform keys and both phases' fields. Fields from the other phase are
/// `NULL`.
#[derive(Debug, Clone)]
pub(crate) struct HandoffRow {
    /// The parent journal partition that contains this row (`conv-{id}`). See
    /// the module's column-selection section.
    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_HANDOFF`] or [`PHASE_HANDOFF_DENIED`].
    pub phase: String,
    /// `handoff`-only: the child conversation this transfer
    /// names (the id becomes the child's own partition suffix once the
    /// controller materializes it).
    pub child_conversation_id: Option<String>,
    /// `handoff`/`handoff_denied`-only: the child agent identifier the
    /// transfer named.
    pub child_agent_id: Option<String>,
    /// `handoff`-only: the number of parent messages carried into the child.
    /// The table does not expose the messages themselves.
    pub carried_count: Option<u32>,
    /// `handoff`/`handoff_denied`-only: the freeform reason given for the
    /// requested transfer.
    pub reason: Option<String>,
    /// `handoff_denied`-only: the parent's bound `Agent` resource name at
    /// request time.
    pub parent_agent_id: Option<String>,
    /// `handoff_denied`-only: plain-language explanation of why the
    /// transfer was refused.
    pub denial_reason: Option<String>,
    /// `handoff_denied`-only: the parent's declared `canHandoffTo` allowlist
    /// at request time, as JSON-array text.
    pub allowed: Option<String>,
    /// Every phase, Fleet-only at registration time (see the module docs'
    /// "Redaction" section): the embedded ed25519 public key the event
    /// claims to be signed by.
    pub signed_by: Option<Vec<u8>>,
    /// Every phase: how the signature reads against this deployment's
    /// handoff-role trust set — `verified`, `invalid`, or `untrusted` (see
    /// the module docs). Every surviving handoff phase is signed, so this
    /// column is never `NULL`.
    pub signature_status: String,
}

/// The `handoffs` typed table's full Arrow schema (every column, including
/// `signed_by`) — what `handoffs_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("phase", DataType::Utf8, false),
        Field::new("child_conversation_id", DataType::Utf8, true),
        Field::new("child_agent_id", DataType::Utf8, true),
        Field::new("carried_count", DataType::UInt32, true),
        Field::new("reason", DataType::Utf8, true),
        Field::new("parent_agent_id", DataType::Utf8, true),
        Field::new("denial_reason", DataType::Utf8, true),
        Field::new("allowed", DataType::Utf8, true),
        Field::new("signed_by", DataType::Binary, true),
        Field::new("signature_status", DataType::Utf8, false),
    ]))
}

/// Decode already-framed [`HandoffRow`]s into the `handoffs_raw` table's
/// Arrow `RecordBatch`, in [`schema`] order.
///
/// # Errors
///
/// Returns [`ArrowError`] if Arrow batch construction fails.
#[allow(clippy::too_many_lines)] // 13 columns' worth of mechanical builder wiring, see `approvals`' own precedent
pub(crate) fn decode_handoffs_batch(rows: &[HandoffRow]) -> 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() * 16);
    let mut child_conversation_id_b = StringBuilder::with_capacity(rows.len(), rows.len() * 16);
    let mut child_agent_id_b = StringBuilder::with_capacity(rows.len(), rows.len() * 16);
    let mut carried_count_b = UInt32Builder::with_capacity(rows.len());
    let mut reason_b = StringBuilder::with_capacity(rows.len(), rows.len() * 16);
    let mut parent_agent_id_b = StringBuilder::with_capacity(rows.len(), rows.len() * 16);
    let mut denial_reason_b = StringBuilder::with_capacity(rows.len(), rows.len() * 16);
    let mut allowed_b = StringBuilder::with_capacity(rows.len(), rows.len() * 16);
    let mut signed_by_b = BinaryBuilder::with_capacity(rows.len(), rows.len() * 32);
    let mut signature_status_b = StringBuilder::with_capacity(rows.len(), rows.len() * 10);

    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);
        match &row.child_conversation_id {
            Some(v) => child_conversation_id_b.append_value(v),
            None => child_conversation_id_b.append_null(),
        }
        match &row.child_agent_id {
            Some(v) => child_agent_id_b.append_value(v),
            None => child_agent_id_b.append_null(),
        }
        match row.carried_count {
            Some(v) => carried_count_b.append_value(v),
            None => carried_count_b.append_null(),
        }
        match &row.reason {
            Some(v) => reason_b.append_value(v),
            None => reason_b.append_null(),
        }
        match &row.parent_agent_id {
            Some(v) => parent_agent_id_b.append_value(v),
            None => parent_agent_id_b.append_null(),
        }
        match &row.denial_reason {
            Some(v) => denial_reason_b.append_value(v),
            None => denial_reason_b.append_null(),
        }
        match &row.allowed {
            Some(v) => allowed_b.append_value(v),
            None => allowed_b.append_null(),
        }
        match &row.signed_by {
            Some(v) => signed_by_b.append_value(v),
            None => signed_by_b.append_null(),
        }
        signature_status_b.append_value(&row.signature_status);
    }

    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(child_conversation_id_b.finish()),
        Arc::new(child_agent_id_b.finish()),
        Arc::new(carried_count_b.finish()),
        Arc::new(reason_b.finish()),
        Arc::new(parent_agent_id_b.finish()),
        Arc::new(denial_reason_b.finish()),
        Arc::new(allowed_b.finish()),
        Arc::new(signed_by_b.finish()),
        Arc::new(signature_status_b.finish()),
    ];
    RecordBatch::try_new(schema(), columns)
}

/// Filters one partition to handoff-family rows. It decodes each payload with
/// [`polyc_facts::fold_handoff_event`] and adds the uniform key columns.
///
/// A malformed payload is skipped and logged. A decodable payload with a bad
/// or untrusted signature remains visible with its verdict. The fold
/// enforces this posture. This function shapes the result into
/// [`HandoffRow`]'s disjoint columns. It renders `allowed` as JSON text.
/// View registration applies the `signed_by` redaction downstream.
#[must_use]
pub(crate) fn decode_handoffs_events(
    partition: &str,
    events: &[(u64, Event)],
    handoff_trust: &RoleTrustSet<HandoffRole>,
) -> Vec<HandoffRow> {
    events
        .iter()
        .filter_map(|(position, event)| {
            let (base, turn_id) = kinds::parse(&event.kind);
            let turn_id = turn_id.map(|id| id.to_string());
            let Some(fact) = polyc_facts::fold_handoff_event(base, &event.payload, handoff_trust)
            else {
                if base == kinds::HANDOFF || base == kinds::HANDOFF_DENIED {
                    warn_corrupt_payload(*position, event.payload.len());
                }
                return None;
            };
            Some(match fact {
                polyc_facts::HandoffFact::Handoff(h) => HandoffRow {
                    partition: partition.to_string(),
                    position: *position,
                    turn_id,
                    phase: PHASE_HANDOFF.to_string(),
                    child_conversation_id: Some(h.child_conversation_id),
                    child_agent_id: Some(h.child_agent_id),
                    carried_count: Some(h.carried_count),
                    reason: Some(h.reason),
                    parent_agent_id: None,
                    denial_reason: None,
                    allowed: None,
                    signed_by: Some(h.signed_by),
                    signature_status: h.signature_status.as_str().to_owned(),
                },
                polyc_facts::HandoffFact::Denied(d) => {
                    let allowed = serde_json::to_string(&d.allowed).unwrap_or_default();
                    HandoffRow {
                        partition: partition.to_string(),
                        position: *position,
                        turn_id,
                        phase: PHASE_HANDOFF_DENIED.to_string(),
                        child_conversation_id: None,
                        child_agent_id: Some(d.child_agent_id),
                        carried_count: None,
                        reason: Some(d.reason),
                        parent_agent_id: Some(d.parent_agent_id),
                        denial_reason: Some(d.denial_reason),
                        allowed: Some(allowed),
                        signed_by: Some(d.signed_by),
                        signature_status: d.signature_status.as_str().to_owned(),
                    }
                }
            })
        })
        .collect()
}

/// Log a skipped, structurally-undecodable `handoffs` payload — the same
/// `tracing::warn!` shape `crate::decode::decode_typed_kind_events` uses for
/// a corrupt non-empty payload, reproduced here because this table's
/// two-way phase dispatch cannot reuse that shared helper (see the module
/// docs for why).
fn warn_corrupt_payload(position: u64, len: usize) {
    tracing::warn!(
        len,
        table = "handoffs",
        position,
        "corrupt event payload; skipping row"
    );
}

#[cfg(test)]
mod tests {
    use arrow::array::Array as _;
    use buffa::Message as _;
    use polyc_crypto::signing_role::HandoffSigner;
    use polyc_proto::proto::polychrome::agent::v1::{Content, Message, TextContent, content};
    use polyc_proto::proto::polychrome::handoff::v1::{Handoff, HandoffDenied};
    use uuid::Uuid;

    use super::*;

    /// The deployment trust set that holds exactly `signer`'s key.
    fn trust(signer: &HandoffSigner) -> RoleTrustSet<HandoffRole> {
        RoleTrustSet::current(signer)
    }

    fn text_msg(role: &str, text: &str) -> Message {
        Message {
            role: role.to_owned(),
            content: buffa::MessageField::some(Content {
                r#type: Some(content::Type::Text(Box::new(TextContent {
                    text: text.to_owned(),
                    ..Default::default()
                }))),
                ..Default::default()
            }),
            internal_only: false,
            ..Default::default()
        }
    }

    fn signed_handoff(signer: &HandoffSigner, child_conversation_id: &str) -> Handoff {
        let mut h = Handoff {
            child_conversation_id: child_conversation_id.to_owned(),
            child_agent_id: "researcher".to_owned(),
            carried_count: 2,
            carried_context: vec![text_msg("user", "find prior art")],
            reason: "delegate research".to_owned(),
            ..Default::default()
        };
        polyc_crypto::handoff::sign_handoff_into(signer, &mut h);
        h
    }

    fn signed_denied(signer: &HandoffSigner) -> HandoffDenied {
        let mut d = HandoffDenied {
            parent_conversation_id: "parent-7".to_owned(),
            parent_agent_id: "assistant".to_owned(),
            child_agent_id: "banned-agent".to_owned(),
            reason: "delegate weird task".to_owned(),
            denial_reason: "this agent can't hand off to that agent".to_owned(),
            allowed: vec!["coding".to_owned(), "research".to_owned()],
            ..Default::default()
        };
        polyc_crypto::handoff::sign_handoff_denied_into(signer, &mut d);
        d
    }

    #[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",
                "child_conversation_id",
                "child_agent_id",
                "carried_count",
                "reason",
                "parent_agent_id",
                "denial_reason",
                "allowed",
                "signed_by",
                "signature_status",
            ]
        );

        let expect = [
            ("partition", DataType::Utf8, false),
            ("position", DataType::UInt64, false),
            ("turn_id", DataType::Utf8, true),
            ("phase", DataType::Utf8, false),
            ("child_conversation_id", DataType::Utf8, true),
            ("child_agent_id", DataType::Utf8, true),
            ("carried_count", DataType::UInt32, true),
            ("reason", DataType::Utf8, true),
            ("parent_agent_id", DataType::Utf8, true),
            ("denial_reason", DataType::Utf8, true),
            ("allowed", DataType::Utf8, true),
            ("signed_by", DataType::Binary, true),
            ("signature_status", 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);
        }
    }

    /// A real signed `handoff` decodes to one row, and a signed
    /// `handoff_denied` decodes to a second row with `allowed` intact.
    #[test]
    fn decode_round_trips_a_signed_handoff_and_a_denial() {
        let turn = Uuid::from_u128(0x0195_abcd_ef01_2345_6789_abcd_ef01_5555);
        let signer = HandoffSigner::from_seed(21);
        let handoff = signed_handoff(&signer, "child-a");
        let denied = signed_denied(&signer);

        let events = vec![
            (
                1,
                Event::new(
                    kinds::tagged(kinds::HANDOFF, &turn),
                    handoff.encode_to_vec(),
                ),
            ),
            (
                2,
                Event::new(
                    kinds::tagged(kinds::HANDOFF_DENIED, &turn),
                    denied.encode_to_vec(),
                ),
            ),
        ];

        let decoded = decode_handoffs_events("conv-rt", &events, &trust(&signer));
        assert_eq!(decoded.len(), 2);

        let h = &decoded[0];
        assert_eq!(h.phase, "handoff");
        assert_eq!(h.child_conversation_id.as_deref(), Some("child-a"));
        assert_eq!(h.child_agent_id.as_deref(), Some("researcher"));
        assert_eq!(h.carried_count, Some(2));
        assert_eq!(h.reason.as_deref(), Some("delegate research"));
        assert_eq!(h.signature_status, "verified");
        assert_eq!(
            h.denial_reason, None,
            "a refusal-only column is NULL on this row"
        );

        let d = &decoded[1];
        assert_eq!(d.phase, "handoff_denied");
        assert_eq!(
            d.child_conversation_id, None,
            "a handoff-only column is NULL on this row"
        );
        assert_eq!(d.child_agent_id.as_deref(), Some("banned-agent"));
        assert_eq!(d.parent_agent_id.as_deref(), Some("assistant"));
        assert_eq!(
            d.denial_reason.as_deref(),
            Some("this agent can't hand off to that agent")
        );
        assert_eq!(
            d.allowed.as_deref(),
            Some(r#"["coding","research"]"#),
            "allowed round-trips as JSON-array text"
        );
        assert_eq!(d.signature_status, "verified");

        let batch = decode_handoffs_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), "handoff");
        assert_eq!(phase.value(1), "handoff_denied");
    }

    /// The load-bearing contrast with `payments`' drop-on-untrusted-signer
    /// test: a `Handoff` tampered AFTER signing (mutating
    /// `child_conversation_id`, mirroring `crates/crypto/src/handoff.rs`'s own
    /// `handoff_tampered_child_id_fails`) still decodes to a row —
    /// `signature_status = "invalid"`, never dropped — exactly
    /// `collect_handoffs`' own posture.
    #[test]
    fn tampered_handoff_keeps_the_row_but_reads_invalid() {
        let turn = Uuid::from_u128(0x0195_abcd_ef01_2345_6789_abcd_ef01_6666);
        let signer = HandoffSigner::from_seed(22);
        let mut handoff = signed_handoff(&signer, "child-tampered");
        handoff.child_conversation_id = "child-evil".to_owned();

        let events = vec![(
            1,
            Event::new(
                kinds::tagged(kinds::HANDOFF, &turn),
                handoff.encode_to_vec(),
            ),
        )];
        let decoded = decode_handoffs_events("conv-tampered", &events, &trust(&signer));
        assert_eq!(
            decoded.len(),
            1,
            "a tampered handoff must still surface as a row, unlike payments"
        );
        assert_eq!(decoded[0].signature_status, "invalid");
        assert_eq!(
            decoded[0].child_conversation_id.as_deref(),
            Some("child-evil"),
            "the tampered (claimed) value is still shown, same as collect_handoffs"
        );
    }

    /// Only a structurally-malformed payload (not even valid protobuf for its
    /// phase's message type) drops the row.
    #[test]
    fn structurally_malformed_payload_drops_the_row() {
        let events = vec![(
            1,
            Event::new(kinds::HANDOFF.to_owned(), vec![0xFF, 0xFE, 0xFD]),
        )];
        let decoded = decode_handoffs_events(
            "conv-corrupt",
            &events,
            &trust(&HandoffSigner::from_seed(24)),
        );
        assert_eq!(decoded.len(), 0);
    }

    #[test]
    fn unrelated_kind_is_not_decoded_as_a_handoff() {
        let events = vec![(1, Event::new(kinds::USAGE.to_owned(), Vec::new()))];
        let decoded = decode_handoffs_events(
            "conv-unrelated",
            &events,
            &trust(&HandoffSigner::from_seed(25)),
        );
        assert_eq!(decoded.len(), 0);
    }

    #[test]
    fn bare_kind_has_no_turn_id() {
        let signer = HandoffSigner::from_seed(23);
        let handoff = signed_handoff(&signer, "child-bare");
        let events = vec![(
            7,
            Event::new(kinds::HANDOFF.to_owned(), handoff.encode_to_vec()),
        )];
        let decoded = decode_handoffs_events("conv-bare", &events, &trust(&signer));
        assert_eq!(decoded.len(), 1);
        assert_eq!(decoded[0].turn_id, None);
    }

    /// A handoff signed by a key this deployment never held keeps its row and
    /// reads `untrusted` — the #1124 verdict the old boolean could not give.
    #[test]
    fn a_foreign_signed_handoff_keeps_the_row_but_reads_untrusted() {
        let signer = HandoffSigner::from_seed(26);
        let deployment = HandoffSigner::from_seed(27);
        let handoff = signed_handoff(&signer, "child-foreign");
        let events = vec![(
            3,
            Event::new(kinds::HANDOFF.to_owned(), handoff.encode_to_vec()),
        )];

        let decoded = decode_handoffs_events("conv-foreign", &events, &trust(&deployment));
        assert_eq!(decoded.len(), 1);
        assert_eq!(decoded[0].signature_status, "untrusted");
        assert_eq!(
            decoded[0].child_conversation_id.as_deref(),
            Some("child-foreign"),
            "the claimed value is still shown, same as a tampered row"
        );
    }
}