polyc-query 2026.9.0

//! One partition's fully-decoded typed tables — the seam the per-partition
//! decode cache (`crate::cache`) caches, and [`super::QueryEngine::build`]
//! now consumes directly instead of decoding partitions itself.
//!
//! Before the cache existed, `QueryEngine::build` decoded every scoped
//! partition's raw [`super::PartitionEvents`] into Arrow inline, once per
//! query, every time.
//! [`decode_partition_tables`] is that same decode work, pulled out to run
//! PER PARTITION rather than per query: it is the one function both the
//! cache-miss path (`crate::authority::ScopedQuery`, fresh replay) and the
//! cache module itself (a hit's tail, decoded and appended to what is
//! already cached) call, so there is exactly one decode implementation
//! either way — never a second, cache-specific copy that could drift from
//! what a cold query decodes.

use arrow::record_batch::RecordBatch;
use polyc_crypto::signing_role::{HandoffRole, RoleTrustSet};
use polyc_eventlog::Event;

use crate::decode;

use super::QueryEngineError;
use super::tables::{
    APPROVALS_RAW_TABLE, ATTRIBUTION_RAW_TABLE, EVENTS_RAW_TABLE, FIRES_RAW_TABLE,
    GRANT_REPLAYS_RAW_TABLE, HANDOFFS_RAW_TABLE, MESSAGES_RAW_TABLE, MODEL_CALL_RAW_TABLE,
    PAYMENTS_RAW_TABLE, REFUSALS_RAW_TABLE, ROUTINE_LIFECYCLE_RAW_TABLE, ROUTINE_SETUP_RAW_TABLE,
    SUMMARY_RAW_TABLE, TOOL_CALLS_RAW_TABLE, TURN_DISPATCH_RAW_TABLE, TURN_FAILED_RAW_TABLE,
    USAGE_RAW_TABLE, WALLET_LINK_LIFECYCLE_RAW_TABLE,
};

/// One partition's decode output: exactly the eighteen journal-derived
/// typed tables [`super::QueryEngine::build`] registers from a partition's
/// own events (every raw table it builds EXCEPT the Fleet-only, non-journal
/// `personas`/`participations`/`persona_identities`/`routines` reference
/// tables — see `crate::cache`'s module doc for why those stay uncached).
///
/// Field order mirrors [`super::QueryEngine::build`]'s own registration
/// order; that order carries no semantic meaning beyond readability.
#[derive(Debug, Clone)]
#[allow(
    clippy::struct_field_names,
    reason = "every field's `_raw` suffix deliberately mirrors its own `*_RAW_TABLE` catalog \
              constant (e.g. `events_raw` <-> EVENTS_RAW_TABLE) — dropping it would break that \
              parallel, not just rename a field"
)]
pub(crate) struct PartitionTables {
    /// [`EVENTS_RAW_TABLE`]'s batch.
    pub events_raw: RecordBatch,
    /// [`USAGE_RAW_TABLE`]'s batch.
    pub usage_raw: RecordBatch,
    /// [`MODEL_CALL_RAW_TABLE`]'s batch.
    pub model_call_raw: RecordBatch,
    /// [`ATTRIBUTION_RAW_TABLE`]'s batch.
    pub attribution_raw: RecordBatch,
    /// [`TURN_FAILED_RAW_TABLE`]'s batch.
    pub turn_failed_raw: RecordBatch,
    /// [`TURN_DISPATCH_RAW_TABLE`]'s batch (issue #1593).
    pub turn_dispatch_raw: RecordBatch,
    /// [`PAYMENTS_RAW_TABLE`]'s batch.
    pub payments_raw: RecordBatch,
    /// [`REFUSALS_RAW_TABLE`]'s batch (`#2090`, INV-W5).
    pub refusals_raw: RecordBatch,
    /// [`WALLET_LINK_LIFECYCLE_RAW_TABLE`]'s batch (`#2123`).
    pub wallet_link_lifecycle_raw: RecordBatch,
    /// [`APPROVALS_RAW_TABLE`]'s batch.
    pub approvals_raw: RecordBatch,
    /// [`HANDOFFS_RAW_TABLE`]'s batch.
    pub handoffs_raw: RecordBatch,
    /// [`GRANT_REPLAYS_RAW_TABLE`]'s batch.
    pub grant_replays_raw: RecordBatch,
    /// [`SUMMARY_RAW_TABLE`]'s batch.
    pub summary_raw: RecordBatch,
    /// [`FIRES_RAW_TABLE`]'s batch (issue #1592).
    pub fires_raw: RecordBatch,
    /// [`ROUTINE_LIFECYCLE_RAW_TABLE`]'s batch (issue #1593).
    pub routine_lifecycle_raw: RecordBatch,
    /// [`ROUTINE_SETUP_RAW_TABLE`]'s batch.
    pub routine_setup_raw: RecordBatch,
    /// [`MESSAGES_RAW_TABLE`]'s batch.
    pub messages_raw: RecordBatch,
    /// [`TOOL_CALLS_RAW_TABLE`]'s batch.
    pub tool_calls_raw: RecordBatch,
}

impl PartitionTables {
    /// Every field, paired with the raw-table name it backs — the one place
    /// that enumerates all eighteen, so [`Self::memory_bytes`] and
    /// `crate::cache`'s concatenation step can iterate rather than naming
    /// each field twice.
    const fn fields(&self) -> [(&'static str, &RecordBatch); 18] {
        [
            (EVENTS_RAW_TABLE, &self.events_raw),
            (USAGE_RAW_TABLE, &self.usage_raw),
            (MODEL_CALL_RAW_TABLE, &self.model_call_raw),
            (ATTRIBUTION_RAW_TABLE, &self.attribution_raw),
            (TURN_FAILED_RAW_TABLE, &self.turn_failed_raw),
            (TURN_DISPATCH_RAW_TABLE, &self.turn_dispatch_raw),
            (PAYMENTS_RAW_TABLE, &self.payments_raw),
            (REFUSALS_RAW_TABLE, &self.refusals_raw),
            (
                WALLET_LINK_LIFECYCLE_RAW_TABLE,
                &self.wallet_link_lifecycle_raw,
            ),
            (APPROVALS_RAW_TABLE, &self.approvals_raw),
            (HANDOFFS_RAW_TABLE, &self.handoffs_raw),
            (GRANT_REPLAYS_RAW_TABLE, &self.grant_replays_raw),
            (SUMMARY_RAW_TABLE, &self.summary_raw),
            (FIRES_RAW_TABLE, &self.fires_raw),
            (ROUTINE_LIFECYCLE_RAW_TABLE, &self.routine_lifecycle_raw),
            (ROUTINE_SETUP_RAW_TABLE, &self.routine_setup_raw),
            (MESSAGES_RAW_TABLE, &self.messages_raw),
            (TOOL_CALLS_RAW_TABLE, &self.tool_calls_raw),
        ]
    }

    /// Total Arrow array memory this partition's cached tables occupy —
    /// `RecordBatch::get_array_memory_size`, summed across all eighteen
    /// tables. The decode cache's LRU eviction (`crate::cache`) sizes its
    /// total-bytes cap against this.
    pub(crate) fn memory_bytes(&self) -> usize {
        self.fields()
            .iter()
            .map(|(_, batch)| batch.get_array_memory_size())
            .sum()
    }

    /// Concatenate `self` (the cached state so far) with `tail` (freshly
    /// decoded events appended since the cached watermark), table by table,
    /// producing the merged state a cache tail-hit stores going forward.
    ///
    /// # Errors
    ///
    /// Returns [`QueryEngineError::Decode`] if any one table's schema
    /// disagrees between `self` and `tail` — this cannot happen in practice
    /// (every table's schema is a fixed constant, `crate::decode::*::schema`,
    /// never derived from the data itself), but `arrow::compute::concat_batches`
    /// is fallible, so the error is threaded rather than unwrapped.
    pub(crate) fn concat(&self, tail: &Self) -> Result<Self, QueryEngineError> {
        Ok(Self {
            events_raw: concat(EVENTS_RAW_TABLE, &self.events_raw, &tail.events_raw)?,
            usage_raw: concat(USAGE_RAW_TABLE, &self.usage_raw, &tail.usage_raw)?,
            model_call_raw: concat(
                MODEL_CALL_RAW_TABLE,
                &self.model_call_raw,
                &tail.model_call_raw,
            )?,
            attribution_raw: concat(
                ATTRIBUTION_RAW_TABLE,
                &self.attribution_raw,
                &tail.attribution_raw,
            )?,
            turn_failed_raw: concat(
                TURN_FAILED_RAW_TABLE,
                &self.turn_failed_raw,
                &tail.turn_failed_raw,
            )?,
            turn_dispatch_raw: concat(
                TURN_DISPATCH_RAW_TABLE,
                &self.turn_dispatch_raw,
                &tail.turn_dispatch_raw,
            )?,
            payments_raw: concat(PAYMENTS_RAW_TABLE, &self.payments_raw, &tail.payments_raw)?,
            refusals_raw: concat(REFUSALS_RAW_TABLE, &self.refusals_raw, &tail.refusals_raw)?,
            wallet_link_lifecycle_raw: concat(
                WALLET_LINK_LIFECYCLE_RAW_TABLE,
                &self.wallet_link_lifecycle_raw,
                &tail.wallet_link_lifecycle_raw,
            )?,
            approvals_raw: concat(
                APPROVALS_RAW_TABLE,
                &self.approvals_raw,
                &tail.approvals_raw,
            )?,
            handoffs_raw: concat(HANDOFFS_RAW_TABLE, &self.handoffs_raw, &tail.handoffs_raw)?,
            grant_replays_raw: concat(
                GRANT_REPLAYS_RAW_TABLE,
                &self.grant_replays_raw,
                &tail.grant_replays_raw,
            )?,
            summary_raw: concat(SUMMARY_RAW_TABLE, &self.summary_raw, &tail.summary_raw)?,
            fires_raw: concat(FIRES_RAW_TABLE, &self.fires_raw, &tail.fires_raw)?,
            routine_lifecycle_raw: concat(
                ROUTINE_LIFECYCLE_RAW_TABLE,
                &self.routine_lifecycle_raw,
                &tail.routine_lifecycle_raw,
            )?,
            routine_setup_raw: concat(
                ROUTINE_SETUP_RAW_TABLE,
                &self.routine_setup_raw,
                &tail.routine_setup_raw,
            )?,
            messages_raw: concat(MESSAGES_RAW_TABLE, &self.messages_raw, &tail.messages_raw)?,
            tool_calls_raw: concat(
                TOOL_CALLS_RAW_TABLE,
                &self.tool_calls_raw,
                &tail.tool_calls_raw,
            )?,
        })
    }
}

/// Concatenate two batches of the same table (`arrow::compute::concat_batches`),
/// wrapping a failure as [`QueryEngineError::Decode`] labeled `table`.
fn concat(
    table: &'static str,
    first: &RecordBatch,
    second: &RecordBatch,
) -> Result<RecordBatch, QueryEngineError> {
    arrow::compute::concat_batches(&first.schema(), [first, second])
        .map_err(|source| QueryEngineError::Decode { table, source })
}

/// Test-only: counts every [`decode_partition_tables`] call across the whole
/// test binary — the one function BOTH a fresh cache miss/tail decode and
/// `crate::engine::QueryEngine::build`'s own (test-only) wrapper ultimately
/// call, so this is the genuine decode fan-out entry point, unlike
/// `crate::engine::BUILD_CALL_COUNT` (which only counts entry into
/// `QueryEngine::build_from_tables`'s ENGINE-ASSEMBLY step — a step the
/// decode cache moved decode out of, so `BUILD_CALL_COUNT` alone does not
/// prove decode never ran).
/// `crate::authority::tests::fleet_query_over_source_budget_is_rejected_before_decode`
/// diffs THIS counter for exactly that reason. Same plain-`AtomicUsize`,
/// diff-not-absolute-value posture as `BUILD_CALL_COUNT` — see that item's
/// own doc for why a per-test reset is unnecessary under `cargo nextest`.
#[cfg(test)]
pub(crate) static DECODE_CALL_COUNT: std::sync::atomic::AtomicUsize =
    std::sync::atomic::AtomicUsize::new(0);

/// Decode `events` (one partition's replayed events, paired with their
/// journal positions) into every typed table [`PartitionTables`] holds.
///
/// This decodes both cache misses and cache-tail additions. See the module
/// documentation. `trusted_signers` reaches the approval-family folds.
/// `handoff_trust` reaches the handoff fold (`#1124`). Each role verifies
/// against its own trust root.
///
/// # Errors
///
/// Returns [`QueryEngineError::Decode`] if any one table's Arrow batch
/// construction fails.
#[allow(
    clippy::too_many_lines,
    reason = "eighteen mechanical per-table decode-then-map-err blocks, one per typed table \
              this partition builds; unlike register_typed_journal_tables (which iterates one \
              uniform (name, schema_fn, accessor) recipe over JOURNAL_TABLE_REGISTRATIONS, \
              crate::engine::registration), each table's own decode fn here has a distinct \
              per-kind signature (trusted_signers threaded through payments/approvals/\
              grant_replays/routine_lifecycle/wallet_link_lifecycle, message_content split \
              into two outputs, ...), so this cannot fold into the same table-driven loop \
              without losing that per-kind variation"
)]
pub(crate) fn decode_partition_tables(
    partition: &str,
    events: &[(u64, Event)],
    trusted_signers: &[Vec<u8>],
    handoff_trust: &RoleTrustSet<HandoffRole>,
) -> Result<PartitionTables, QueryEngineError> {
    #[cfg(test)]
    DECODE_CALL_COUNT.fetch_add(1, std::sync::atomic::Ordering::SeqCst);

    // `#743` under D5: a turn that paused recorded a content-free marker, and
    // the redacted views (`MESSAGES_REDACTED_VIEW_SQL`,
    // `TOOL_CALLS_REDACTED_VIEW_SQL`) filter on `internal_only = false`. The
    // flag has to be right in the rows those views read, so the marker is
    // applied here — the one place a partition's events become tables, so
    // every scope and both views get the same answer.
    //
    // The copy is taken only when a marker is actually present, which is the
    // rare case: a turn waiting on a person.
    let withheld = polyc_facts::withheld_turn_ids_positioned(events);
    let owned;
    let events = if withheld.is_empty() {
        events
    } else {
        owned = {
            let mut copy = events.to_vec();
            polyc_facts::withhold_paused_turn_text_positioned(&mut copy, &withheld);
            copy
        };
        &owned
    };

    let events_raw =
        decode::events_batch(partition, events).map_err(|source| QueryEngineError::Decode {
            table: EVENTS_RAW_TABLE,
            source,
        })?;

    let usage_raw =
        decode::usage::decode_usage_batch(&decode::usage::decode_usage_events(partition, events))
            .map_err(|source| QueryEngineError::Decode {
            table: USAGE_RAW_TABLE,
            source,
        })?;

    let model_call_raw = decode::model_call::decode_model_call_batch(
        &decode::model_call::decode_model_call_events(partition, events),
    )
    .map_err(|source| QueryEngineError::Decode {
        table: MODEL_CALL_RAW_TABLE,
        source,
    })?;

    let attribution_raw = decode::attribution::decode_attribution_batch(
        &decode::attribution::decode_attribution_events(partition, events),
    )
    .map_err(|source| QueryEngineError::Decode {
        table: ATTRIBUTION_RAW_TABLE,
        source,
    })?;

    let turn_failed_raw = decode::turn_failed::decode_turn_failed_batch(
        &decode::turn_failed::decode_turn_failed_events(partition, events),
    )
    .map_err(|source| QueryEngineError::Decode {
        table: TURN_FAILED_RAW_TABLE,
        source,
    })?;

    let turn_dispatch_raw = decode::turn_dispatch::decode_turn_dispatch_batch(
        &decode::turn_dispatch::decode_turn_dispatch_events(partition, events),
    )
    .map_err(|source| QueryEngineError::Decode {
        table: TURN_DISPATCH_RAW_TABLE,
        source,
    })?;

    let payments_raw = decode::payments::decode_payments_batch(
        &decode::payments::decode_payments_events(partition, events, trusted_signers),
    )
    .map_err(|source| QueryEngineError::Decode {
        table: PAYMENTS_RAW_TABLE,
        source,
    })?;

    let refusals_raw = decode::refusals::decode_refusals_batch(
        &decode::refusals::decode_refusals_events(partition, events, trusted_signers),
    )
    .map_err(|source| QueryEngineError::Decode {
        table: REFUSALS_RAW_TABLE,
        source,
    })?;

    let wallet_link_lifecycle_raw =
        decode::wallet_link_lifecycle::decode_wallet_link_lifecycle_batch(
            &decode::wallet_link_lifecycle::decode_wallet_link_lifecycle_events(
                partition,
                events,
                trusted_signers,
            ),
        )
        .map_err(|source| QueryEngineError::Decode {
            table: WALLET_LINK_LIFECYCLE_RAW_TABLE,
            source,
        })?;

    let approvals_raw = decode::approvals::decode_approvals_batch(
        &decode::approvals::decode_approvals_events(partition, events, trusted_signers),
    )
    .map_err(|source| QueryEngineError::Decode {
        table: APPROVALS_RAW_TABLE,
        source,
    })?;

    let handoffs_raw = decode::handoffs::decode_handoffs_batch(
        &decode::handoffs::decode_handoffs_events(partition, events, handoff_trust),
    )
    .map_err(|source| QueryEngineError::Decode {
        table: HANDOFFS_RAW_TABLE,
        source,
    })?;

    let grant_replays_raw = decode::grant_replays::decode_grant_replays_batch(
        &decode::grant_replays::decode_grant_replays_events(partition, events, trusted_signers),
    )
    .map_err(|source| QueryEngineError::Decode {
        table: GRANT_REPLAYS_RAW_TABLE,
        source,
    })?;

    let summary_raw = decode::summary::decode_summary_batch(
        &decode::summary::decode_summary_events(partition, events),
    )
    .map_err(|source| QueryEngineError::Decode {
        table: SUMMARY_RAW_TABLE,
        source,
    })?;

    let fires_raw =
        decode::fires::decode_fires_batch(&decode::fires::decode_fires_events(partition, events))
            .map_err(|source| QueryEngineError::Decode {
            table: FIRES_RAW_TABLE,
            source,
        })?;

    let routine_lifecycle_raw = decode::routine_lifecycle::decode_routine_lifecycle_batch(
        &decode::routine_lifecycle::decode_routine_lifecycle_events(
            partition,
            events,
            trusted_signers,
        ),
    )
    .map_err(|source| QueryEngineError::Decode {
        table: ROUTINE_LIFECYCLE_RAW_TABLE,
        source,
    })?;

    let routine_setup_raw = decode::routine_setup::decode_routine_setup_batch(
        &decode::routine_setup::decode_routine_setup_events(partition, events),
    )
    .map_err(|source| QueryEngineError::Decode {
        table: ROUTINE_SETUP_RAW_TABLE,
        source,
    })?;

    let mut message_rows = decode::message_content::messages::MessageRows::default();
    let mut tool_call_rows = decode::message_content::tool_calls::ToolCallRows::default();
    decode::message_content::decode_message_content_events(
        partition,
        events,
        &mut message_rows,
        &mut tool_call_rows,
    );
    let messages_raw = decode::message_content::messages::decode_messages_batch(&message_rows)
        .map_err(|source| QueryEngineError::Decode {
            table: MESSAGES_RAW_TABLE,
            source,
        })?;
    let tool_calls_raw = decode::message_content::tool_calls::decode_tool_calls_batch(
        &tool_call_rows,
    )
    .map_err(|source| QueryEngineError::Decode {
        table: TOOL_CALLS_RAW_TABLE,
        source,
    })?;

    Ok(PartitionTables {
        events_raw,
        usage_raw,
        model_call_raw,
        attribution_raw,
        turn_failed_raw,
        turn_dispatch_raw,
        payments_raw,
        refusals_raw,
        wallet_link_lifecycle_raw,
        approvals_raw,
        handoffs_raw,
        grant_replays_raw,
        summary_raw,
        fires_raw,
        routine_lifecycle_raw,
        routine_setup_raw,
        messages_raw,
        tool_calls_raw,
    })
}

#[cfg(test)]
mod tests {
    use polyc_eventlog::Event;
    use polyc_proto::kinds;
    use uuid::Uuid;

    use super::*;

    /// The fixture handoff trust these partition-shape tests decode under.
    /// None of them writes a handoff event, so the set only has to exist.
    fn handoff_trust() -> RoleTrustSet<HandoffRole> {
        RoleTrustSet::current(&polyc_crypto::signing_role::HandoffSigner::from_seed(700))
    }

    /// An empty partition decodes cleanly to eighteen zero-row tables — the
    /// baseline [`decode_partition_tables`]'s callers (a brand-new cache
    /// entry, or a partition nobody has ever written to) rely on.
    #[test]
    fn empty_partition_decodes_to_empty_tables() {
        let tables =
            decode_partition_tables("conv-empty", &[], &[], &handoff_trust()).expect("decode");
        assert_eq!(tables.events_raw.num_rows(), 0);
        assert_eq!(tables.usage_raw.num_rows(), 0);
        assert_eq!(tables.messages_raw.num_rows(), 0);
        assert_eq!(tables.tool_calls_raw.num_rows(), 0);
    }

    /// [`PartitionTables::concat`] appends a tail's rows onto a base's rows,
    /// per table — the mechanic a cache tail-hit relies on to grow its
    /// cached state without re-decoding the base rows.
    #[test]
    fn concat_appends_tail_rows_onto_base_rows() {
        let turn = Uuid::now_v7();
        let base = decode_partition_tables(
            "conv-tail",
            &[(
                0,
                Event::new(kinds::tagged(kinds::TURN_START, &turn), Vec::new()),
            )],
            &[],
            &handoff_trust(),
        )
        .expect("decode base");
        let tail = decode_partition_tables(
            "conv-tail",
            &[(
                1,
                Event::new(kinds::tagged(kinds::TURN_COMPLETE, &turn), Vec::new()),
            )],
            &[],
            &handoff_trust(),
        )
        .expect("decode tail");

        let merged = base.concat(&tail).expect("concat");
        assert_eq!(merged.events_raw.num_rows(), 2);
    }

    /// [`PartitionTables::memory_bytes`] is positive for a non-empty
    /// partition and reflects growth after a concat — the signal the
    /// eviction LRU's total-bytes cap sums across every cached partition.
    #[test]
    fn memory_bytes_grows_after_concat() {
        let turn = Uuid::now_v7();
        let base = decode_partition_tables(
            "conv-mem",
            &[(
                0,
                Event::new(kinds::tagged(kinds::TURN_START, &turn), Vec::new()),
            )],
            &[],
            &handoff_trust(),
        )
        .expect("decode base");
        let tail = decode_partition_tables(
            "conv-mem",
            &[(
                1,
                Event::new(kinds::tagged(kinds::TURN_COMPLETE, &turn), Vec::new()),
            )],
            &[],
            &handoff_trust(),
        )
        .expect("decode tail");
        let merged = base.concat(&tail).expect("concat");

        assert!(merged.memory_bytes() >= base.memory_bytes());
    }
}