polyc-query 2026.8.3

//! `routine_lifecycle` typed-table decoder — the five signed routine
//! lifecycle audit kinds (`routine_created`/`routine_paused`/
//! `routine_resumed`/`routine_deleted`, `#1497`/`#1495`, plus
//! `routine_scope_changed`, `#1806`) → Arrow (issue #1593, the routines
//! explorer page's per-routine lifecycle timeline source).
//!
//! `kinds::ROUTINE_CREATED`/`ROUTINE_PAUSED`/`ROUTINE_RESUMED`/
//! `ROUTINE_DELETED` were all [`crate::decode::Decode::Opaque`] until this
//! table — [`crate::decode::mod`]'s own `REGISTRY` comment named exactly this
//! plan: "give all four a typed table together... rather than one now and
//! three later." This is the fact model's thirteenth typed table.
//! `kinds::ROUTINE_SCOPE_CHANGED` (`#1806`) joined the same table as a fifth
//! phase rather than a new one: it lands on the identical
//! `"routine-scheduler"` partition, is signed with the identical
//! `ApprovalSigner` JSON construction, and represents the same "lifecycle
//! audit trail of a routine CR" fact — the only new information is which
//! sharing scope the owner flipped to, carried by the phase-specific `scope`
//! column below.
//!
//! # One table, a `phase` discriminator with FIVE values — mirrors `handoffs`
//!
//! Same shape [`crate::decode::handoffs`] already establishes for its own
//! three-kind family: one table, a `phase` column (here, the kind-base
//! string itself — `"routine_created"`/`"routine_paused"`/
//! `"routine_resumed"`/`"routine_deleted"`/`"routine_scope_changed"`, needing
//! no separate constant), and every column that applies to only some phases
//! `NULL` on a row of a phase it doesn't apply to. `reason` is
//! `routine_paused`-only (the only one whose signed payload carries an
//! optional freeform reason); `scope` is `routine_scope_changed`-only (the
//! only one whose signed payload names a target sharing scope); `channel`
//! (POLY-160) is `routine_paused`/`routine_resumed`/`routine_deleted`-only —
//! `"chat"` or `"rpc"`, which surface minted the event, so a forensics reader
//! can tell "persona P paused routine X over the RPC channel" apart from the
//! same action taken mid-conversation, rather than reading `conversation_id`
//! alone (which an RPC mutation sets to the routine's own fire conversation,
//! and could otherwise look like the routine acted on itself). Every other
//! column (`routine`, `actor_persona`, `conversation_id`, `at_ms`) applies to
//! all five.
//!
//! # `trusted_signers` — dropped if unverified, NOT kept-and-flagged
//!
//! Unlike [`crate::decode::grant_replays`] (an audit surface that
//! deliberately KEEPS a tampered/forged record to show the forgery itself),
//! this table follows [`crate::decode::payments`]'s stricter posture: a
//! payload that fails `polyc_crypto::approval::verify_routine_created`/
//! `verify_routine_paused`/`verify_routine_resumed`/`verify_routine_deleted`/
//! `verify_routine_scope_changed`
//! (malformed JSON, a missing field, or a signature that doesn't check out —
//! that verify call conflates all three into one `None`, so this table
//! cannot distinguish them, unlike `handoffs`' separate decode/verify steps)
//! is dropped from the batch, and so is a payload that verifies against its
//! own embedded key but whose key is outside `trusted_signers` (the
//! deployment's approval-signer allow-list, threaded exactly the way it
//! reaches [`crate::decode::payments::decode_payments_events`]/
//! [`crate::decode::approvals::decode_approvals_events`]/
//! [`crate::decode::grant_replays::decode_grant_replays_events`]). A routine's
//! lifecycle timeline is a plain audit trail of the control plane's own
//! signed record-keeping, not a surface built to expose a forgery attempt —
//! fail closed, the same posture `kinds.rs` documents for
//! `taint_excision`/`grant_suspension`. Dropped from the *batch*, never from
//! the *log*: a real signer-rotation event (#1834's class) would otherwise
//! empty this table's history with zero diagnostic (#1919), so every dropped
//! row logs a `tracing::warn!` naming the partition, journal position, event
//! kind, and drop reason — [`warn_verification_failed`]/
//! [`warn_untrusted_signer`] below — matching the warn-on-drop posture every
//! decoder in this crate uses for a dropped/skipped row (`crate::decode::mod`'s
//! shared `decode_typed_kind_events` loop and `handoffs`' own
//! `warn_corrupt_payload` both warn rather than stay silent on a drop).
//!
//! # Fleet-only, the same way as `fires` — NOT the `payments`/`attribution`
//! # column-redaction shape
//!
//! Every one of these five kinds lands on the scheduler's own
//! `"routine-scheduler"` partition (never a conversation partition), the
//! IDENTICAL partition [`crate::decode::fires`] reads — admitted into replay
//! only for [`crate::session::QueryScope::Fleet`] (`crate::authority`'s
//! module doc). This table is therefore registered ONLY for `Fleet`, the
//! same "never even registered outside Fleet" posture [`crate::decode::fires`]
//! and [`crate::engine::SUMMARY_TABLE`] use, for the identical underlying
//! reason: a non-Fleet replay never reads this partition at all, so a
//! non-Fleet `routine_lifecycle` would always resolve to zero rows anyway.
//!
//! # Uniform keys, with the same caveat as `fires`
//!
//! `partition` is always `"routine-scheduler"` and `turn_id` is always
//! `None` — none of these five kinds is ever tagged with a turn (a routine's
//! lifecycle is not itself a turn) — see [`crate::decode::fires`]'s own
//! module docs for the identical caveat. Both columns are kept anyway,
//! matching every other typed table's uniform-key discipline (#1311).

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_crypto::approval::{
    VerifiedRoutineCreated, VerifiedRoutineDeleted, VerifiedRoutinePaused, VerifiedRoutineResumed,
    VerifiedRoutineScopeChanged, verify_routine_created, verify_routine_deleted,
    verify_routine_paused, verify_routine_resumed, verify_routine_scope_changed,
};
use polyc_eventlog::Event;
use polyc_proto::kinds;

/// One decoded `routine_lifecycle` row — one of the five phases,
/// discriminated by [`Self::phase`] (the kind-base string itself). The fact
/// model's uniform key columns (`partition`, `position`, `turn_id`) plus
/// every phase's own fields, `NULL` on the columns (`reason`, `scope`) that
/// don't apply to every phase.
#[derive(Debug, Clone)]
pub(crate) struct RoutineLifecycleRow {
    /// The journal partition this row's event was read from — always
    /// `"routine-scheduler"` (see the module doc).
    pub partition: String,
    /// The journal's own monotonic append position for this event.
    pub position: u64,
    /// Always `None` — see the module doc's "Uniform keys" section.
    pub turn_id: Option<String>,
    /// `kinds::ROUTINE_CREATED`/`ROUTINE_PAUSED`/`ROUTINE_RESUMED`/
    /// `ROUTINE_DELETED`/`ROUTINE_SCOPE_CHANGED` — the kind-base string
    /// itself, needing no separate constant (see the module doc).
    pub phase: String,
    /// The routine CR's name every one of the five payloads names.
    pub routine: String,
    /// The persona id who took the action: the creator for
    /// `routine_created`, the pausing/resuming/deleting/scope-flipping
    /// persona for the other four.
    pub actor_persona: String,
    /// The conversation the action was taken from.
    pub conversation_id: String,
    /// Unix ms the action was recorded — `created_at_ms`/`paused_at_ms`/
    /// `resumed_at_ms`/`deleted_at_ms`/`changed_at_ms`, unified onto one
    /// column since every phase carries exactly one such timestamp.
    pub at_ms: u64,
    /// `routine_paused`-only: why the routine was paused, if the pauser gave
    /// one. `NULL` on every other phase's row.
    pub reason: Option<String>,
    /// `routine_scope_changed`-only: the sharing scope (`"public"` or
    /// `"private"`) the owner flipped the routine to. `NULL` on every other
    /// phase's row.
    pub scope: Option<String>,
    /// `routine_paused`/`routine_resumed`/`routine_deleted`-only: `"chat"` or
    /// `"rpc"` — which surface minted the event (POLY-160). `NULL` on
    /// `routine_created`/`routine_scope_changed`, which predate the RPC
    /// surface and have no second channel to distinguish.
    pub channel: Option<String>,
    /// The embedded ed25519 public key the payload claims to be signed by —
    /// already checked against `trusted_signers` before this row exists (see
    /// the module doc's "`trusted_signers`" section), kept for a caller that
    /// wants to see which deployment key signed it.
    pub signer_public_key: Vec<u8>,
}

/// The `routine_lifecycle` typed table's Arrow schema.
#[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("routine", DataType::Utf8, false),
        Field::new("actor_persona", DataType::Utf8, false),
        Field::new("conversation_id", DataType::Utf8, false),
        Field::new("at_ms", DataType::UInt64, false),
        Field::new("reason", DataType::Utf8, true),
        Field::new("scope", DataType::Utf8, true),
        Field::new("channel", DataType::Utf8, true),
        Field::new("signer_public_key", DataType::Binary, false),
    ]))
}

/// Decode already-framed [`RoutineLifecycleRow`]s into the
/// `routine_lifecycle_raw` table's Arrow `RecordBatch`, in [`schema`] order.
///
/// # Errors
///
/// Returns [`ArrowError`] if Arrow batch construction fails.
pub(crate) fn decode_routine_lifecycle_batch(
    rows: &[RoutineLifecycleRow],
) -> Result<RecordBatch, ArrowError> {
    let mut partition_b = StringBuilder::with_capacity(rows.len(), rows.len() * 16);
    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 routine_b = StringBuilder::with_capacity(rows.len(), rows.len() * 24);
    let mut actor_persona_b = StringBuilder::with_capacity(rows.len(), rows.len() * 16);
    let mut conversation_id_b = StringBuilder::with_capacity(rows.len(), rows.len() * 16);
    let mut at_ms_b = UInt64Builder::with_capacity(rows.len());
    let mut reason_b = StringBuilder::with_capacity(rows.len(), rows.len() * 16);
    let mut scope_b = StringBuilder::with_capacity(rows.len(), rows.len() * 8);
    let mut channel_b = StringBuilder::with_capacity(rows.len(), rows.len() * 4);
    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(),
        }
        phase_b.append_value(&row.phase);
        routine_b.append_value(&row.routine);
        actor_persona_b.append_value(&row.actor_persona);
        conversation_id_b.append_value(&row.conversation_id);
        at_ms_b.append_value(row.at_ms);
        match &row.reason {
            Some(v) => reason_b.append_value(v),
            None => reason_b.append_null(),
        }
        match &row.scope {
            Some(v) => scope_b.append_value(v),
            None => scope_b.append_null(),
        }
        match &row.channel {
            Some(v) => channel_b.append_value(v),
            None => channel_b.append_null(),
        }
        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(phase_b.finish()),
        Arc::new(routine_b.finish()),
        Arc::new(actor_persona_b.finish()),
        Arc::new(conversation_id_b.finish()),
        Arc::new(at_ms_b.finish()),
        Arc::new(reason_b.finish()),
        Arc::new(scope_b.finish()),
        Arc::new(channel_b.finish()),
        Arc::new(signer_public_key_b.finish()),
    ];
    RecordBatch::try_new(schema(), columns)
}

/// `true` when `key` is a member of `trusted_signers` — the same allow-list
/// check [`crate::decode::payments::decode_payments_events`] applies via
/// `polyc_facts::verified_receipts`, reproduced here directly since this
/// table has no shared fold to apply it inside.
fn is_trusted(key: &[u8], trusted_signers: &[Vec<u8>]) -> bool {
    trusted_signers.iter().any(|k| k.as_slice() == key)
}

/// Logs the drop of a row whose payload failed `verify_routine_*` — malformed
/// JSON, a missing field, or a signature that doesn't check out. The verify
/// call conflates all three into one `None` (module doc), so this can only
/// name which of the two drop reasons this is, not a finer cause.
fn warn_verification_failed(partition: &str, position: u64, kind_base: &str) {
    tracing::warn!(
        partition,
        position,
        kind = kind_base,
        table = "routine_lifecycle",
        "routine_lifecycle: dropping row — payload failed verify_routine_* \
         (malformed payload or a signature that doesn't check out)"
    );
}

/// Logs the drop of a row whose payload verified against its own embedded
/// key, but whose key is outside `trusted_signers` — e.g. the real
/// signer-rotation case #1919 exists to make loud instead of silent. The
/// rejected key is logged lowercase-hex via [`polyc_crypto::hex::lower`], the
/// same encoding this table's own signed payloads carry it in
/// (`signed_by`/`signature_hex`).
fn warn_untrusted_signer(
    partition: &str,
    position: u64,
    kind_base: &str,
    signer_public_key: &[u8],
) {
    tracing::warn!(
        partition,
        position,
        kind = kind_base,
        table = "routine_lifecycle",
        signer_public_key = %polyc_crypto::hex::lower(signer_public_key),
        "routine_lifecycle: dropping row — signer_public_key is not in trusted_signers"
    );
}

/// Runs one phase's `verify_routine_*` against `payload`, then checks the
/// verified signer against `trusted_signers` — the two-step drop gate every
/// phase branch in [`decode_routine_lifecycle_events`] applies, factored out
/// once so each branch is a single call instead of repeating both checks
/// (and their now-mandatory warns, #1919) five times over.
///
/// A payload that fails to verify at all (malformed, or a signature that
/// doesn't check out) or whose embedded key is outside `trusted_signers` is
/// dropped from the batch — see the module doc's "`trusted_signers`" section
/// for why this table follows `payments`' drop posture rather than
/// `grant_replays`'/`handoffs`' keep-and-flag one. Every drop still logs a
/// `tracing::warn!` (via [`warn_verification_failed`]/
/// [`warn_untrusted_signer`]) naming the partition, position, kind, and
/// reason — a real signer-rotation gap (`#1834`) is loud, never a silent
/// zero.
fn verify_trusted<V>(
    partition: &str,
    position: u64,
    kind_base: &str,
    payload: &[u8],
    trusted_signers: &[Vec<u8>],
    verify: impl FnOnce(&[u8]) -> Option<V>,
    signer_public_key: impl Fn(&V) -> &[u8],
) -> Option<V> {
    let Some(v) = verify(payload) else {
        warn_verification_failed(partition, position, kind_base);
        return None;
    };
    if is_trusted(signer_public_key(&v), trusted_signers) {
        Some(v)
    } else {
        warn_untrusted_signer(partition, position, kind_base, signer_public_key(&v));
        None
    }
}

/// One phase's row-decode contract: the fixed pieces
/// [`decode_phase_row`] needs to turn a payload into a
/// [`RoutineLifecycleRow`] — the kind-base `phase` string, the
/// `verify_routine_*` function, an accessor for the verified record's
/// embedded signer key (for the `trusted_signers` check), and `fields`,
/// which pulls the phase-specific `(routine, actor_persona,
/// conversation_id, at_ms, reason, scope, signer_public_key)` tuple out of
/// the verified record. None of the four capture anything, so every phase
/// instantiates this with plain function items — no closures, no per-phase
/// wrapper function to hand-roll.
struct PhaseSpec<V> {
    /// The kind-base string this phase decodes — also becomes
    /// [`RoutineLifecycleRow::phase`].
    phase: &'static str,
    /// `verify_routine_*` for this phase.
    verify: fn(&[u8]) -> Option<V>,
    /// Reads the verified record's embedded signer key, for the
    /// `trusted_signers` check.
    signer_public_key: fn(&V) -> &[u8],
    /// Pulls this phase's row fields out of the verified record.
    #[allow(clippy::type_complexity)]
    fields: fn(
        V,
    ) -> (
        String,
        String,
        String,
        u64,
        Option<String>,
        Option<String>,
        Option<String>,
        Vec<u8>,
    ),
}

/// One phase's row-decode: verify+trust-check `payload` against `spec` (via
/// [`verify_trusted`], which owns the drop-and-warn policy for both failure
/// modes), then map the verified record into a [`RoutineLifecycleRow`].
/// Generic over the verified record type `V` so every phase shares this one
/// builder instead of hand-rolling its own near-identical function; each of
/// [`decode_routine_lifecycle_events`]'s five dispatch arms is now a single
/// call passing that phase's [`PhaseSpec`].
fn decode_phase_row<V>(
    partition: &str,
    position: u64,
    turn_id: Option<String>,
    payload: &[u8],
    trusted_signers: &[Vec<u8>],
    spec: &PhaseSpec<V>,
) -> Option<RoutineLifecycleRow> {
    let v = verify_trusted(
        partition,
        position,
        spec.phase,
        payload,
        trusted_signers,
        spec.verify,
        spec.signer_public_key,
    )?;
    let (routine, actor_persona, conversation_id, at_ms, reason, scope, channel, signer_public_key) =
        (spec.fields)(v);
    Some(RoutineLifecycleRow {
        partition: partition.to_string(),
        position,
        turn_id,
        phase: spec.phase.to_string(),
        routine,
        actor_persona,
        conversation_id,
        at_ms,
        reason,
        scope,
        channel,
        signer_public_key,
    })
}

#[allow(clippy::type_complexity)] // one distinct row field per line, mirrors PhaseSpec::fields
fn created_fields(
    v: VerifiedRoutineCreated,
) -> (
    String,
    String,
    String,
    u64,
    Option<String>,
    Option<String>,
    Option<String>,
    Vec<u8>,
) {
    (
        v.routine,
        v.creator_persona,
        v.conversation_id,
        v.created_at_ms,
        None,
        None,
        None,
        v.signer_public_key,
    )
}

#[allow(clippy::type_complexity)] // one distinct row field per line, mirrors PhaseSpec::fields
fn paused_fields(
    v: VerifiedRoutinePaused,
) -> (
    String,
    String,
    String,
    u64,
    Option<String>,
    Option<String>,
    Option<String>,
    Vec<u8>,
) {
    (
        v.routine,
        v.actor_persona,
        v.conversation_id,
        v.paused_at_ms,
        v.reason,
        None,
        Some(v.channel),
        v.signer_public_key,
    )
}

#[allow(clippy::type_complexity)] // one distinct row field per line, mirrors PhaseSpec::fields
fn resumed_fields(
    v: VerifiedRoutineResumed,
) -> (
    String,
    String,
    String,
    u64,
    Option<String>,
    Option<String>,
    Option<String>,
    Vec<u8>,
) {
    (
        v.routine,
        v.actor_persona,
        v.conversation_id,
        v.resumed_at_ms,
        None,
        None,
        Some(v.channel),
        v.signer_public_key,
    )
}

#[allow(clippy::type_complexity)] // one distinct row field per line, mirrors PhaseSpec::fields
fn deleted_fields(
    v: VerifiedRoutineDeleted,
) -> (
    String,
    String,
    String,
    u64,
    Option<String>,
    Option<String>,
    Option<String>,
    Vec<u8>,
) {
    (
        v.routine,
        v.actor_persona,
        v.conversation_id,
        v.deleted_at_ms,
        None,
        None,
        Some(v.channel),
        v.signer_public_key,
    )
}

#[allow(clippy::type_complexity)] // one distinct row field per line, mirrors PhaseSpec::fields
fn scope_changed_fields(
    v: VerifiedRoutineScopeChanged,
) -> (
    String,
    String,
    String,
    u64,
    Option<String>,
    Option<String>,
    Option<String>,
    Vec<u8>,
) {
    (
        v.routine,
        v.actor_persona,
        v.conversation_id,
        v.changed_at_ms,
        None,
        Some(v.scope),
        None,
        v.signer_public_key,
    )
}

/// Filter `partition`'s framed `events` to the five routine lifecycle kinds,
/// verify each payload against its own embedded signer key AND
/// `trusted_signers` (via [`verify_trusted`], applied per phase by
/// [`decode_phase_row`]), and pair a fully-verified record with that row's
/// uniform key columns.
#[must_use]
pub(crate) fn decode_routine_lifecycle_events(
    partition: &str,
    events: &[(u64, Event)],
    trusted_signers: &[Vec<u8>],
) -> Vec<RoutineLifecycleRow> {
    events
        .iter()
        .filter_map(|(position, event)| {
            let (base, turn_id) = kinds::parse(&event.kind);
            let turn_id = turn_id.map(|id| id.to_string());
            if base == kinds::ROUTINE_CREATED {
                decode_phase_row(
                    partition,
                    *position,
                    turn_id,
                    &event.payload,
                    trusted_signers,
                    &PhaseSpec {
                        phase: kinds::ROUTINE_CREATED,
                        verify: verify_routine_created,
                        signer_public_key: |v| v.signer_public_key.as_slice(),
                        fields: created_fields,
                    },
                )
            } else if base == kinds::ROUTINE_PAUSED {
                decode_phase_row(
                    partition,
                    *position,
                    turn_id,
                    &event.payload,
                    trusted_signers,
                    &PhaseSpec {
                        phase: kinds::ROUTINE_PAUSED,
                        verify: verify_routine_paused,
                        signer_public_key: |v| v.signer_public_key.as_slice(),
                        fields: paused_fields,
                    },
                )
            } else if base == kinds::ROUTINE_RESUMED {
                decode_phase_row(
                    partition,
                    *position,
                    turn_id,
                    &event.payload,
                    trusted_signers,
                    &PhaseSpec {
                        phase: kinds::ROUTINE_RESUMED,
                        verify: verify_routine_resumed,
                        signer_public_key: |v| v.signer_public_key.as_slice(),
                        fields: resumed_fields,
                    },
                )
            } else if base == kinds::ROUTINE_DELETED {
                decode_phase_row(
                    partition,
                    *position,
                    turn_id,
                    &event.payload,
                    trusted_signers,
                    &PhaseSpec {
                        phase: kinds::ROUTINE_DELETED,
                        verify: verify_routine_deleted,
                        signer_public_key: |v| v.signer_public_key.as_slice(),
                        fields: deleted_fields,
                    },
                )
            } else if base == kinds::ROUTINE_SCOPE_CHANGED {
                decode_phase_row(
                    partition,
                    *position,
                    turn_id,
                    &event.payload,
                    trusted_signers,
                    &PhaseSpec {
                        phase: kinds::ROUTINE_SCOPE_CHANGED,
                        verify: verify_routine_scope_changed,
                        signer_public_key: |v| v.signer_public_key.as_slice(),
                        fields: scope_changed_fields,
                    },
                )
            } else {
                None
            }
        })
        .collect()
}

#[cfg(test)]
mod tests {
    use arrow::array::Array as _;
    use polyc_crypto::approval::{
        ApprovalSigner, routine_created_payload, routine_deleted_payload, routine_paused_payload,
        routine_resumed_payload, routine_scope_changed_payload,
    };

    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",
                "routine",
                "actor_persona",
                "conversation_id",
                "at_ms",
                "reason",
                "scope",
                "channel",
                "signer_public_key",
            ]
        );
        let expect = [
            ("partition", DataType::Utf8, false),
            ("position", DataType::UInt64, false),
            ("turn_id", DataType::Utf8, true),
            ("phase", DataType::Utf8, false),
            ("routine", DataType::Utf8, false),
            ("actor_persona", DataType::Utf8, false),
            ("conversation_id", DataType::Utf8, false),
            ("at_ms", DataType::UInt64, false),
            ("reason", DataType::Utf8, true),
            ("scope", DataType::Utf8, true),
            ("channel", DataType::Utf8, true),
            ("signer_public_key", DataType::Binary, 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);
        }
    }

    /// Acceptance criterion: a signed `routine_created`/`routine_paused`/
    /// `routine_resumed`/`routine_deleted`/`routine_scope_changed` event each
    /// round-trips into a timeline row.
    #[test]
    #[allow(clippy::too_many_lines)] // one payload builder call per phase, now each multi-line for tool_call_id (#1638)
    fn decode_round_trips_all_five_signed_lifecycle_kinds() {
        let signer = ApprovalSigner::from_seed(1);
        let trusted_signers = vec![signer.public_key_bytes()];

        let (created, _, _) = routine_created_payload(
            "daily-standup",
            "persona-1",
            "conv-1",
            "call-1",
            "hash-1",
            1_000,
            &signer,
        );
        let (paused, _, _) = routine_paused_payload(
            "daily-standup",
            "persona-1",
            "conv-2",
            "call-2",
            "hash-2",
            "chat",
            2_000,
            Some("rotating out old announcements"),
            &signer,
        );
        let (resumed, _, _) = routine_resumed_payload(
            "daily-standup",
            "persona-1",
            "conv-3",
            "call-3",
            "hash-3",
            "chat",
            3_000,
            &signer,
        );
        let (deleted, _, _) = routine_deleted_payload(
            "daily-standup",
            "persona-1",
            "conv-4",
            "call-4",
            "hash-4",
            "chat",
            4_000,
            &signer,
        );
        let (scope_changed, _, _) = routine_scope_changed_payload(
            "daily-standup",
            "persona-1",
            "conv-5",
            "call-5",
            "hash-5",
            "public",
            5_000,
            &signer,
        );

        let events = vec![
            (10, Event::new(kinds::ROUTINE_CREATED, created)),
            (11, Event::new(kinds::ROUTINE_PAUSED, paused)),
            (12, Event::new(kinds::ROUTINE_RESUMED, resumed)),
            (13, Event::new(kinds::ROUTINE_DELETED, deleted)),
            (14, Event::new(kinds::ROUTINE_SCOPE_CHANGED, scope_changed)),
        ];

        let decoded =
            decode_routine_lifecycle_events("routine-scheduler", &events, &trusted_signers);
        assert_eq!(decoded.len(), 5);

        assert_eq!(decoded[0].phase, "routine_created");
        assert_eq!(decoded[0].routine, "daily-standup");
        assert_eq!(decoded[0].actor_persona, "persona-1");
        assert_eq!(decoded[0].conversation_id, "conv-1");
        assert_eq!(decoded[0].at_ms, 1_000);
        assert_eq!(decoded[0].reason, None);
        assert_eq!(decoded[0].scope, None);
        assert_eq!(decoded[0].channel, None);

        assert_eq!(decoded[1].phase, "routine_paused");
        assert_eq!(decoded[1].at_ms, 2_000);
        assert_eq!(
            decoded[1].reason.as_deref(),
            Some("rotating out old announcements")
        );
        assert_eq!(decoded[1].scope, None);
        assert_eq!(decoded[1].channel.as_deref(), Some("chat"));

        assert_eq!(decoded[2].phase, "routine_resumed");
        assert_eq!(decoded[2].at_ms, 3_000);
        assert_eq!(decoded[2].reason, None);
        assert_eq!(decoded[2].scope, None);
        assert_eq!(decoded[2].channel.as_deref(), Some("chat"));

        assert_eq!(decoded[3].phase, "routine_deleted");
        assert_eq!(decoded[3].at_ms, 4_000);
        assert_eq!(decoded[3].reason, None);
        assert_eq!(decoded[3].scope, None);
        assert_eq!(decoded[3].channel.as_deref(), Some("chat"));

        assert_eq!(decoded[4].phase, "routine_scope_changed");
        assert_eq!(decoded[4].routine, "daily-standup");
        assert_eq!(decoded[4].actor_persona, "persona-1");
        assert_eq!(decoded[4].conversation_id, "conv-5");
        assert_eq!(decoded[4].at_ms, 5_000);
        assert_eq!(decoded[4].reason, None);
        assert_eq!(decoded[4].scope.as_deref(), Some("public"));
        assert_eq!(decoded[4].channel, None);

        let batch = decode_routine_lifecycle_batch(&decoded).expect("batch build");
        assert_eq!(batch.num_rows(), 5);
        assert_eq!(batch.schema(), schema());
        let phase = batch
            .column(3)
            .as_any()
            .downcast_ref::<arrow::array::StringArray>()
            .unwrap();
        assert_eq!(phase.value(0), "routine_created");
        assert_eq!(phase.value(3), "routine_deleted");
        assert_eq!(phase.value(4), "routine_scope_changed");

        let scope = batch
            .column(9)
            .as_any()
            .downcast_ref::<arrow::array::StringArray>()
            .unwrap();
        assert!(scope.is_null(0));
        assert_eq!(scope.value(4), "public");
    }

    /// A scope flip's `scope` value round-trips exactly, including a flip
    /// back to `"private"`.
    #[test]
    fn scope_changed_carries_the_target_scope() {
        let signer = ApprovalSigner::from_seed(6);
        let trusted_signers = vec![signer.public_key_bytes()];
        let (payload, _, _) = routine_scope_changed_payload(
            "weekly-digest",
            "persona-2",
            "conv-b",
            "call-6",
            "hash-6",
            "private",
            6_000,
            &signer,
        );
        let events = vec![(1, Event::new(kinds::ROUTINE_SCOPE_CHANGED, payload))];
        let decoded =
            decode_routine_lifecycle_events("routine-scheduler", &events, &trusted_signers);
        assert_eq!(decoded.len(), 1);
        assert_eq!(decoded[0].phase, "routine_scope_changed");
        assert_eq!(decoded[0].scope.as_deref(), Some("private"));
        assert_eq!(decoded[0].reason, None);
    }

    /// A pause with no reason decodes with `reason = NULL`, not an empty
    /// string.
    #[test]
    fn paused_without_a_reason_has_null_reason() {
        let signer = ApprovalSigner::from_seed(2);
        let trusted_signers = vec![signer.public_key_bytes()];
        let (paused, _, _) = routine_paused_payload(
            "weekly-digest",
            "persona-2",
            "conv-a",
            "call-7",
            "hash-7",
            "chat",
            5_000,
            None,
            &signer,
        );
        let events = vec![(1, Event::new(kinds::ROUTINE_PAUSED, paused))];
        let decoded =
            decode_routine_lifecycle_events("routine-scheduler", &events, &trusted_signers);
        assert_eq!(decoded.len(), 1);
        assert_eq!(decoded[0].reason, None);
    }

    /// A structurally malformed payload (not even valid JSON) is dropped.
    #[test]
    fn malformed_payload_drops_the_row() {
        let events = vec![(1, Event::new(kinds::ROUTINE_CREATED, b"not json".to_vec()))];
        let decoded = decode_routine_lifecycle_events("routine-scheduler", &events, &[]);
        assert_eq!(decoded.len(), 0);
    }

    /// An internally-consistent but UNTRUSTED signer is dropped — the
    /// `payments`-style posture this table follows (see the module doc),
    /// unlike `grant_replays`/`handoffs`.
    #[test]
    fn untrusted_signer_drops_the_row() {
        let untrusted = ApprovalSigner::from_seed(3);
        let trusted = ApprovalSigner::from_seed(4);
        let (created, _, _) = routine_created_payload(
            "daily-standup",
            "persona-1",
            "conv-1",
            "call-8",
            "hash-8",
            1_000,
            &untrusted,
        );
        let events = vec![(1, Event::new(kinds::ROUTINE_CREATED, created))];
        let decoded = decode_routine_lifecycle_events(
            "routine-scheduler",
            &events,
            &[trusted.public_key_bytes()],
        );
        assert_eq!(
            decoded.len(),
            0,
            "an untrusted-signer lifecycle record must not surface as a row"
        );
    }

    /// A tampered (mutated after signing) payload fails self-verification and
    /// is dropped, same as a forged one.
    #[test]
    fn tampered_payload_drops_the_row() {
        let signer = ApprovalSigner::from_seed(5);
        let (created, _, _) = routine_created_payload(
            "daily-standup",
            "persona-1",
            "conv-1",
            "call-9",
            "hash-9",
            1_000,
            &signer,
        );
        let mut tampered: serde_json::Value = serde_json::from_slice(&created).unwrap();
        tampered["routine"] = serde_json::json!("evil-routine");
        let events = vec![(
            1,
            Event::new(kinds::ROUTINE_CREATED, tampered.to_string().into_bytes()),
        )];
        let decoded = decode_routine_lifecycle_events(
            "routine-scheduler",
            &events,
            &[signer.public_key_bytes()],
        );
        assert_eq!(decoded.len(), 0);
    }

    /// #1919: an untrusted-signer drop is no longer silent — it logs a
    /// `tracing::warn!` naming the partition, position, and kind, plus the
    /// rejected signer's hex-encoded key, matching the pattern
    /// `dashboard`'s own `an_unreadable_receipt_amount_is_counted_not_silently_skipped`
    /// test already uses for a different drop-with-diagnostic call site.
    #[test]
    #[tracing_test::traced_test]
    fn untrusted_signer_drop_is_logged() {
        let untrusted = ApprovalSigner::from_seed(9);
        let trusted = ApprovalSigner::from_seed(10);
        let (created, _, _) = routine_created_payload(
            "daily-standup",
            "persona-1",
            "conv-1",
            "call-10",
            "hash-10",
            1_000,
            &untrusted,
        );
        let events = vec![(42, Event::new(kinds::ROUTINE_CREATED, created))];
        let decoded = decode_routine_lifecycle_events(
            "routine-scheduler",
            &events,
            &[trusted.public_key_bytes()],
        );
        assert_eq!(decoded.len(), 0);
        assert!(
            logs_contain("routine_lifecycle"),
            "the drop must be logged, not silent"
        );
        assert!(
            logs_contain("routine-scheduler"),
            "the log line must name the partition"
        );
        assert!(
            logs_contain("trusted_signers"),
            "the log line must name the drop reason"
        );
        assert!(
            logs_contain(&polyc_crypto::hex::lower(&untrusted.public_key_bytes())),
            "the log line must name the rejected signer's key"
        );
    }

    /// #1919: a verification-failure drop (malformed payload, here) also
    /// logs, distinct from the untrusted-signer wording above.
    #[test]
    #[tracing_test::traced_test]
    fn verification_failure_drop_is_logged() {
        let events = vec![(7, Event::new(kinds::ROUTINE_CREATED, b"not json".to_vec()))];
        let decoded = decode_routine_lifecycle_events("routine-scheduler", &events, &[]);
        assert_eq!(decoded.len(), 0);
        assert!(
            logs_contain("routine_lifecycle"),
            "the drop must be logged, not silent"
        );
        assert!(
            logs_contain("routine-scheduler"),
            "the log line must name the partition"
        );
        assert!(
            logs_contain("verify_routine_"),
            "the log line must name the verification-failure drop reason"
        );
    }

    #[test]
    fn unrelated_kind_is_not_decoded() {
        let events = vec![(1, Event::new(kinds::ROUTINE_FIRED, Vec::new()))];
        let decoded = decode_routine_lifecycle_events("routine-scheduler", &events, &[]);
        assert_eq!(decoded.len(), 0);
    }

    #[test]
    fn empty_events_yield_zero_rows() {
        let decoded = decode_routine_lifecycle_events("routine-scheduler", &[], &[]);
        assert_eq!(decoded.len(), 0);
    }

    /// `#1834`: an event signed by a retired key (present in `trusted_signers`
    /// alongside the current key, exactly the way `load_trusted_signers`
    /// expands it) still decodes into a row — a signer rotation must not
    /// erase history a since-retired key signed.
    #[test]
    fn retired_signer_key_still_decodes() {
        let retired = ApprovalSigner::from_seed(7);
        let current = ApprovalSigner::from_seed(8);
        let (created, _, _) = routine_created_payload(
            "daily-standup",
            "persona-1",
            "conv-1",
            "call-11",
            "hash-11",
            1_000,
            &retired,
        );
        let events = vec![(1, Event::new(kinds::ROUTINE_CREATED, created))];
        // The trust set a `#1834`-aware caller passes: current + retired.
        let trusted_signers = vec![current.public_key_bytes(), retired.public_key_bytes()];
        let decoded =
            decode_routine_lifecycle_events("routine-scheduler", &events, &trusted_signers);
        assert_eq!(
            decoded.len(),
            1,
            "an event signed by a retired-but-trusted key must still decode"
        );
        assert_eq!(decoded[0].signer_public_key, retired.public_key_bytes());
    }

    /// `#1834`: a rotation from key A to key B must not drop either key's
    /// events — both an A-signed and a B-signed event decode once the trust
    /// set carries both (the shape `load_trusted_signers` produces after a
    /// rotation).
    #[test]
    fn rotation_keeps_both_old_and_new_signer_events_visible() {
        let key_a = ApprovalSigner::from_seed(9);
        let key_b = ApprovalSigner::from_seed(10);
        let (created_a, _, _) = routine_created_payload(
            "daily-standup",
            "persona-1",
            "conv-1",
            "call-12",
            "hash-12",
            1_000,
            &key_a,
        );
        let (created_b, _, _) = routine_created_payload(
            "weekly-digest",
            "persona-2",
            "conv-2",
            "call-13",
            "hash-13",
            2_000,
            &key_b,
        );
        let events = vec![
            (1, Event::new(kinds::ROUTINE_CREATED, created_a)),
            (2, Event::new(kinds::ROUTINE_CREATED, created_b)),
        ];
        let trusted_signers = vec![key_b.public_key_bytes(), key_a.public_key_bytes()];
        let decoded =
            decode_routine_lifecycle_events("routine-scheduler", &events, &trusted_signers);
        assert_eq!(
            decoded.len(),
            2,
            "both the pre-rotation and post-rotation signer's events must decode"
        );
        assert_eq!(decoded[0].routine, "daily-standup");
        assert_eq!(decoded[1].routine, "weekly-digest");
    }
}