polyc-query 2026.8.3

//! Table/view registration helpers for [`super::QueryEngine::build`].
//!
//! Pure decomposition — extracted out of the `engine` module once its
//! registration cluster pushed `engine.rs` past the repo's file-length
//! threshold, no behavior change; see the `engine` module docs for the
//! catalog-scoping rules these helpers enforce.

use std::sync::Arc;

use arrow::datatypes::SchemaRef;
use arrow::record_batch::RecordBatch;
use datafusion::datasource::MemTable;
use datafusion::error::DataFusionError;
use datafusion::execution::context::SessionContext;

use crate::decode;
use crate::decode::persona::{
    ParticipationRow, PersonaCredentialRow, PersonaIdentityRow, PersonaRow, PersonaSpendPolicyRow,
    PersonaUsageRow, PersonaWalletRow,
};
use crate::views::{
    APPROVALS_REDACTED_VIEW_SQL, APPROVALS_VIEW_SQL, ATTRIBUTION_REDACTED_VIEW_SQL,
    ATTRIBUTION_VIEW_SQL, FIRES_OWNED_VIEW_SQL, FIRES_VIEW_SQL, GRANT_REPLAYS_REDACTED_VIEW_SQL,
    GRANT_REPLAYS_VIEW_SQL, HANDOFFS_REDACTED_VIEW_SQL, HANDOFFS_VIEW_SQL,
    MESSAGES_REDACTED_VIEW_SQL, MESSAGES_VIEW_SQL, MODEL_CALL_VIEW_SQL, PAYMENTS_REDACTED_VIEW_SQL,
    PAYMENTS_VIEW_SQL, REFUSALS_REDACTED_VIEW_SQL, REFUSALS_VIEW_SQL,
    ROUTINE_ACTIVE_GRANTS_VIEW_SQL, ROUTINE_GRANTS_VIEW_SQL, ROUTINE_LIFECYCLE_VIEW_SQL,
    ROUTINE_OVERVIEW_VIEW_SQL, ROUTINE_SETUP_OWNED_VIEW_SQL, ROUTINE_SETUP_VIEW_SQL,
    SUMMARY_VIEW_SQL, TOOL_CALLS_REDACTED_VIEW_SQL, TOOL_CALLS_VIEW_SQL, TURN_DISPATCH_VIEW_SQL,
    TURN_FAILED_VIEW_SQL, USAGE_VIEW_SQL, WALLET_LINK_LIFECYCLE_REDACTED_VIEW_SQL,
    WALLET_LINK_LIFECYCLE_VIEW_SQL,
};

#[allow(unused_imports)] // only used by intra-doc links above/below
use super::QueryEngine;
use super::tables::{
    APPROVALS_RAW_TABLE, APPROVALS_TABLE, ATTRIBUTION_RAW_TABLE, ATTRIBUTION_TABLE,
    DASHBOARD_TABLE, FIRES_RAW_TABLE, FIRES_TABLE, GRANT_REPLAYS_RAW_TABLE, GRANT_REPLAYS_TABLE,
    HANDOFFS_RAW_TABLE, HANDOFFS_TABLE, MESSAGES_RAW_TABLE, MESSAGES_TABLE, MODEL_CALL_RAW_TABLE,
    MODEL_CALL_TABLE, PARTICIPATIONS_TABLE, PAYMENTS_RAW_TABLE, PAYMENTS_TABLE,
    PERSONA_CREDENTIALS_TABLE, PERSONA_IDENTITIES_TABLE, PERSONA_SPEND_POLICIES_TABLE,
    PERSONA_USAGE_TABLE, PERSONA_WALLETS_TABLE, PERSONAS_TABLE, REFUSALS_RAW_TABLE, REFUSALS_TABLE,
    ROUTINE_ACTIVE_GRANTS_TABLE, ROUTINE_GRANTS_TABLE, ROUTINE_LIFECYCLE_RAW_TABLE,
    ROUTINE_LIFECYCLE_TABLE, ROUTINE_OVERVIEW_TABLE, ROUTINE_SETUP_RAW_TABLE, ROUTINE_SETUP_TABLE,
    ROUTINES_TABLE, SUMMARY_RAW_TABLE, SUMMARY_TABLE, TOOL_CALLS_RAW_TABLE, TOOL_CALLS_TABLE,
    TURN_DISPATCH_RAW_TABLE, TURN_DISPATCH_TABLE, TURN_FAILED_RAW_TABLE, TURN_FAILED_TABLE,
    USAGE_RAW_TABLE, USAGE_TABLE, WALLET_LINK_LIFECYCLE_RAW_TABLE, WALLET_LINK_LIFECYCLE_TABLE,
};
use super::{PartitionTables, QueryEngineError, ReferenceData};
use crate::session::QueryScope;

/// Which of the three registration postures a [`super::QueryEngine::build_from_tables`]
/// call takes for its scope-dependent views (issue #1882's review, finding
/// 5) — replaces the `(is_fleet, owner_scoped_routines)` boolean pair this
/// crate carried before, whose one illegal combination (`is_fleet &&
/// owner_scoped_routines`) had to be defensively re-derived away on every
/// call (`owner_scoped_routines = !is_fleet && register_owner_scoped_routines`)
/// rather than being simply unrepresentable, with one three-variant enum.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(super) enum RegistrationScope {
    /// A verified admin session: every row, every column, unfiltered.
    Fleet,
    /// A verified persona-scoped session that owns the routines
    /// `reference.routines` already carries (issue #1882) — `routines`/
    /// `fires` register filtered to that persona's own rows; every other
    /// scope-dependent view gets its REDACTED body, the same as [`Self::Grant`].
    Owner,
    /// A conversation grant, or a persona-scoped session with no routine
    /// ownership context — every scope-dependent view gets its REDACTED
    /// body; `routines`/`fires` register for neither.
    Grant,
}

impl RegistrationScope {
    /// Derive the posture from `scope` plus `register_owner_scoped_routines`
    /// (`true` iff this build is for a verified persona-scoped session that
    /// owns routines — see
    /// [`super::QueryEngine::build_from_tables`]'s own doc). Ignored when
    /// `scope` is [`QueryScope::Fleet`] — that scope is always [`Self::Fleet`]
    /// regardless of this argument, the same posture the old boolean pair
    /// enforced by re-deriving `owner_scoped_routines` away, now enforced by
    /// this match simply never reaching [`Self::Owner`] for that arm.
    pub(super) const fn new(scope: &QueryScope, register_owner_scoped_routines: bool) -> Self {
        match scope {
            QueryScope::Fleet => Self::Fleet,
            QueryScope::Conversations(_) if register_owner_scoped_routines => Self::Owner,
            QueryScope::Conversations(_) => Self::Grant,
        }
    }

    /// Whether every scope-dependent view should get its FULL, unredacted
    /// body — `true` for [`Self::Fleet`] only.
    pub(super) const fn is_fleet(self) -> bool {
        matches!(self, Self::Fleet)
    }

    /// Whether `routines`/`fires` should register filtered to this session's
    /// own persona — `true` for [`Self::Owner`] only.
    pub(super) const fn is_owner(self) -> bool {
        matches!(self, Self::Owner)
    }
}

/// Wrap `batches` (one per caller-supplied partition, in order) as a
/// [`MemTable`] over `schema`, one physical partition per input batch — the
/// tables the caller registers this way naturally union every supplied
/// partition. A caller supplying zero partitions still gets a valid, empty
/// single-partition table: [`MemTable::try_new`] rejects an empty `Vec` of
/// partitions outright, so an empty `batches` maps to one empty partition
/// rather than none.
fn mem_table(schema: SchemaRef, batches: Vec<RecordBatch>) -> Result<MemTable, DataFusionError> {
    let partitions = if batches.is_empty() {
        vec![vec![]]
    } else {
        batches.into_iter().map(|batch| vec![batch]).collect()
    };
    MemTable::try_new(schema, partitions)
}

/// Build a [`MemTable`] over `schema`/`batches` (via [`mem_table`]) and
/// register it as `name` on `ctx` — the `mem_table` + `register_table` pair
/// every typed-table registration in [`QueryEngine::build`] repeats
/// identically, differing only in name/schema/batches. Decoding each
/// table's own rows stays at the call site (that step differs per table);
/// this helper starts only once a caller already has the decoded batches in
/// hand.
pub(super) fn register_typed_table(
    ctx: &SessionContext,
    name: &str,
    schema: SchemaRef,
    batches: Vec<RecordBatch>,
) -> Result<(), QueryEngineError> {
    let table = mem_table(schema, batches)?;
    ctx.register_table(name, Arc::new(table))?;
    Ok(())
}

/// One typed table's registration recipe: its raw-table name, its schema
/// constructor, and how to pull its already-decoded batch out of one
/// partition's [`PartitionTables`] — what
/// [`register_typed_journal_tables`]/[`register_message_content_tables`]
/// iterate over instead of repeating an identical `register_typed_table`
/// call by hand, once per table. Mirrors `PartitionTables`'s own private
/// `fields` method (its `(name, &RecordBatch)` enumeration, used by
/// [`PartitionTables::memory_bytes`] and `crate::cache`'s concatenation
/// step), extended here with a schema constructor since registration needs
/// one and those callers don't; the accessor is a plain field-projecting
/// `fn` (no captures), so it coerces from a closure literal with no boxing.
type TypedTableRegistration = (
    &'static str,
    fn() -> SchemaRef,
    fn(&PartitionTables) -> &RecordBatch,
);

/// Every per-journal-kind typed table [`register_typed_journal_tables`]
/// registers, in order — see that function's own doc.
const JOURNAL_TABLE_REGISTRATIONS: [TypedTableRegistration; 15] = [
    (USAGE_RAW_TABLE, decode::usage::schema, |t| &t.usage_raw),
    (MODEL_CALL_RAW_TABLE, decode::model_call::schema, |t| {
        &t.model_call_raw
    }),
    (ATTRIBUTION_RAW_TABLE, decode::attribution::schema, |t| {
        &t.attribution_raw
    }),
    (TURN_FAILED_RAW_TABLE, decode::turn_failed::schema, |t| {
        &t.turn_failed_raw
    }),
    (
        TURN_DISPATCH_RAW_TABLE,
        decode::turn_dispatch::schema,
        |t| &t.turn_dispatch_raw,
    ),
    (PAYMENTS_RAW_TABLE, decode::payments::schema, |t| {
        &t.payments_raw
    }),
    (REFUSALS_RAW_TABLE, decode::refusals::schema, |t| {
        &t.refusals_raw
    }),
    (
        WALLET_LINK_LIFECYCLE_RAW_TABLE,
        decode::wallet_link_lifecycle::schema,
        |t| &t.wallet_link_lifecycle_raw,
    ),
    (APPROVALS_RAW_TABLE, decode::approvals::schema, |t| {
        &t.approvals_raw
    }),
    (HANDOFFS_RAW_TABLE, decode::handoffs::schema, |t| {
        &t.handoffs_raw
    }),
    (
        GRANT_REPLAYS_RAW_TABLE,
        decode::grant_replays::schema,
        |t| &t.grant_replays_raw,
    ),
    (SUMMARY_RAW_TABLE, decode::summary::schema, |t| {
        &t.summary_raw
    }),
    (FIRES_RAW_TABLE, decode::fires::schema, |t| &t.fires_raw),
    (
        ROUTINE_LIFECYCLE_RAW_TABLE,
        decode::routine_lifecycle::schema,
        |t| &t.routine_lifecycle_raw,
    ),
    (
        ROUTINE_SETUP_RAW_TABLE,
        decode::routine_setup::schema,
        |t| &t.routine_setup_raw,
    ),
];

/// The two tables [`register_message_content_tables`] registers, in order —
/// see that function's own doc.
const MESSAGE_CONTENT_TABLE_REGISTRATIONS: [TypedTableRegistration; 2] = [
    (
        MESSAGES_RAW_TABLE,
        decode::message_content::messages_schema,
        |t| &t.messages_raw,
    ),
    (
        TOOL_CALLS_RAW_TABLE,
        decode::message_content::tool_calls_schema,
        |t| &t.tool_calls_raw,
    ),
];

/// Register `recipe` (a `(name, schema_fn, accessor)` triple) on `ctx`, one
/// physical partition per entry of `partition_tables` — the transposition
/// step [`register_typed_journal_tables`]/[`register_message_content_tables`]
/// both loop over [`JOURNAL_TABLE_REGISTRATIONS`]/
/// [`MESSAGE_CONTENT_TABLE_REGISTRATIONS`] to run once per table.
fn register_typed_table_recipe(
    ctx: &SessionContext,
    recipe: TypedTableRegistration,
    partition_tables: &[PartitionTables],
) -> Result<(), QueryEngineError> {
    let (name, schema_fn, accessor) = recipe;
    register_typed_table(
        ctx,
        name,
        schema_fn(),
        partition_tables
            .iter()
            .map(|tables| accessor(tables).clone())
            .collect(),
    )
}

/// Register every per-journal-kind typed table — [`USAGE_RAW_TABLE`],
/// [`MODEL_CALL_RAW_TABLE`], [`ATTRIBUTION_RAW_TABLE`],
/// [`TURN_FAILED_RAW_TABLE`], [`TURN_DISPATCH_RAW_TABLE`],
/// [`PAYMENTS_RAW_TABLE`], [`REFUSALS_RAW_TABLE`], [`WALLET_LINK_LIFECYCLE_RAW_TABLE`],
/// [`APPROVALS_RAW_TABLE`], [`HANDOFFS_RAW_TABLE`],
/// [`GRANT_REPLAYS_RAW_TABLE`], [`SUMMARY_RAW_TABLE`], [`FIRES_RAW_TABLE`],
/// [`ROUTINE_LIFECYCLE_RAW_TABLE`], and [`ROUTINE_SETUP_RAW_TABLE`], in
/// that order — on `ctx`, one
/// physical partition per entry of `partition_tables`.
///
/// Decode already happened before this call — either freshly
/// ([`super::decode_partition_tables`], the [`QueryEngine::build`] wrapper's
/// own miss path) or served from `crate::cache`'s decode cache — so this
/// function does no decode work of its own; it only transposes each typed
/// table's already-built [`RecordBatch`] out of every partition's
/// [`PartitionTables`] and registers the per-partition `Vec` as one
/// [`MemTable`], driven by [`JOURNAL_TABLE_REGISTRATIONS`] rather than twelve
/// near-identical calls written out by hand.
///
/// `usage`/`model_call`/`turn_failed`/`turn_dispatch`/`summary` register
/// under their `_raw` names here, not their public table names — unlike
/// before QRY-1's fix, each now needs [`create_scope_dependent_views`]
/// (`usage`/`model_call`/`turn_failed`/`turn_dispatch`) or Fleet-only gating
/// (`summary`/`fires`/`routine_lifecycle`) built on top, the same raw/view
/// split `ATTRIBUTION_RAW_TABLE`/`ATTRIBUTION_TABLE` already used; see
/// `crate::views`'s module docs' "Committed-turn filter invariant" section.
/// `turn_failed` is no longer an exception — a prior review pass exempted it
/// on a false rationale (a failed turn's `turn_id` never carries
/// `turn_complete` by construction); see [`TURN_FAILED_TABLE`]'s doc for why
/// that rationale does not hold and `turn_failed` now needs the identical
/// raw/view split every other committed-turn-filtered table uses.
pub(super) fn register_typed_journal_tables(
    ctx: &SessionContext,
    partition_tables: &[PartitionTables],
) -> Result<(), QueryEngineError> {
    for recipe in JOURNAL_TABLE_REGISTRATIONS {
        register_typed_table_recipe(ctx, recipe, partition_tables)?;
    }
    Ok(())
}

/// Register `messages_raw`/`tool_calls_raw` on `ctx`, one physical partition
/// per entry of `partition_tables` — the same already-decoded transposition
/// [`register_typed_journal_tables`] does, split out only because this pair
/// shares one field slot each ([`PartitionTables::messages_raw`]/
/// [`PartitionTables::tool_calls_raw`]) rather than [`register_typed_journal_tables`]'s
/// flat one-field-per-table shape.
pub(super) fn register_message_content_tables(
    ctx: &SessionContext,
    partition_tables: &[PartitionTables],
) -> Result<(), QueryEngineError> {
    for recipe in MESSAGE_CONTENT_TABLE_REGISTRATIONS {
        register_typed_table_recipe(ctx, recipe, partition_tables)?;
    }
    Ok(())
}

/// Build every `CREATE VIEW` a typed table needs on top of its
/// already-registered raw table — [`ATTRIBUTION_TABLE`], [`PAYMENTS_TABLE`],
/// [`MESSAGES_TABLE`], [`TOOL_CALLS_TABLE`], [`APPROVALS_TABLE`],
/// [`HANDOFFS_TABLE`], [`GRANT_REPLAYS_TABLE`] (each scope-DEPENDENT: full or
/// redacted SQL body based on `is_fleet`), plus [`USAGE_TABLE`]/
/// [`MODEL_CALL_TABLE`]/[`TURN_FAILED_TABLE`] (one committed-turn-filtered
/// body for every scope — no Fleet-only column on any of the three) and
/// [`SUMMARY_TABLE`]/[`FIRES_TABLE`] (each built ONLY when `is_fleet`, per
/// those constants' own Fleet-only docs).
///
/// Extracted out of [`QueryEngine::build`] (which calls this for every
/// scope, right after registering the raw tables and the always-identical
/// `events` view) once the fifth and sixth typed-table registrations
/// (`messages`/`tool_calls`, this crate's fold-coupled content tables)
/// pushed `build` past `clippy::too_many_lines`, the same reason
/// [`register_reference_tables`] and [`register_decoded_table`] were
/// themselves extracted earlier. See the module docs' "Identity redaction
/// invariant", "Payment signer-key redaction invariant",
/// "Message-content redaction invariant" sections, and
/// [`crate::decode::approvals`]'s own "Column selection and redaction"
/// section, plus [`crate::decode::handoffs`]'s own "Redaction" section and
/// [`crate::decode::grant_replays`]'s own "Column selection and redaction"
/// section, for why each pair of bodies must never share a column list
/// (`attribution`/`payments`/`approvals`/`handoffs`/`grant_replays`) or must
/// share the identical column list filtered by row
/// (`messages`/`tool_calls`); and `crate::views`'s module docs' "Committed-
/// turn filter invariant" section for the semijoin every view body here now
/// adds against `events`.
///
/// `scope`'s [`RegistrationScope::Owner`] arm (issue #1882) requires the
/// caller already registered [`ROUTINES_TABLE`] against this same `ctx` (see
/// [`super::QueryEngine::build_from_tables`]'s own doc for the
/// registration-ordering requirement this depends on) — when set, `fires`
/// is built here too, via [`FIRES_OWNED_VIEW_SQL`]'s owner-filtering join.
#[allow(clippy::too_many_lines)] // every typed table's view creation, one mechanical block each
pub(super) async fn create_scope_dependent_views(
    ctx: &SessionContext,
    scope: RegistrationScope,
) -> Result<(), QueryEngineError> {
    let is_fleet = scope.is_fleet();
    let attribution_view_body = if is_fleet {
        ATTRIBUTION_VIEW_SQL
    } else {
        ATTRIBUTION_REDACTED_VIEW_SQL
    };
    let create_attribution_view_sql =
        format!("CREATE VIEW {ATTRIBUTION_TABLE} AS {attribution_view_body}");
    ctx.sql(&create_attribution_view_sql)
        .await?
        .collect()
        .await?;

    let payments_view_body = if is_fleet {
        PAYMENTS_VIEW_SQL
    } else {
        PAYMENTS_REDACTED_VIEW_SQL
    };
    let create_payments_view_sql = format!("CREATE VIEW {PAYMENTS_TABLE} AS {payments_view_body}");
    ctx.sql(&create_payments_view_sql).await?.collect().await?;

    // `refusals` (`#2090`, INV-W5): the same raw/redacted split `payments`
    // uses above (`signer_public_key` is Fleet-only), but — deliberately —
    // no committed-turn filter; see [`REFUSALS_TABLE`]'s own doc for why.
    let refusals_view_body = if is_fleet {
        REFUSALS_VIEW_SQL
    } else {
        REFUSALS_REDACTED_VIEW_SQL
    };
    let create_refusals_view_sql = format!("CREATE VIEW {REFUSALS_TABLE} AS {refusals_view_body}");
    ctx.sql(&create_refusals_view_sql).await?.collect().await?;

    // `wallet_link_lifecycle` (`#2123`): the same raw/redacted split
    // `refusals` uses above (`signer_public_key` is Fleet-only), and — like
    // `refusals` — no committed-turn filter, but for a THIRD, distinct
    // reason (structurally inapplicable, not deliberately skipped); see
    // [`WALLET_LINK_LIFECYCLE_TABLE`]'s own doc for why.
    let wallet_link_lifecycle_view_body = if is_fleet {
        WALLET_LINK_LIFECYCLE_VIEW_SQL
    } else {
        WALLET_LINK_LIFECYCLE_REDACTED_VIEW_SQL
    };
    let create_wallet_link_lifecycle_view_sql =
        format!("CREATE VIEW {WALLET_LINK_LIFECYCLE_TABLE} AS {wallet_link_lifecycle_view_body}");
    ctx.sql(&create_wallet_link_lifecycle_view_sql)
        .await?
        .collect()
        .await?;

    let messages_view_body = if is_fleet {
        MESSAGES_VIEW_SQL
    } else {
        MESSAGES_REDACTED_VIEW_SQL
    };
    let create_messages_view_sql = format!("CREATE VIEW {MESSAGES_TABLE} AS {messages_view_body}");
    ctx.sql(&create_messages_view_sql).await?.collect().await?;

    let tool_calls_view_body = if is_fleet {
        TOOL_CALLS_VIEW_SQL
    } else {
        TOOL_CALLS_REDACTED_VIEW_SQL
    };
    let create_tool_calls_view_sql =
        format!("CREATE VIEW {TOOL_CALLS_TABLE} AS {tool_calls_view_body}");
    ctx.sql(&create_tool_calls_view_sql)
        .await?
        .collect()
        .await?;

    let approvals_view_body = if is_fleet {
        APPROVALS_VIEW_SQL
    } else {
        APPROVALS_REDACTED_VIEW_SQL
    };
    let create_approvals_view_sql =
        format!("CREATE VIEW {APPROVALS_TABLE} AS {approvals_view_body}");
    ctx.sql(&create_approvals_view_sql).await?.collect().await?;

    let handoffs_view_body = if is_fleet {
        HANDOFFS_VIEW_SQL
    } else {
        HANDOFFS_REDACTED_VIEW_SQL
    };
    let create_handoffs_view_sql = format!("CREATE VIEW {HANDOFFS_TABLE} AS {handoffs_view_body}");
    ctx.sql(&create_handoffs_view_sql).await?.collect().await?;

    let grant_replays_view_body = if is_fleet {
        GRANT_REPLAYS_VIEW_SQL
    } else {
        GRANT_REPLAYS_REDACTED_VIEW_SQL
    };
    let create_grant_replays_view_sql =
        format!("CREATE VIEW {GRANT_REPLAYS_TABLE} AS {grant_replays_view_body}");
    ctx.sql(&create_grant_replays_view_sql)
        .await?
        .collect()
        .await?;

    // `usage`/`model_call`: one view body for every scope (no Fleet-only
    // column on either table), but still a `CREATE VIEW` — not a second
    // registration under the final name — because each now needs the
    // committed-turn filter [`crate::views::USAGE_VIEW_SQL`]/
    // [`crate::views::MODEL_CALL_VIEW_SQL`] applies against the
    // already-built `events` view; see `crate::views`'s module docs'
    // "Committed-turn filter invariant" section.
    let create_usage_view_sql = format!("CREATE VIEW {USAGE_TABLE} AS {USAGE_VIEW_SQL}");
    ctx.sql(&create_usage_view_sql).await?.collect().await?;

    let create_model_call_view_sql =
        format!("CREATE VIEW {MODEL_CALL_TABLE} AS {MODEL_CALL_VIEW_SQL}");
    ctx.sql(&create_model_call_view_sql)
        .await?
        .collect()
        .await?;

    // `turn_failed`: the identical one-view-body-for-every-scope treatment
    // `usage`/`model_call` just received above — no Fleet-only column on
    // this table either, and the committed-turn filter
    // [`crate::views::TURN_FAILED_VIEW_SQL`] applies is correct for it too;
    // see [`TURN_FAILED_TABLE`]'s doc for why the exemption a prior review
    // pass carved out here was wrong.
    let create_turn_failed_view_sql =
        format!("CREATE VIEW {TURN_FAILED_TABLE} AS {TURN_FAILED_VIEW_SQL}");
    ctx.sql(&create_turn_failed_view_sql)
        .await?
        .collect()
        .await?;

    // `turn_dispatch`: the identical one-view-body-for-every-scope treatment
    // `usage`/`model_call`/`turn_failed` just received above (issue #1593) —
    // no external identity on this table either.
    let create_turn_dispatch_view_sql =
        format!("CREATE VIEW {TURN_DISPATCH_TABLE} AS {TURN_DISPATCH_VIEW_SQL}");
    ctx.sql(&create_turn_dispatch_view_sql)
        .await?
        .collect()
        .await?;

    // `summary`: Fleet-only — the view is never created at all for any
    // other scope, so the name itself is unresolvable, the same posture
    // [`register_reference_tables`] uses for `personas`/`participations`
    // rather than the "resolvable but redacted" posture every view above
    // uses; see [`SUMMARY_TABLE`]'s doc and `crate::views`'s module docs'
    // "Committed-turn filter invariant" section for why.
    if is_fleet {
        let create_summary_view_sql = format!("CREATE VIEW {SUMMARY_TABLE} AS {SUMMARY_VIEW_SQL}");
        ctx.sql(&create_summary_view_sql).await?.collect().await?;
    }

    // `fires` (issue #1882): [`RegistrationScope::Fleet`] gets the unfiltered
    // body, [`RegistrationScope::Owner`] gets the owner-filtering join body
    // instead, against the `routines` table the caller already registered
    // before calling this function (see this function's own doc), and
    // [`RegistrationScope::Grant`] gets neither — the match is exhaustive, so
    // "both" and "neither of the first two" are both compiler-enforced,
    // rather than relying on the old `is_fleet`/`owner_scoped_routines`
    // boolean pair's own if/else-if ordering to keep them mutually exclusive.
    match scope {
        RegistrationScope::Fleet => {
            let create_fires_view_sql = format!("CREATE VIEW {FIRES_TABLE} AS {FIRES_VIEW_SQL}");
            ctx.sql(&create_fires_view_sql).await?.collect().await?;
        }
        RegistrationScope::Owner => {
            let create_fires_view_sql =
                format!("CREATE VIEW {FIRES_TABLE} AS {FIRES_OWNED_VIEW_SQL}");
            ctx.sql(&create_fires_view_sql).await?.collect().await?;
        }
        RegistrationScope::Grant => {}
    }

    // `routine_lifecycle`: the identical Fleet-only-gate treatment `fires`
    // just received above (issue #1593) — this table's one source partition
    // (`"routine-scheduler"`) is itself admitted into replay only for a
    // Fleet scope.
    if is_fleet {
        let create_routine_lifecycle_view_sql =
            format!("CREATE VIEW {ROUTINE_LIFECYCLE_TABLE} AS {ROUTINE_LIFECYCLE_VIEW_SQL}");
        ctx.sql(&create_routine_lifecycle_view_sql)
            .await?
            .collect()
            .await?;
    }

    // `routine_setup` : the identical Fleet/Owner/Grant treatment
    // `fires` receives above — unfiltered for Fleet, uid-join-filtered for
    // Owner (against the `routines` table the caller already registered),
    // absent for Grant.
    match scope {
        RegistrationScope::Fleet => {
            let sql = format!("CREATE VIEW {ROUTINE_SETUP_TABLE} AS {ROUTINE_SETUP_VIEW_SQL}");
            ctx.sql(&sql).await?.collect().await?;
        }
        RegistrationScope::Owner => {
            let sql =
                format!("CREATE VIEW {ROUTINE_SETUP_TABLE} AS {ROUTINE_SETUP_OWNED_VIEW_SQL}");
            ctx.sql(&sql).await?.collect().await?;
        }
        RegistrationScope::Grant => {}
    }

    // `routine_grants`/`routine_active_grants`/`routine_overview` :
    // ONE body each for Fleet and Owner — the join against the
    // already-scope-filtered `routines` table IS the row filter, so the two
    // scopes share it (see [`crate::views::ROUTINE_GRANTS_VIEW_SQL`]'s doc) —
    // and neither for Grant. Created in dependency order: `routine_grants`
    // reads `approvals_raw` + `routines`, `routine_active_grants` reads
    // `routine_grants`, and `routine_overview` reads both plus `fires_raw`/
    // `routine_setup_raw` — every referenced provider is baked into the
    // stored view plan here, BEFORE the non-Fleet raw-table deregistration
    // runs (the same mechanism that keeps the `fires` owner view alive after
    // `fires_raw` deregisters).
    if matches!(scope, RegistrationScope::Fleet | RegistrationScope::Owner) {
        let sql = format!("CREATE VIEW {ROUTINE_GRANTS_TABLE} AS {ROUTINE_GRANTS_VIEW_SQL}");
        ctx.sql(&sql).await?.collect().await?;
        let sql = format!(
            "CREATE VIEW {ROUTINE_ACTIVE_GRANTS_TABLE} AS {ROUTINE_ACTIVE_GRANTS_VIEW_SQL}"
        );
        ctx.sql(&sql).await?.collect().await?;
        let sql = format!("CREATE VIEW {ROUTINE_OVERVIEW_TABLE} AS {ROUTINE_OVERVIEW_VIEW_SQL}");
        ctx.sql(&sql).await?.collect().await?;
    }

    Ok(())
}

/// Decode `reference`'s persona/participation/wallet/spend-policy/
/// credential/usage-rollup/dashboard records and register
/// [`PERSONAS_TABLE`]/[`PARTICIPATIONS_TABLE`]/[`PERSONA_IDENTITIES_TABLE`]/
/// [`PERSONA_WALLETS_TABLE`]/[`PERSONA_SPEND_POLICIES_TABLE`]/
/// [`PERSONA_CREDENTIALS_TABLE`]/[`PERSONA_USAGE_TABLE`]/[`DASHBOARD_TABLE`]
/// on `ctx` — every Fleet-only reference table EXCEPT [`ROUTINES_TABLE`],
/// which [`register_routines_table`] now registers separately (issue #1882:
/// `routines` is no longer Fleet-only, so it needs registering at a
/// different POINT in [`super::QueryEngine::build_from_tables`]'s call
/// sequence than the tables that stay Fleet-only here — see that function's
/// own doc for the ordering requirement).
///
/// Extracted out of [`QueryEngine::build`] (which calls this only for
/// [`QueryScope::Fleet`] — see the module docs' "Reference data" section)
/// once a fourth typed-table registration (`turn_failed`, #1311) pushed
/// `build` past `clippy::too_many_lines`, the same reason
/// [`register_decoded_table`] itself was extracted for the third typed
/// table. The four #1578 tables (`persona_wallets`/`persona_spend_policies`/
/// `persona_credentials`/`persona_usage`) each register only the rows
/// `reference` actually carries — a candidate persona with no linked
/// wallet/no set spend policy/no enrolled credential/no usage rollup gets
/// no row in that table, mirroring `participations`' own "no tie, no row"
/// posture rather than a sentinel-filled row.
#[allow(clippy::too_many_lines)] // four more tables, each a decode+register pair; no shared shape to extract
pub(super) fn register_reference_tables(
    ctx: &SessionContext,
    reference: &ReferenceData,
) -> Result<(), QueryEngineError> {
    let persona_rows: Vec<PersonaRow> = reference.personas.iter().map(PersonaRow::from).collect();
    let persona_batch =
        decode::persona::decode_personas_batch(&persona_rows).map_err(|source| {
            QueryEngineError::Decode {
                table: PERSONAS_TABLE,
                source,
            }
        })?;
    register_typed_table(
        ctx,
        PERSONAS_TABLE,
        decode::persona::personas_schema(),
        vec![persona_batch],
    )?;

    let participation_rows: Vec<ParticipationRow> = reference
        .participations
        .iter()
        .map(|(persona_id, participation)| ParticipationRow::new(persona_id, participation))
        .collect();
    let participation_batch = decode::persona::decode_participations_batch(&participation_rows)
        .map_err(|source| QueryEngineError::Decode {
            table: PARTICIPATIONS_TABLE,
            source,
        })?;
    register_typed_table(
        ctx,
        PARTICIPATIONS_TABLE,
        decode::persona::participations_schema(),
        vec![participation_batch],
    )?;

    // No new `ReferenceData` field needed: each `PersonaProfile` already
    // carries its `identities` (see the module docs' "Reference data"
    // section), so this table projects data `personas` already decoded
    // rather than pulling a second host call.
    let persona_identity_rows: Vec<PersonaIdentityRow> = reference
        .personas
        .iter()
        .flat_map(|profile| {
            profile
                .identities
                .iter()
                .map(move |identity| PersonaIdentityRow::new(&profile.persona_id, identity))
        })
        .collect();
    let persona_identity_batch = decode::persona::decode_persona_identities_batch(
        &persona_identity_rows,
    )
    .map_err(|source| QueryEngineError::Decode {
        table: PERSONA_IDENTITIES_TABLE,
        source,
    })?;
    register_typed_table(
        ctx,
        PERSONA_IDENTITIES_TABLE,
        decode::persona::persona_identities_schema(),
        vec![persona_identity_batch],
    )?;

    let wallet_rows: Vec<PersonaWalletRow> = reference
        .wallets
        .iter()
        .map(|(persona_id, wallet)| PersonaWalletRow::new(persona_id, wallet))
        .collect();
    let wallet_batch =
        decode::persona::decode_persona_wallets_batch(&wallet_rows).map_err(|source| {
            QueryEngineError::Decode {
                table: PERSONA_WALLETS_TABLE,
                source,
            }
        })?;
    register_typed_table(
        ctx,
        PERSONA_WALLETS_TABLE,
        decode::persona::persona_wallets_schema(),
        vec![wallet_batch],
    )?;

    let spend_policy_rows: Vec<PersonaSpendPolicyRow> = reference
        .spend_policies
        .iter()
        .map(|(persona_id, policy)| PersonaSpendPolicyRow::new(persona_id, policy))
        .collect();
    let spend_policy_batch = decode::persona::decode_persona_spend_policies_batch(
        &spend_policy_rows,
    )
    .map_err(|source| QueryEngineError::Decode {
        table: PERSONA_SPEND_POLICIES_TABLE,
        source,
    })?;
    register_typed_table(
        ctx,
        PERSONA_SPEND_POLICIES_TABLE,
        decode::persona::persona_spend_policies_schema(),
        vec![spend_policy_batch],
    )?;

    let credential_rows: Vec<PersonaCredentialRow> = reference
        .credentials
        .iter()
        .map(|(persona_id, credential)| PersonaCredentialRow::new(persona_id, credential))
        .collect();
    let credential_batch = decode::persona::decode_persona_credentials_batch(&credential_rows)
        .map_err(|source| QueryEngineError::Decode {
            table: PERSONA_CREDENTIALS_TABLE,
            source,
        })?;
    register_typed_table(
        ctx,
        PERSONA_CREDENTIALS_TABLE,
        decode::persona::persona_credentials_schema(),
        vec![credential_batch],
    )?;

    let usage_rows: Vec<PersonaUsageRow> = reference
        .usage_rollups
        .iter()
        .map(|(persona_id, rollup)| PersonaUsageRow::new(persona_id, rollup))
        .collect();
    let usage_batch =
        decode::persona::decode_persona_usage_batch(&usage_rows).map_err(|source| {
            QueryEngineError::Decode {
                table: PERSONA_USAGE_TABLE,
                source,
            }
        })?;
    register_typed_table(
        ctx,
        PERSONA_USAGE_TABLE,
        decode::persona::persona_usage_schema(),
        vec![usage_batch],
    )?;

    // The dashboard maintained projection (#1584/#1585), decoded straight
    // from `reference.dashboard` — see `crate::decode::dashboard`'s module
    // docs for why this needs no raw/redacted split (unlike
    // `attribution`/`payments`/...): the whole table is already Fleet-only,
    // the same posture `personas`/`participations` above use.
    let dashboard_batch =
        decode::dashboard::decode_dashboard_batch(&reference.dashboard).map_err(|source| {
            QueryEngineError::Decode {
                table: DASHBOARD_TABLE,
                source,
            }
        })?;
    register_typed_table(
        ctx,
        DASHBOARD_TABLE,
        decode::dashboard::schema(),
        vec![dashboard_batch],
    )?;

    Ok(())
}

/// Decode `reference.routines` and register [`ROUTINES_TABLE`] on `ctx` —
/// one row per [`crate::routine_catalog::RoutineStatusRecord`] `reference`
/// carries, independent of every persona-side vector
/// [`register_reference_tables`] registers.
///
/// Split out of [`register_reference_tables`] (issue #1882): `routines` is
/// the one reference table that is NOT Fleet-only, so it needs registering
/// at its own point in [`super::QueryEngine::build_from_tables`]'s call
/// sequence — before [`create_scope_dependent_views`] for a persona-scoped
/// session, so the owner-filtered `fires` view
/// ([`crate::views::FIRES_OWNED_VIEW_SQL`]) can join against it; alongside
/// every other Fleet-only reference table (via [`register_reference_tables`]
/// itself, further down that same sequence) for Fleet. This function applies
/// NO filter of its own either way: for a persona-scoped session,
/// `reference.routines` already carries only that persona's own rows,
/// filtered by `crate::authority::ScopedQuery::resolve_routines` before this
/// module ever sees them — see [`ROUTINES_TABLE`]'s own doc.
///
/// # Errors
///
/// Returns [`QueryEngineError::Decode`] if `reference.routines` fails to
/// decode into Arrow.
pub(super) fn register_routines_table(
    ctx: &SessionContext,
    reference: &ReferenceData,
) -> Result<(), QueryEngineError> {
    let routines_batch =
        decode::routines::decode_routines_batch(&reference.routines).map_err(|source| {
            QueryEngineError::Decode {
                table: ROUTINES_TABLE,
                source,
            }
        })?;
    register_typed_table(
        ctx,
        ROUTINES_TABLE,
        decode::routines::schema(),
        vec![routines_batch],
    )
}