polyc-query 2026.8.3

//! The phase-1 query engine: a scoped `DataFusion` `SessionContext` over
//! caller-supplied, already-replayed partitions.
//!
//! [`QueryEngine`] is the crate's public seam (docs/reference/datafusion-data-layer.md,
//! "Verification seams"): real journal fixtures go in as [`PartitionEvents`]
//! (this crate never opens a journal handle itself — that stays the caller's
//! job, matching "Rollout: a phased plan" phase 1's "`MemTable`-style batch
//! decode... refreshed on demand per request"), SQL goes into
//! [`QueryEngine::execute`], and `RecordBatch`es come out.
//!
//! # Catalog scoping and the raw-table hiding problem
//!
//! [`QueryEngine::build`] registers three tables:
//!
//! - [`EVENTS_RAW_TABLE`] (`"events_raw"`) — the wide decode of every
//!   supplied partition's raw replay, via [`crate::decode::events_batch`].
//!   Registered for every scope (build needs it to construct the view
//!   below), but only left resolvable for [`QueryScope::Fleet`] — see below.
//! - [`EVENTS_VIEW`] (`"events"`) — the committed-turns projection
//!   ([`crate::views::COMMITTED_TURNS_VIEW_SQL`]), the query layer's default
//!   surface for every scope.
//! - [`USAGE_TABLE`] (`"usage"`) — a view over [`USAGE_RAW_TABLE`]
//!   ([`crate::decode::usage`]), registered for every scope with IDENTICAL
//!   columns (no Fleet-only column on this table) but filtered to committed
//!   turns — see the "Committed-turn filter invariant" section below.
//! - [`MODEL_CALL_TABLE`] (`"model_call"`) — a view over
//!   [`MODEL_CALL_RAW_TABLE`] ([`crate::decode::model_call`]), the same
//!   treatment as `usage`: a model call is within-conversation data (never
//!   spans conversations), so it carries no cross-conversation leak risk
//!   and is safe to register for [`QueryScope::Conversations`] too, once
//!   committed-turn filtered — see the "Reference data" section below for
//!   the contrast with `personas`/`participations`, which are NOT scoped
//!   this way.
//! - [`ATTRIBUTION_TABLE`] (`"attribution"`) — a view over
//!   [`ATTRIBUTION_RAW_TABLE`] ([`crate::decode::attribution`]), registered
//!   for every scope but with SCOPE-DEPENDENT columns — mirroring the
//!   `events_raw`/`events` split above, not the flat `usage`/`model_call`
//!   treatment. `partition, position, turn_id, persona_id, role` are
//!   conversation-scoped-safe the same way `usage`/`model_call` are: a
//!   `caller`/`participant` row names a persona tied to ONE turn in ONE
//!   conversation, and `persona_id` is already the RESOLVED, opaque
//!   principal id. `identity_provider, identity_scope, identity_external_id,
//!   identity_display_name` — the flattened `ExternalIdentity`, i.e. who a
//!   participant really is on the edge — are NOT: they name every OTHER
//!   participant's raw external identity, so they are registered only for
//!   [`QueryScope::Fleet`]; see the "Identity redaction invariant" section
//!   below for the mechanism.
//! - [`TURN_FAILED_TABLE`] (`"turn_failed"`) — a view over
//!   [`TURN_FAILED_RAW_TABLE`] ([`crate::decode::turn_failed`]), the same
//!   treatment as `usage`/`model_call`: a turn failure is within-conversation
//!   data and carries no external identity at all, so it needs no Fleet-only
//!   gate, and it IS filtered to committed turns — see the "Committed-turn
//!   filter invariant" section below for why the exemption a prior review
//!   pass carved out here was wrong.
//! - [`SUMMARY_TABLE`] (`"summary"`) — a view over [`SUMMARY_RAW_TABLE`]
//!   ([`crate::decode::summary`]), built ONLY for [`QueryScope::Fleet`] — a
//!   persona/conversation-scoped session must never see summary content, and
//!   unlike `attribution`/`payments`/..., this is not a column/row redaction
//!   but a Fleet-only gate on the whole view (the same "never registered for
//!   this scope" posture the "Reference data" section below describes for
//!   `personas`/`participations`), because `summary`'s own `turn_id` is a
//!   synthetic tag rather than a real turn's id and so cannot be
//!   committed-turn filtered the way `usage`/`model_call` are — see the
//!   "Committed-turn filter invariant" section below. See
//!   [`crate::decode::summary`]'s module docs for why its
//!   `covers_through_position` column must never be used for a
//!   position-based transcript cutoff — that cutoff keys on this table's own
//!   `position` column instead.
//! - [`PAYMENTS_TABLE`] (`"payments"`) — a view over [`PAYMENTS_RAW_TABLE`]
//!   ([`crate::decode::payments`]), the fact model's first FOLD-COUPLED
//!   table (decoded through the shared `polyc_facts::verified_receipts`
//!   fold, not a per-kind protobuf decode — see that module's docs).
//!   Registered for every scope but, like `attribution`, with
//!   SCOPE-DEPENDENT columns: every column for [`QueryScope::Fleet`], every
//!   column except the raw `signer_public_key` for every other scope — see
//!   the "Payment signer-key redaction invariant" section below.
//! - [`MESSAGES_TABLE`] (`"messages"`)/[`TOOL_CALLS_TABLE`] (`"tool_calls"`) —
//!   views over [`MESSAGES_RAW_TABLE`]/[`TOOL_CALLS_RAW_TABLE`]
//!   ([`crate::decode::message_content`]), the fact model's sixth typed
//!   table — the first whose decode (one shared content-block fold over
//!   `user_msg`/`output_msg`) fans out into TWO sibling SQL tables rather
//!   than one; see that module's docs for the fold and for why `messages`/
//!   `tool_calls` share one registration pass
//!   ([`register_message_content_tables`]). Registered for every scope but,
//!   like `attribution`/`events`, with a SCOPE-DEPENDENT ROW set: every row
//!   for [`QueryScope::Fleet`], every row except `internal_only = true` for
//!   every other scope — see the "Message-content redaction invariant"
//!   section below.
//! - [`APPROVALS_TABLE`] (`"approvals"`) — a view over
//!   [`APPROVALS_RAW_TABLE`] ([`crate::decode::approvals`]), the fact
//!   model's second FOLD-COUPLED table (decoded through the shared
//!   `polyc_facts::fold_approval_event` fold — the SAME fold
//!   `forensics::classify_response_signature`/`parse_request_entry`/
//!   `parse_response_entry` and `trace::response_signature_evidence` read
//!   through, not a second, independently-written decode). Registered for
//!   every scope but, like `attribution`/`payments`, with SCOPE-DEPENDENT
//!   columns: every column for [`QueryScope::Fleet`], the
//!   participant-visible subset for every other scope — see the "Approval
//!   redaction invariant" section below.
//! - [`HANDOFFS_TABLE`] (`"handoffs"`) — a view over [`HANDOFFS_RAW_TABLE`]
//!   ([`crate::decode::handoffs`]), the fact model's eighth typed table and
//!   a DIRECT decode (unlike `payments`/`approvals`, no `polyc_facts` fold
//!   backs it — see that module's docs for why). Registered for every scope
//!   but, like `payments`, with SCOPE-DEPENDENT columns: every column for
//!   [`QueryScope::Fleet`], every column except the raw `signed_by` for
//!   every other scope — see the "Handoff signed-by redaction invariant"
//!   section below.
//! - [`GRANT_REPLAYS_TABLE`] (`"grant_replays"`) — a view over
//!   [`GRANT_REPLAYS_RAW_TABLE`] ([`crate::decode::grant_replays`]), the
//!   fact model's ninth typed table and the THIRD FOLD-COUPLED one (decoded
//!   through the shared `polyc_facts::fold_grant_replay_event` fold — the
//!   SAME fold `forensics::parse_grant_replay_entry`/
//!   `trace::decode_grant_replay_fields` read through). Registered for
//!   every scope but, like `payments`/`handoffs`, with SCOPE-DEPENDENT
//!   columns: every column for [`QueryScope::Fleet`], every column except
//!   the raw `signer_public_key` for every other scope — see the
//!   "Grant-replay signer-key redaction invariant" section below.
//! - [`TURN_DISPATCH_TABLE`] (`"turn_dispatch"`) — a view over
//!   [`TURN_DISPATCH_RAW_TABLE`] ([`crate::decode::turn_dispatch`]), the same
//!   treatment as `usage`/`model_call`/`turn_failed`: a turn's dispatch
//!   marker (carrying the scheduled-occurrence identity a routine's fire
//!   opened it from, if any) is within-conversation data with no external
//!   identity, so it needs no Fleet-only gate, and IS committed-turn
//!   filtered (issue #1593). Joins to [`FIRES_TABLE`] by `occurrence` — the
//!   routines explorer page's fire-history-to-turns link.
//! - [`ROUTINE_LIFECYCLE_TABLE`] (`"routine_lifecycle"`) — a view over
//!   [`ROUTINE_LIFECYCLE_RAW_TABLE`] ([`crate::decode::routine_lifecycle`]),
//!   the four signed routine lifecycle audit kinds
//!   (created/paused/resumed/deleted) sharing one `phase`-discriminated
//!   table, built ONLY for [`QueryScope::Fleet`] — its one source partition
//!   is itself admitted into replay only for Fleet, and unlike
//!   [`ROUTINES_TABLE`]/[`FIRES_TABLE`] (issue #1882), this table stays
//!   Fleet-only: a routine's own lifecycle audit trail is out of this
//!   slice's scope. The routines explorer page's per-routine lifecycle
//!   timeline source (issue #1593).
//! - [`FIRES_TABLE`] (`"fires"`) — a view over [`FIRES_RAW_TABLE`]
//!   ([`crate::decode::fires`]), built for [`QueryScope::Fleet`] (every row)
//!   and, as of issue #1882, for a persona-scoped session too (rows
//!   belonging to that persona's own routines only, via an INNER JOIN
//!   against [`ROUTINES_TABLE`] — never for a conversation-grant scope.
//!   `FIRES_RAW_TABLE` is deregistered for every non-`Fleet` scope, owner-
//!   scoped session included — the same fail-closed posture [`EVENTS_RAW_TABLE`]
//!   gets (see below) — so `SELECT * FROM fires_raw` cannot bypass the
//!   owner-filtering join. See [`ROUTINES_TABLE`]'s own paragraph below and
//!   [`crate::decode::fires`]'s module docs for the mechanism.
//!
//! A non-`Fleet` session must never reach `events_raw`'s rows under any
//! name. This is enforced by construction, not by a filter a mis-planned
//! query could route around: `events` is built via `ctx.sql("CREATE VIEW
//! events AS ...")`, which resolves `events_raw` and bakes the *resolved*
//! `TableProvider` directly into the view's stored `LogicalPlan`
//! (`datafusion-catalog`'s `ViewTable::scan` clones that plan and executes
//! it straight — it never re-looks-up `events_raw` by name). Once that view
//! exists, [`QueryEngine::build`] calls `ctx.deregister_table("events_raw")`
//! for every non-`Fleet` scope. The view keeps working (its plan already
//! holds the provider), but the name `events_raw` is gone from the schema
//! provider entirely — `SELECT * FROM events_raw`, `SELECT * FROM
//! datafusion.public.events_raw`, or any other qualification of that name
//! fails to resolve, the same "the table simply isn't registered" fail-closed
//! posture the design commits to for catalog scoping generally
//! (docs/reference/datafusion-data-layer.md, "The design: mint a session, then
//! filter server-side": "a mis-planned query against an unregistered table
//! simply fails to resolve rather than silently returning rows it
//! shouldn't"). [`Fleet`](QueryScope::Fleet) sessions skip the deregister
//! step, so `events_raw` stays resolvable alongside `events` and `usage`,
//! matching "Rollout: a phased plan" phase 1's maintainer-only raw view.
//!
//! `information_schema` follows the same on/off-by-scope rule
//! (`SessionConfig::with_information_schema`, set only for `Fleet`), and
//! `enable_url_table` is never called at all — ad hoc external-table
//! references (`SELECT * FROM 'foo.csv'`) stay unreachable from every scope,
//! not just non-admin ones.
//!
//! ## Committed-turn filter invariant (QRY-1)
//!
//! `events` hides an uncommitted/orphaned turn's rows for every scope,
//! Fleet included (see above). Before this invariant every OTHER typed
//! table's public view selected straight from its own `*_raw` table with no
//! such filter, so a turn that produced a real side effect and then crashed
//! before its `turn_complete` marker landed left that side effect readable
//! through `usage`/`model_call`/`attribution`/`payments`/`messages`/
//! `tool_calls`/`approvals`/`handoffs`/`grant_replays`/`turn_failed` even
//! though `events` never shows the identical turn's rows — an asymmetry a
//! [`QueryScope::Conversations`]-scoped persona could read directly, none of
//! those tables being Fleet-only. `crate::views`'s module docs (own section
//! of the same name) carry the full mechanism and the SQL itself; in short,
//! every one of those views now adds a correlated `WHERE EXISTS (SELECT 1
//! FROM events e WHERE e.partition = t.partition AND e.turn_id = t.turn_id)`
//! on top of whatever redaction it already applied — correlated on BOTH
//! `partition` AND `turn_id` (QRY-5), not `turn_id` alone, so a client-minted
//! `turn_id` shared by two DIFFERENT conversations can never let one
//! conversation's committed turn vouch for another's uncommitted one; see
//! `crate::views`'s "QRY-5" section for the full rationale. `turn_failed`
//! was, for one review pass, wrongly believed exempt (the theory: a failed
//! turn's `turn_id` never carries `turn_complete` by construction) — that
//! theory is false, since the control plane's sole `turn_failed` emission
//! (`crates/control-plane/src/grpc/mod.rs`) pushes `turn_failed` and then
//! `turn_complete` unconditionally, in the same atomic batch, so the filter
//! only hides an orphaned/uncommitted `turn_failed` row, exactly like every
//! other typed table above; see [`TURN_FAILED_TABLE`]'s doc. [`SUMMARY_TABLE`]
//! is the ONE remaining documented exception (a summary's `turn_id` is a
//! synthetic tag, not a real turn's id) — see that constant's own doc.
//!
//! ## Payload redaction invariant (phase 2 contract)
//!
//! docs/reference/datafusion-data-layer.md's access-control boundary conditions
//! require that "the persona-scoped `payload` column is bounded to the
//! fields the existing transcript and approval surfaces already show a
//! participant; raw signer keys and other participants' full external
//! identities stay out of persona-scoped results. This redaction applies to
//! every non-maintainer registration — the conversation-scoped agent tool
//! included." As of this
//! writing the `payload`/`payload_json` half of that requirement is
//! satisfied structurally, with no redaction
//! logic to write, because **no relation registered for a non-`Fleet` scope
//! carries the raw `payload` column, or its `payload_json` sibling
//! (#1313's "opaque and JSON-queryable" exposure,
//! [`crate::decode::events_batch`]), at all**:
//!
//! - [`EVENTS_RAW_TABLE`] — the only registration with `payload` and
//!   `payload_json` columns ([`crate::provider::EventsTableProvider::schema`])
//!   — is deregistered for every non-`Fleet` scope before `build` returns
//!   (see above); the name fails to resolve under any qualification.
//! - [`EVENTS_VIEW`] projects [`COMMITTED_TURNS_VIEW_SQL`]'s explicit column
//!   list — `partition, position, kind, turn_id` — which never included
//!   `payload` or `payload_json` to begin with, for any scope.
//! - [`USAGE_TABLE`] is `input_tokens, output_tokens`
//!   ([`crate::decode::usage::schema`]) — no `payload`/`payload_json`
//!   column, for any scope.
//! - [`MODEL_CALL_TABLE`] is `provider, model, captured_clock_unix_ms`
//!   ([`crate::decode::model_call::schema`]) — likewise no `payload`/
//!   `payload_json` column, for any scope.
//! - [`ATTRIBUTION_TABLE`] is, for [`QueryScope::Fleet`], `partition,
//!   position, turn_id, persona_id, role, identity_provider, identity_scope,
//!   identity_external_id, identity_display_name`
//!   ([`crate::decode::attribution::schema`]); for every other scope it is
//!   the redacted `partition, position, turn_id, persona_id, role` (see the
//!   "Identity redaction invariant" section below) — either way, no
//!   `payload`/`payload_json` column, for any scope.
//! - `information_schema` (the one way to enumerate relations a session
//!   cannot already name and select from — `SELECT * FROM t LIMIT 0` returns
//!   `t`'s columns, but only for a `t` this scope registered) is off for
//!   every non-`Fleet` scope, so a non-`Fleet` session cannot even discover
//!   a `payload`/`payload_json` column exists elsewhere in the catalog.
//!
//! This module's `#[cfg(test)]`
//! `conversations_scope_payload_column_is_unreachable` test pins this: it
//! asserts `payload` and `payload_json` both fail to resolve from every
//! relation a `QueryScope::Conversations` session can see, and that neither
//! exposed relation's own schema names either field. **If a future change
//! adds a richer non-`Fleet` `events` view or a new typed table, and that
//! addition carries a `payload`- or `payload_json`-shaped column, this
//! invariant breaks** — the new column must be reviewed against the
//! redaction contract above (bounded to fields already shown on
//! transcript/approval surfaces, no raw signer keys or other participants'
//! full external identities) before it ships, not discovered by a failing
//! test after the fact.
//!
//! ## Identity redaction invariant (attribution's `identity_*` columns)
//!
//! The other half of the same access-control boundary condition quoted
//! above — "raw signer keys and other participants' full external
//! identities stay out of persona-scoped results" — is what
//! [`ATTRIBUTION_TABLE`]'s `identity_provider, identity_scope,
//! identity_external_id, identity_display_name` columns make concrete: each
//! is a caller/participant's raw external identity as observed at the edge
//! (who they really are on the chat edge, e-mail, an API key, …), not the
//! RESOLVED, opaque `persona_id` a conversation-scoped session already
//! legitimately sees for its own conversation. Unlike `payload` above, this
//! one is **not** satisfied structurally by the typed table's own shape —
//! [`crate::decode::attribution`]'s decode keeps every field
//! `AttributionEvent`/`ExternalIdentity` carries, identity included — so the
//! redaction happens at registration instead, the same fail-closed
//! mechanism [`EVENTS_RAW_TABLE`]/[`EVENTS_VIEW`] already use (PR #1352's
//! review finding: the prior `attribution` registration exposed every
//! participant's raw identity to every conversation-scoped session,
//! violating the boundary condition above):
//!
//! - [`ATTRIBUTION_RAW_TABLE`] (`"attribution_raw"`) — the full decode,
//!   `identity_*` columns included — is registered for every scope during
//!   `build` (needed to construct the view below), then deregistered for
//!   every non-`Fleet` scope before `build` returns, exactly like
//!   [`EVENTS_RAW_TABLE`]; the name fails to resolve under any
//!   qualification outside Fleet.
//! - [`ATTRIBUTION_TABLE`] (`"attribution"`) is built via `CREATE VIEW
//!   attribution AS ...`, with the SQL text itself scope-dependent:
//!   [`crate::views::ATTRIBUTION_VIEW_SQL`] (every column, `identity_*`
//!   included) for [`QueryScope::Fleet`], or
//!   [`crate::views::ATTRIBUTION_REDACTED_VIEW_SQL`] (`partition, position,
//!   turn_id, persona_id, role` — no `identity_*`) for every other scope.
//!   Same mechanism `events`/`events_raw` already rely on: `CREATE VIEW`
//!   bakes the *resolved* `TableProvider` straight into the view's stored
//!   `LogicalPlan`, so deregistering `attribution_raw` afterward does not
//!   disturb either view — a non-`Fleet` session gets a real, working
//!   `attribution` table that simply never had an `identity_*` column to
//!   begin with, the same "unregistered/unresolvable" fail-closed posture
//!   as `payload` above, here applied to specific columns of a table a
//!   maintainer session still sees in full under the identical name.
//!
//! This module's `#[cfg(test)]`
//! `conversations_scope_attribution_identity_columns_are_unreachable` test
//! pins this: it asserts each of the four `identity_*` columns fails to
//! resolve for a `QueryScope::Conversations` session, that the same
//! session's `attribution` schema names only its five redacted columns, and
//! that `persona_id`/`role` still resolve and return real values — the
//! session keeps its own conversation's resolved attribution, just never
//! another participant's raw external identity.
//! `attribution_rows_are_queryable_and_join_to_their_conversation` pins the
//! Fleet side: a [`QueryScope::Fleet`] session's `attribution` still
//! carries every `identity_*` column with real values.
//!
//! ## Payment signer-key redaction invariant (`payments.signer_public_key`)
//!
//! The identical mechanism, applied to [`PAYMENTS_TABLE`]'s one raw-key
//! column: docs/reference/datafusion-data-layer.md's boundary conditions state
//! "raw signer keys ... stay out of persona-scoped results. This redaction
//! applies to every non-maintainer registration." [`PAYMENTS_RAW_TABLE`]
//! (full decode, `signer_public_key` included) is registered for every
//! scope during `build`, then deregistered for every non-`Fleet` scope,
//! exactly like [`ATTRIBUTION_RAW_TABLE`]. [`PAYMENTS_TABLE`] is built via
//! `CREATE VIEW payments AS ...`, with the SQL text scope-dependent:
//! [`crate::views::PAYMENTS_VIEW_SQL`] (every column) for
//! [`QueryScope::Fleet`], [`crate::views::PAYMENTS_REDACTED_VIEW_SQL`]
//! (every column except `signer_public_key`) for every other scope. Every
//! OTHER payment column — `subject` (the already-resolved, opaque principal
//! the spend is attributed to) included — stays visible at every scope; see
//! `crate::decode::payments`'s module docs for why `subject` is not treated
//! as raw identity. This module's `#[cfg(test)]`
//! `conversations_scope_payments_signer_public_key_is_unreachable` test
//! pins this the same way
//! `conversations_scope_attribution_identity_columns_are_unreachable` pins
//! `attribution`'s redaction.
//!
//! ## Message-content redaction invariant (`messages`/`tool_calls`)
//!
//! The same access-control boundary condition once more, applied to a ROW
//! filter rather than a column filter: `crate::decode::message_content`'s
//! module docs explain why an `internal_only` message — one the wire
//! `Message` type's own doc comment says "is not part of the conversation
//! history and is not emitted to clients" — must never reach a non-Fleet
//! session, the identical "bounded to what the existing transcript surface
//! already shows a participant" rule [`ATTRIBUTION_TABLE`]'s `identity_*`
//! gate enforces for a different column shape:
//!
//! - [`MESSAGES_RAW_TABLE`]/[`TOOL_CALLS_RAW_TABLE`] (full decode,
//!   `internal_only = true` rows included) are registered for every scope
//!   during `build`, then deregistered for every non-`Fleet` scope, exactly
//!   like [`ATTRIBUTION_RAW_TABLE`].
//! - [`MESSAGES_TABLE`]/[`TOOL_CALLS_TABLE`] are built via `CREATE VIEW`,
//!   with the SQL text scope-dependent:
//!   [`crate::views::MESSAGES_VIEW_SQL`]/[`crate::views::TOOL_CALLS_VIEW_SQL`]
//!   (every row) for [`QueryScope::Fleet`],
//!   [`crate::views::MESSAGES_REDACTED_VIEW_SQL`]/
//!   [`crate::views::TOOL_CALLS_REDACTED_VIEW_SQL`] (`WHERE internal_only =
//!   false`) for every other scope. Unlike `attribution`'s redaction, the
//!   column list is IDENTICAL between the two view bodies here — this is a
//!   ROW filter, the same shape [`crate::views::COMMITTED_TURNS_VIEW_SQL`]
//!   already uses to filter `events_raw` down to committed-turn rows, not a
//!   column-level redaction (there is no raw-identity-shaped column on
//!   either table to hide the way `identity_*` is on `attribution`).
//!
//! Every OTHER field on both tables — full tool-call arguments, full
//! tool-result payloads, plain message text, all untruncated — stays
//! visible at every scope: `crate::decode::message_content`'s module docs
//! establish that these are already within what a committed turn's own
//! transcript/approval surface shows a participant, just without that
//! surface's display-length truncation. This module's `#[cfg(test)]`
//! `conversations_scope_hides_internal_only_messages_and_tool_calls` test
//! pins the redaction; `message_content_rows_are_queryable_and_join_to_their_conversation`
//! and `tool_call_and_its_result_share_a_tool_call_id` pin ordinary
//! queryability/joins.
//!
//! ## Approval redaction invariant (`approvals`)
//!
//! The identical raw-table/scope-dependent-view mechanism
//! [`PAYMENTS_TABLE`] uses, applied to [`APPROVALS_TABLE`]'s NINE Fleet-only
//! columns rather than one: `crate::decode::approvals`'s module docs name
//! `request_reason`, `request_sandbox_mode`, `signer_public_key`,
//! `modified_args_json`, `approved_for_session`, `caller`, `approver`,
//! `response_sandbox_mode`, and `injected_context` as exceeding
//! `forensics::collect_approvals`' own participant-visible set (raw signer
//! keys are out of persona scope unconditionally per
//! docs/reference/datafusion-data-layer.md's boundary conditions; the richer
//! response fields go further, matching that same document's requirement
//! that the persona-scoped surface be "bounded to the fields the existing
//! transcript and approval surfaces already show a participant").
//!
//! - [`APPROVALS_RAW_TABLE`] (full decode, every Fleet-only column included)
//!   is registered for every scope during `build`, then deregistered for
//!   every non-`Fleet` scope, exactly like [`PAYMENTS_RAW_TABLE`].
//! - [`APPROVALS_TABLE`] is built via `CREATE VIEW approvals AS ...`, with
//!   the SQL text scope-dependent: [`crate::views::APPROVALS_VIEW_SQL`]
//!   (every column) for [`QueryScope::Fleet`],
//!   [`crate::views::APPROVALS_REDACTED_VIEW_SQL`] (`partition, position,
//!   turn_id, phase, request_id, tool_name, args_json, approved,
//!   response_reason, signature_status` — exactly
//!   `forensics::collect_approvals`'s participant-visible fields) for every
//!   other scope.
//!
//! This module's `#[cfg(test)]` `conversations_scope_approvals_signer_public_key_is_unreachable`
//! test pins this the same way
//! `conversations_scope_payments_signer_public_key_is_unreachable` pins
//! `payments`' redaction, and `crates/control-plane/src/forensics.rs`'s
//! non-admin parity test proves the redacted view's rows equal what
//! `collect_approvals` computes for the identical events, field for field.
//!
//! ## Handoff signed-by redaction invariant (`handoffs.signed_by`)
//!
//! This uses the same raw-table and scope-view pattern as [`PAYMENTS_TABLE`].
//! `signed_by` is the only Fleet-only handoff column. It contains a raw public
//! key. `signature_status` contains only its trust-pinned verdict, so every
//! scope can read it.
//!
//! - [`HANDOFFS_RAW_TABLE`] (full decode, `signed_by` included) is
//!   registered for every scope during `build`, then deregistered for every
//!   non-`Fleet` scope, exactly like [`PAYMENTS_RAW_TABLE`].
//! - [`HANDOFFS_TABLE`] is built via `CREATE VIEW handoffs AS ...`, with the
//!   SQL text scope-dependent: [`crate::views::HANDOFFS_VIEW_SQL`] (every
//!   column) for [`QueryScope::Fleet`],
//!   [`crate::views::HANDOFFS_REDACTED_VIEW_SQL`] (every column except
//!   `signed_by`) for every other scope.
//!
//! This module's `#[cfg(test)]` `conversations_scope_handoffs_signed_by_is_unreachable`
//! test pins this the same way
//! `conversations_scope_payments_signer_public_key_is_unreachable` pins
//! `payments`' redaction, and `crates/control-plane/src/forensics.rs`'s
//! non-admin parity test proves the redacted view's rows equal what
//! `collect_handoffs`/`collect_handoff_denials` compute for the identical
//! events, field for field — except `reason`, which neither collector's JSON
//! entry surfaces today, so that column is diffed against the source proto
//! values instead (see that test's own doc for the gap).
//!
//! ## Grant-replay signer-key redaction invariant (`grant_replays.signer_public_key`)
//!
//! The identical raw-table/scope-dependent-view mechanism [`PAYMENTS_TABLE`]
//! uses, applied to [`GRANT_REPLAYS_TABLE`]'s ONE Fleet-only column:
//! [`crate::decode::grant_replays`]'s module docs name `signer_public_key` —
//! a raw ed25519 public key — as the only column that exceeds
//! `forensics::collect_grant_replays`'s own already-participant-visible
//! field set (`GrantReplayEntry`'s `tool`/`grant_ref`/`covered_capabilities`/
//! `coverage_hash`/`signature` fields). `signature_status` (the STRING
//! outcome of checking that key, not the key itself) stays visible at every
//! scope, mirroring `handoffs.signature_status`.
//!
//! - [`GRANT_REPLAYS_RAW_TABLE`] (full decode, `signer_public_key` included)
//!   is registered for every scope during `build`, then deregistered for
//!   every non-`Fleet` scope, exactly like [`PAYMENTS_RAW_TABLE`]/
//!   [`HANDOFFS_RAW_TABLE`].
//! - [`GRANT_REPLAYS_TABLE`] is built via `CREATE VIEW grant_replays AS ...`,
//!   with the SQL text scope-dependent:
//!   [`crate::views::GRANT_REPLAYS_VIEW_SQL`] (every column) for
//!   [`QueryScope::Fleet`], [`crate::views::GRANT_REPLAYS_REDACTED_VIEW_SQL`]
//!   (every column except `signer_public_key`) for every other scope.
//!
//! This module's `#[cfg(test)]`
//! `conversations_scope_grant_replays_signer_public_key_is_unreachable` test
//! pins this the same way
//! `conversations_scope_handoffs_signed_by_is_unreachable` pins `handoffs`'
//! redaction, and `crates/control-plane/src/forensics.rs`'s non-admin parity
//! test proves the redacted view's rows equal what `collect_grant_replays`
//! computes for the identical events, field for field.
//!
//! ## Reference data: `personas`/`participations` (#1312, part of #1178)
//!
//! [`ReferenceData`] is this crate's first NON-journal data source: persona
//! profiles and participation ties are general domain reference data that
//! live in `PersonaHost`/`PersonaStore` (`crates/persona`) — a commonware
//! qmdb key-value store, a substrate entirely separate from the event
//! journal every other table in this module decodes. [`QueryEngine::build`]
//! takes a caller-supplied [`ReferenceData`] alongside `partitions` and
//! decodes it into seven more `MemTable` registrations —
//! [`PERSONAS_TABLE`]/[`PARTICIPATIONS_TABLE`]/[`PERSONA_IDENTITIES_TABLE`]
//! (#1312) plus [`PERSONA_WALLETS_TABLE`]/[`PERSONA_SPEND_POLICIES_TABLE`]/
//! [`PERSONA_CREDENTIALS_TABLE`]/[`PERSONA_USAGE_TABLE`] (#1578, Phase D) —
//! via [`crate::decode::persona`], plus an EIGHTH ([`ROUTINES_TABLE`], issue
//! #1592, see below) via [`crate::decode::routines`] — mirroring
//! [`PartitionEvents`]'s own "the caller replays/reads, this crate only
//! decodes" division of labor, so a further non-journal source can plug
//! into the same `build` call the same way in a future phase without a
//! bespoke second ingestion path.
//! [`PERSONA_IDENTITIES_TABLE`] needs no new [`ReferenceData`] field of its
//! own (#1312's G2 gap): each [`PersonaProfile`] in `reference.personas`
//! already carries its `identities`, so this third table is a projection of
//! data [`PERSONAS_TABLE`] already decodes, not a new data source. The four
//! #1578 tables DO each need their own [`ReferenceData`] field
//! (`wallets`/`spend_policies`/`credentials`/`usage_rollups`) — a persona's
//! wallet link, spend policy, passkey credential, and usage rollup are each
//! independently absent-or-present, unlike `identities` which nests inside
//! the profile that is already fetched. `crate::authority`'s
//! `resolve_reference_data` populates all seven persona-side Fleet tables'
//! backing vectors from ONE [`polyc_persona::PersonaReferenceSnapshot`]
//! round trip per candidate persona, not five separate `PersonaHost` calls
//! — see that type's own doc and [`crate::decode::persona`]'s module docs'
//! "Bulk-export invariant (QRY-8)" section for the redaction bar the four
//! new tables hold to. [`DASHBOARD_TABLE`] (#1584/#1585, part of the
//! read-path-convergence epic #1574) is a ninth `MemTable` registration,
//! decoded from [`ReferenceData::dashboard`] via [`crate::decode::dashboard`]
//! — see that module's docs for its column set and why it needs no
//! raw/redacted split.
//!
//! **Fleet-only for this phase, except `routines` (issue #1882).** Eight of
//! the nine tables are registered ONLY for [`QueryScope::Fleet`] — never for
//! `QueryScope::Conversations` — because they span every persona (or, for
//! `dashboard`, every conversation) in the deployment, and per-persona
//! visibility (a caller seeing only their OWN row plus the participations
//! that name them) is `#1178`'s phase 6
//! (docs/reference/datafusion-data-layer.md, "Roles and the trust boundary":
//! persona-scoped access is filtered by `PersonaStore::participations`, not
//! shipped yet). Registering these tables unconditionally today would let a
//! conversation-scoped session read every OTHER persona's profile, every
//! OTHER conversation's participation ties, every OTHER persona's linked
//! external identities, and — since #1578 — every OTHER persona's wallet
//! address, spend cap, host allowlist, and passkey-enrollment metadata: a
//! strictly worse leak than the raw-`payload` redaction this module already
//! guards above. This module's `#[cfg(test)]`
//! `conversations_scope_hides_reference_tables` test (and its #1578 sibling
//! covering the four new tables, and its own #1585 sibling covering
//! `dashboard`) pins it, mirroring
//! `conversations_scope_hides_raw_table_by_every_name`'s "the name is gone
//! from the catalog entirely" pattern for `events_raw`.
//!
//! [`ROUTINES_TABLE`] (`"routines"`, issue #1592) is an EIGHTH reference
//! table, added the same deliberate way: [`ReferenceData::routines`] carries
//! `crate::routine_catalog::RoutineStatusRecord`s the caller (in production,
//! `crate::authority::ScopedQuery`'s own `RoutineCatalog` handle) already
//! resolved, decoded here via [`crate::decode::routines`] — the identical
//! "caller resolves, this crate only decodes" division of labor
//! `personas`/`participations` establish.
//!
//! Unlike every other reference table, `routines` (and, riding on it,
//! `fires`) is NOT Fleet-only as of issue #1882: a routine's own
//! `creator_persona` is a first-class row-ownership column, and the trust
//! boundary here is member↔member, not member↔fleet — a member seeing their
//! own routines and fire history is not the same leak as a member seeing
//! every OTHER persona's profile, wallet, or conversation participation, the
//! actual leaks the "Fleet-only for this phase" paragraph above guards
//! against. `crate::authority::ScopedQuery::resolve_routines` filters
//! `reference.routines` to the caller's own `creator_persona` rows BEFORE
//! this module ever sees them for a persona-scoped session, so this crate's
//! own decode/registration step applies no filter of its own for either
//! scope — it is uniformly "decode whatever rows the caller resolved,"
//! matching every other reference table's division of labor. See
//! [`ROUTINES_TABLE`]'s own doc and `crate::engine::registration`'s
//! `register_routines_table` for exactly which scopes register it and when.

use std::collections::HashSet;
use std::path::PathBuf;
use std::sync::Arc;
use std::time::Duration;

use arrow::datatypes::SchemaRef;
use arrow::error::ArrowError;
use arrow::record_batch::RecordBatch;
use datafusion::catalog::MemoryCatalogProviderList;
use datafusion::common::{Column, SchemaError};
use datafusion::error::DataFusionError;
use datafusion::execution::context::{SQLOptions, SessionConfig, SessionContext};
use datafusion::execution::{SessionState, SessionStateBuilder};
use datafusion::scalar::ScalarValue;
use datafusion_functions_json::register_all as register_json_functions;
use polyc_eventlog::Event;
use polyc_proto::proto::polychrome::persona::v1::{
    Participation, PersonaCredential, PersonaProfile, SpendPolicy, UsageRollup, WalletLink,
};

use crate::provider::EventsTableProvider;
use crate::session::QueryScope;
use crate::statement_gate::{self, AllowedStatement, StatementRejected};
use crate::views::COMMITTED_TURNS_VIEW_SQL;

mod registration;
use registration::{
    create_scope_dependent_views, register_message_content_tables, register_reference_tables,
    register_routines_table, register_typed_journal_tables, register_typed_table,
};

mod partition_tables;
#[cfg(test)]
pub(crate) use partition_tables::DECODE_CALL_COUNT;
pub(crate) use partition_tables::{PartitionTables, decode_partition_tables};

/// The handoff signer every `#[cfg(test)]` fixture in this crate signs with.
///
/// [`QueryEngine::build`] trusts this key in fixtures. Its handoffs read as
/// `verified`; other keys read as `untrusted`. Production resolves its trust
/// from custody through `crate::authority::QueryAuthority::new_state_backed`.
#[cfg(test)]
pub(crate) fn fixture_handoff_signer() -> polyc_crypto::signing_role::HandoffSigner {
    polyc_crypto::signing_role::HandoffSigner::from_seed(0x0112_4000)
}

/// The trust set holding exactly [`fixture_handoff_signer`]'s key.
#[cfg(test)]
pub(crate) fn fixture_handoff_trust()
-> polyc_crypto::signing_role::RoleTrustSet<polyc_crypto::signing_role::HandoffRole> {
    polyc_crypto::signing_role::RoleTrustSet::current(&fixture_handoff_signer())
}

mod tables;
pub(crate) use tables::*;

/// One journal partition's already-replayed events, supplied by the caller.
///
/// Phase 1 is batch decode on demand: the caller — since the A2 sealed-funnel
/// retrofit, exclusively [`crate::authority::ScopedQuery::execute`], never the
/// control plane directly — replays a partition via
/// [`polyc_eventlog_host::EventLogHost::replay_with_positions`] and hands the
/// resulting `(position, Event)` pairs straight to [`QueryEngine::build`].
/// This crate never opens a raw journal handle itself; it drives the same
/// `EventLogHost` handle [`crate::authority::QueryAuthority`] holds.
#[derive(Debug)]
pub(crate) struct PartitionEvents {
    /// The journal partition name (`conv-{id}`), stored verbatim into
    /// `events_raw`'s `partition` column.
    pub partition: String,
    /// This partition's events, each paired with its journal position, in
    /// append order — the shape
    /// [`polyc_eventlog::EventLog::replay_with_positions`] already returns.
    pub events: Vec<(u64, Event)>,
}

/// Fleet-scoped reference data external to the event journal.
///
/// Supplied by the caller alongside `partitions` — this crate's first
/// non-journal data source (see the module docs' "Reference data" section).
/// Registered as the
/// [`PERSONAS_TABLE`]/[`PARTICIPATIONS_TABLE`]/[`PERSONA_IDENTITIES_TABLE`]/
/// [`PERSONA_WALLETS_TABLE`]/[`PERSONA_SPEND_POLICIES_TABLE`]/
/// [`PERSONA_CREDENTIALS_TABLE`]/[`PERSONA_USAGE_TABLE`] `MemTable`s, for
/// [`QueryScope::Fleet`] sessions only.
///
/// Like [`PartitionEvents`], this module never reads `PersonaHost`/
/// `PersonaStore` itself — [`crate::authority::ScopedQuery::execute`] (the
/// SAME `PersonaCell` handle every other persona-reading surface shares — see
/// that module's doc) resolves the records and hands them over already
/// typed, via ONE [`polyc_persona::PersonaReferenceSnapshot`] fetch per
/// candidate persona (#1578, Phase D) rather than five separate calls — see
/// that type's own doc.
#[derive(Debug)]
pub(crate) struct ReferenceData {
    /// Every persona profile this Fleet session's `personas` table exposes.
    pub personas: Vec<PersonaProfile>,
    /// Every participation tie this Fleet session's `participations` table
    /// exposes, paired with the persona id it belongs to — [`Participation`]
    /// itself does not carry `persona_id` (see
    /// `crate::decode::persona`'s module docs), so the caller supplies it
    /// alongside each record.
    pub participations: Vec<(String, Participation)>,
    /// Every linked wallet this Fleet session's `persona_wallets` table
    /// exposes, paired with the persona id it belongs to — [`WalletLink`]
    /// already carries its own `persona_id` field, but the pairing here
    /// mirrors `participations`' shape for consistency across every
    /// reference-data vector this struct carries.
    pub wallets: Vec<(String, WalletLink)>,
    /// Every set spend policy this Fleet session's `persona_spend_policies`
    /// table exposes, paired with the persona id it belongs to.
    pub spend_policies: Vec<(String, SpendPolicy)>,
    /// Every enrolled passkey credential this Fleet session's
    /// `persona_credentials` table exposes, paired with the persona id it
    /// belongs to.
    pub credentials: Vec<(String, PersonaCredential)>,
    /// Every maintained usage rollup this Fleet session's `persona_usage`
    /// table exposes, paired with the persona id it belongs to.
    pub usage_rollups: Vec<(String, UsageRollup)>,
    /// Every routine's status surface this session's `routines` table
    /// exposes (issue #1592) — resolved via
    /// `crate::authority::ScopedQuery`'s own
    /// `crate::routine_catalog::RoutineCatalog` handle, the identical
    /// "caller resolves already-typed values, this crate only decodes"
    /// division of labor `personas`/`participations` above already
    /// establish. See `crate::routine_catalog`'s module doc for why this is
    /// a plain record type, never `polyc_controller::Routine` itself.
    ///
    /// Unlike every other field on this struct, `routines` is populated for
    /// a persona-scoped session too (issue #1882) — already FILTERED to
    /// that persona's own `creator_persona` rows by
    /// `crate::authority::ScopedQuery::resolve_routines` before it ever
    /// reaches here, so this crate applies no filter of its own; it decodes
    /// whatever rows the caller resolved, same as always.
    pub routines: Vec<crate::routine_catalog::RoutineStatusRecord>,
    /// Every row this Fleet session's `dashboard` table exposes — a snapshot
    /// of [`crate::dashboard::DashboardProjection::rows`] at the moment the
    /// query's `ScopedQuery` was built, independent of which partitions this
    /// particular query happens to replay (see
    /// [`crate::decode::dashboard`]'s module docs).
    pub dashboard: Vec<crate::dashboard::DashboardRow>,
}

impl ReferenceData {
    /// No persona/participation/wallet/spend-policy/credential/usage-rollup/
    /// routine/dashboard rows — a convenient empty fixture for tests that
    /// don't exercise the reference tables. `#[cfg(test)]` only: since issue
    /// #1882 gave `routines` its own non-Fleet fallback
    /// ([`Self::empty_except_routines`]), production code has no remaining
    /// caller for this fully-empty constructor — every real non-Fleet build
    /// resolves `routines` (possibly to zero rows) rather than skipping it
    /// outright.
    #[cfg(test)]
    #[must_use]
    pub(crate) const fn empty() -> Self {
        Self {
            personas: Vec::new(),
            participations: Vec::new(),
            wallets: Vec::new(),
            spend_policies: Vec::new(),
            credentials: Vec::new(),
            usage_rollups: Vec::new(),
            routines: Vec::new(),
            dashboard: Vec::new(),
        }
    }

    /// Every reference table empty except `dashboard` — the
    /// persona-host-unavailable fallback in
    /// [`crate::authority::ScopedQuery::resolve_reference_data`], which keeps
    /// serving the fleet-wide dashboard snapshot (it never depends on the
    /// persona store — see that method's own doc) while every persona-backed
    /// table goes empty.
    ///
    /// Named field-by-field, deliberately never a `..Self::empty()` spread:
    /// this is a PRODUCTION fallback (unlike every other spread of this
    /// struct's shape, which is a test fixture), and it is the one branch a
    /// future ninth reference table would otherwise silently serve empty
    /// instead of forcing its author to decide what "the persona host is
    /// down" means for that field.
    #[must_use]
    pub(crate) const fn empty_except_dashboard(
        dashboard: Vec<crate::dashboard::DashboardRow>,
    ) -> Self {
        Self {
            personas: Vec::new(),
            participations: Vec::new(),
            wallets: Vec::new(),
            spend_policies: Vec::new(),
            credentials: Vec::new(),
            usage_rollups: Vec::new(),
            routines: Vec::new(),
            dashboard,
        }
    }

    /// Every reference table empty except `routines` — a persona-scoped
    /// session's own build (issue #1882): every Fleet-only reference table
    /// (`personas`/`participations`/.../`dashboard`) stays empty for this
    /// scope, but `routines` carries this persona's own routines, already
    /// filtered by `crate::authority::ScopedQuery::resolve_routines`. Named
    /// field-by-field for the same reason [`Self::empty_except_dashboard`]
    /// is: a future tenth reference table must force its author to decide
    /// what a persona-scoped session sees for it, never inherit an empty
    /// default silently via a `..Self::empty()` spread.
    #[must_use]
    pub(crate) const fn empty_except_routines(
        routines: Vec<crate::routine_catalog::RoutineStatusRecord>,
    ) -> Self {
        Self {
            personas: Vec::new(),
            participations: Vec::new(),
            wallets: Vec::new(),
            spend_policies: Vec::new(),
            credentials: Vec::new(),
            usage_rollups: Vec::new(),
            routines,
            dashboard: Vec::new(),
        }
    }
}

/// Resource ceilings applied to every query a `QueryEngine` runs.
///
/// Every rollout phase requires a memory-pool ceiling, a hard wall-clock
/// timeout, a row cap whose truncation is flagged in the response
/// (docs/reference/datafusion-data-layer.md, "Rollout: a phased plan" phase 1),
/// and (QRY-3) a pre-execution source-event budget — see
/// [`QueryLimits::max_source_events`]'s own doc for why the first three
/// alone leave a gap this one closes, and — IMPORTANT — for why that budget
/// is a decode-amplification backstop, not a memory or OOM bound. The
/// remaining two fields, [`QueryLimits::spill_dir`]/
/// [`QueryLimits::spill_quota_bytes`], configure where and how much
/// `DataFusion`'s own execution-time spill is allowed to write.
///
/// Not `Copy` — [`QueryLimits::spill_dir`] owns a [`PathBuf`] — so a caller
/// holding one behind a shared reference clones it explicitly at the two
/// points that need an owned copy (`crate::authority::QueryAuthority::scope_for`
/// building a [`crate::authority::ScopedQuery`], and
/// [`crate::authority::ScopedQuery::execute`] passing one into
/// `QueryEngine::build`) rather than relying on an implicit bitwise copy.
#[derive(Debug, Clone)]
pub struct QueryLimits {
    /// Ceiling, in bytes, on `DataFusion`'s own memory-pool allocations.
    ///
    /// Consumed exactly ONCE, by [`crate::authority::QueryAuthority::new_state_backed`],
    /// to size the ONE process-wide
    /// [`datafusion::execution::memory_pool::FairSpillPool`] every
    /// [`crate::authority::ScopedQuery`] shares (docs/reference/datafusion-data-layer.md's
    /// 2026-07-21 decision) — NOT a per-query ceiling any more: a shared
    /// *greedy* pool would let one query's allocation starve every
    /// concurrent sibling into `ResourcesExhausted` instead of the fair
    /// share `FairSpillPool` gives each. Exceeding the shared ceiling
    /// surfaces as a typed `QueryEngineError::DataFusion` on whichever
    /// query's allocation tipped it over, never an OOM kill. Crucially, this
    /// pool only accounts for `DataFusion`'s OWN execution-time allocations —
    /// see [`QueryLimits::max_source_events`]'s doc for the pre-execution gap
    /// it does NOT cover.
    ///
    /// This field's own [`Self::default`] deliberately sizes the pool to a
    /// FRACTION of the pod's memory limit, not the whole of it — see that
    /// impl's doc for why headroom below the pod ceiling is load-bearing,
    /// not merely conservative.
    pub memory_bytes: usize,
    /// Hard wall-clock ceiling on `QueryEngine::execute`'s collect step. A
    /// query still running past this deadline surfaces
    /// `QueryEngineError::Timeout` (see that method's docs for what
    /// happens to the abandoned work). `crate::authority::ScopedQuery::execute`
    /// wraps its ENTIRE pipeline (replay, decode, build, execute) in this
    /// same wall-clock timeout — but see [`QueryLimits::max_source_events`]'s
    /// doc for why that wrapper cannot actually preempt the replay/decode
    /// portion of that pipeline before it finishes.
    pub timeout: Duration,
    /// Maximum rows `QueryEngine::execute` returns, across every collected
    /// batch combined. A result with more rows is truncated to exactly this
    /// many, flagged via `QueryOutput::truncated`.
    pub row_cap: usize,
    /// Ceiling on the total number of events replayed across every scoped
    /// partition, checked by `crate::authority::ScopedQuery::execute` right
    /// after `replay_scoped_partitions` returns and BEFORE any of those
    /// events reach `QueryEngine::build`'s decode step. A scope whose total
    /// event count exceeds this returns
    /// `crate::authority::ScopedQueryError::SourceBudgetExceeded` instead of
    /// proceeding.
    ///
    /// # What this actually is: a decode-amplification backstop, NOT a memory bound
    ///
    /// This field caps how many source events feed the raw-plus-typed Arrow
    /// DECODE FAN-OUT `QueryEngine::build` runs before execution — roughly
    /// ten typed tables (`usage`, `model_call`, `attribution`, `payments`,
    /// `approvals`, `handoffs`, `messages`, `tool_calls`, `turn_failed`,
    /// `summary`) plus the wide `events_raw` table's OWN duplicated
    /// `payload`/`payload_json` columns, each pass materializing its own set
    /// of Arrow arrays over the same replayed events. It is checked because
    /// `timeout` cannot preempt that fan-out (every decode loop is
    /// synchronous — see below) — it is NOT, and must never be read as, any
    /// of the following:
    ///
    /// - **Not an OOM bound.** `memory_bytes`'s `FairSpillPool` ceiling is
    ///   sized as a fraction of the pod's own memory limit (see
    ///   [`QueryLimits::default`]'s doc), but that pool meters ONLY
    ///   `DataFusion`'s execution-time allocations. Both the replay step
    ///   (`EventLogHost::replay_with_positions`, building `Vec<(u64, Event)>`
    ///   in ordinary heap memory) and this field's own decode fan-out happen
    ///   entirely OUTSIDE the pool's accounting, on the SAME pod, competing
    ///   for the SAME physical memory the pool's ceiling assumes it alone
    ///   governs. A query that passes this check can still exhaust the pod's
    ///   real memory before `DataFusion` ever gets a chance to enforce
    ///   `memory_bytes` — this field narrows that risk, it does not close it.
    /// - **Does not bound the replay allocation that precedes it.** The
    ///   `Vec<(u64, Event)>` `replay_scoped_partitions` builds for every
    ///   scoped partition is already fully allocated by the time this field
    ///   is even checked — an event count says nothing about how large any
    ///   one event's `payload` is, so the replay step's own memory cost is
    ///   unbounded by this field in either direction.
    /// - **Is event-COUNT, not bytes.** A deployment with unusually large
    ///   individual event payloads (large tool outputs, large embedded
    ///   attachments) can trip its real memory ceiling at a small fraction of
    ///   this many events; a deployment with unusually small payloads never
    ///   comes close at the full count. See
    ///   `crate::metrics`'s replayed-bytes histogram (issue #1541) for the
    ///   observability this field cannot give on its own.
    ///
    /// Issue #1541 tracks the real fixes this backstop stands in for: a
    /// byte-based meter (not event-count), selective/lazy decode (only the
    /// typed tables a query's own `FROM`/`JOIN` actually touch, not all ~10
    /// unconditionally), a retention policy bounding how much history a
    /// Fleet scope can even accumulate, and infra-backed spill sized to a
    /// real per-deployment ceiling. This field is an accepted incremental
    /// backstop against a clearly pathological/unbounded-growth query, never
    /// a resolved OOM finding — see [`Self::default`]'s doc for the
    /// worst-case reasoning behind its default value.
    ///
    /// It is also NOT preemptible by `timeout`: every decode loop
    /// `QueryEngine::build` runs is synchronous (no `.await` inside the
    /// per-partition loops), so `tokio::time::timeout` can only ever observe
    /// the decode step's completion (success or a stack overflow / OOM
    /// kill), never interrupt it mid-pass. This field is the actual bound on
    /// that unpreemptible work; the wall-clock timeout bounds everything
    /// else.
    pub max_source_events: usize,

    /// Ceiling, in bytes, on the total replayed event PAYLOAD across every
    /// scoped partition — issue #1541's real fix for the gap
    /// [`QueryLimits::max_source_events`]'s own doc names: this budget is
    /// enforced DURING replay (`crate::authority::ScopedQuery::replay_scoped_partitions`,
    /// via `polyc_eventlog_host::EventLogHost::replay_with_positions_bounded`),
    /// not after a full `Vec` of every scoped partition's events is already
    /// materialized. Replay stops reading from the journal the instant
    /// cumulative payload bytes exceed this budget, so the peak allocation a
    /// single query's replay step can reach is bounded to roughly this many
    /// bytes plus one partition's worth of overrun — never the whole
    /// deployment's history, even for a Fleet scope.
    ///
    /// A scope whose replay trips this returns
    /// `crate::authority::ScopedQueryError::SourceBudgetExceeded` — the SAME
    /// variant, and the same scope-aware, count-free message shape, as
    /// [`QueryLimits::max_source_events`]'s own rejection; only the
    /// underlying cause differs. [`QueryLimits::max_source_events`] stays in
    /// place alongside this field as a secondary, decode-amplification
    /// backstop — it is NOT superseded, since it also guards the
    /// per-partition Arrow decode fan-out this field says nothing about
    /// directly (see that field's own "not a memory bound" doc).
    ///
    /// See [`Self::default`]'s doc for the worst-case reasoning behind the
    /// default value.
    pub max_source_bytes: u64,

    /// Directory `DataFusion`'s spillable operators write their spill files
    /// under, via `RuntimeEnvBuilder::with_temp_file_path`
    /// (`crate::authority::build_base_session_state`) — consumed exactly
    /// ONCE, the same as [`QueryLimits::memory_bytes`].
    ///
    /// This crate's OWN [`Self::default`] keeps this portable — a
    /// [`std::env::temp_dir`]-rooted path safe for any embedder (a bare
    /// test, a standalone tool, a control plane not yet running a paired
    /// manifest PR) — deliberately NOT a Kubernetes-specific absolute path,
    /// since [`QueryLimits`] carries no notion of "this deployment's real
    /// mounts." A real deployment names its own, deployment-specific value
    /// through its embedding layer — the control plane's
    /// `Config::query_spill_dir` keeps this same portable default and a
    /// Kubernetes deployment overrides it to a dedicated mount via
    /// `POLYCHROME_QUERY_SPILL_DIR` — see that field's own doc for the
    /// ordering dependency on a dedicated mount it imposes.
    ///
    /// # This directory is created EAGERLY, not lazily — a bad path PANICS at construction
    ///
    /// `RuntimeEnvBuilder::build`/`build_arc` (called once, synchronously,
    /// from [`crate::authority::QueryAuthority::new_state_backed`]) creates this
    /// directory (and an initial working subdirectory inside it)
    /// IMMEDIATELY — verified against `datafusion-execution-54.0.0`'s own
    /// `disk_manager.rs::create_local_dirs`, which calls `std::fs::create_dir`
    /// synchronously during `DiskManager::try_new`, itself called from
    /// `RuntimeEnvBuilder::build`. It is NOT deferred to first spill.
    /// `crate::authority::build_base_session_state`'s `.expect(..)` on that call means a
    /// path this process cannot create or write to PANICS the whole
    /// [`crate::authority::QueryAuthority::new_state_backed`] call — i.e. control-plane
    /// STARTUP — not merely the first spilling query. Any embedder pointing
    /// this field at a Kubernetes-specific mount (as the control-plane
    /// deployment does via `POLYCHROME_QUERY_SPILL_DIR`, e.g.
    /// `/var/query-spill`) MUST ensure that mount already exists before this
    /// code path runs — see `Config::query_spill_dir`'s own doc for the exact
    /// ordering requirement.
    pub spill_dir: PathBuf,

    /// Disk quota, in bytes, `DataFusion`'s `DiskManager` enforces against
    /// [`QueryLimits::spill_dir`], via
    /// `RuntimeEnvBuilder::with_max_temp_directory_size` — `DataFusion`
    /// 54.0.0's current API for this (the older `DiskManagerConfig`-based
    /// `with_disk_manager` is deprecated as of 48.0.0 in favor of this
    /// builder pair). Exceeding this quota surfaces
    /// `DataFusionError::ResourcesExhausted`
    /// (`RefCountedTempFile::update_disk_usage`), logged and collapsed into
    /// `crate::authority::ScopedQueryError::Internal` by
    /// `ScopedQuery::execute`, the same as any other execution-time
    /// `DataFusion` failure.
    ///
    /// This crate's OWN [`Self::default`] sizes this as a generic backstop
    /// (16 GiB) against a spilling query filling whatever
    /// [`QueryLimits::spill_dir`] resolves to — appropriate for that field's
    /// own portable, non-Kubernetes-specific default. A real deployment
    /// names its own, tighter, TIER-ORDERED value through its embedding
    /// layer (`Config::query_spill_quota_bytes`, defaulting to 8 GiB,
    /// deliberately below a paired manifest PR's `emptyDir.sizeLimit`,
    /// itself below that container's `ephemeral-storage` limit) — see that
    /// field's own doc for the full three-tier reasoning.
    pub spill_quota_bytes: u64,
}

impl Default for QueryLimits {
    /// 192 MiB / 30 s / 10,000 rows / 500,000 source events / a portable
    /// temp-dir spill path / 16 GiB spill quota.
    ///
    /// `memory_bytes`: deliberately a FRACTION of the pod's own memory
    /// limit, not the whole of it. The control-plane container's memory
    /// limit is 1 GiB (`manifests/base/control-plane-deployment.yaml`);
    /// sizing the pool at 192 MiB — not 1 GiB — leaves roughly 832 MiB of
    /// modeled headroom on that same pod for everything that is NOT this
    /// pool: the control plane's own steady-state working set, the replay
    /// step's `Vec<(u64, Event)>` allocation, `max_source_events`'s
    /// raw-plus-typed Arrow decode fan-out, response JSON serialization,
    /// allocator fragmentation, and any concurrent non-query work the same
    /// process is doing — all of which run entirely OUTSIDE `memory_bytes`'s
    /// `FairSpillPool` accounting (see this field's own doc, and
    /// [`QueryLimits::max_source_events`]'s doc, for exactly what falls
    /// outside the pool). A pool sized to the FULL pod limit would model
    /// zero headroom for any of that and let `DataFusion`'s own accounting
    /// promise a ceiling the pod cannot actually honor once that
    /// out-of-pool cost is added back in. 192 MiB is a conservative,
    /// pilot-oriented starting point, not a measured optimum — an operator
    /// running this deployment should raise it once
    /// `polychrome_query_replayed_bytes` and real pod memory metrics
    /// (`crate::metrics`) show how much of that headroom actually goes
    /// unused for their workload.
    ///
    /// `timeout`: generous for an interactive maintainer query (the spike's
    /// eight queries totalled well under 100 ms of wall time), short enough
    /// that a pathological plan can't hang a forensics request indefinitely.
    ///
    /// `row_cap`: a query response is read by a human or folded into an
    /// agent's own context window either way; 10,000 rows is already far
    /// past what either consumer renders usefully, so it bounds the wire
    /// payload without a caller needing to remember to add `LIMIT`.
    ///
    /// # `max_source_events`: worst-case reasoning, not a fixture extrapolation
    ///
    /// The PREVIOUS default (2,000,000) was sized purely as "two-to-three
    /// orders of magnitude past the spike's own 729-event fixture" — headroom
    /// with no relationship to this deployment's real memory ceiling. Redone
    /// against that ceiling instead: the worst-case math below uses 1024 MiB,
    /// this deployment's actual POD memory limit
    /// (`manifests/base/control-plane-deployment.yaml`) — a distinct number
    /// from `memory_bytes`'s own 192 MiB pool size above, which is
    /// deliberately smaller than the pod limit, not equal to it.
    /// `max_source_events`'s own field doc documents that the raw-plus-typed
    /// decode fan-out `QueryEngine::build` runs materializes roughly an
    /// order of magnitude over the source payload bytes it decodes, entirely
    /// OUTSIDE `memory_bytes`'s pool. At even a modest 2 KiB average event
    /// payload (a typical tool-call or message event — plainly NOT a worst
    /// case; a large tool result or embedded attachment runs far larger),
    /// 500,000 events alone is ~977 MiB of raw replayed payload — already at
    /// or near the ENTIRE 1024 MiB pod limit before the ~10x decode fan-out
    /// multiplies it further to single-digit GiB. There is no event-count
    /// value here that is simultaneously "safe" by a memory-accounting
    /// argument AND permissive enough for a real Fleet-wide history — an
    /// event-COUNT budget cannot be both. 500,000 is chosen only as a
    /// RUNAWAY backstop (a bug enumerating partitions twice, a fleet with
    /// far more partitions than intended), a 4x tighter ceiling than the
    /// previous 2,000,000 default, never a claimed OOM guarantee — see
    /// `crate::engine::QueryLimits::max_source_events`'s own doc for the
    /// full "not a memory bound" framing and issue #1541 for the real,
    /// byte-based fix this stands in for.
    ///
    /// Whoever operates this deployment should CONFIRM 500,000 against this
    /// deployment's own real replayed-byte metrics
    /// (`polychrome_query_replayed_bytes`, `crate::metrics`) rather than
    /// trusting the illustrative 2 KiB figure above — that histogram is
    /// exactly the observability this backstop cannot give on its own, and
    /// is the intended input for tuning this number for a real workload.
    ///
    /// `spill_dir`/`spill_quota_bytes`: deliberately portable, NOT the
    /// Kubernetes-specific values a real deployment should use — see each
    /// field's own doc for why, and for where the deployment-specific
    /// values (and their ordering dependency on a paired manifest PR) live
    /// instead (`crates/control-plane/src/config.rs`'s
    /// `Config::query_spill_dir`/`query_spill_quota_bytes`).
    ///
    /// # `max_source_bytes`: 48 MiB — generous headroom for a real explorer read, inside a 1 GiB pod
    ///
    /// This crate's own tail-replay fix (the byte-bounded cache-tail arm,
    /// `crate::authority::ScopedQuery::resolve_partitions_cached`) closed the
    /// unbounded-materialization gap. Bounding the ALLOCATION and choosing
    /// WHERE to set the rejection threshold are two separate questions, and
    /// this is the second one: the tail arm previously replayed without any
    /// cap, so enforcing the former 16 MiB budget on it would have turned
    /// reads that silently succeeded into browser-visible errors. This
    /// default is raised in the same change that bounds the path, so the
    /// allocation is capped without tightening what a reader can ask for.
    ///
    /// # The arithmetic (a 1 GiB / 1024 MiB pod memory limit, as set in the manifest)
    ///
    /// [`Self::memory_bytes`]'s 192 MiB `FairSpillPool` is unchanged, leaving
    /// ~832 MiB (1024 − 192) for everything outside the pool.
    /// `max_source_events`'s own doc documents the raw-plus-typed decode
    /// fan-out `QueryEngine::build` runs as roughly a 10x multiplier over the
    /// source payload bytes it decodes, entirely OUTSIDE that pool. At 48 MiB
    /// replayed, the fan-out tops out near 480 MiB; replay allocation plus
    /// fan-out land near 528 MiB, and the decode cache's own 64 MiB ceiling
    /// (`crate::cache::CacheConfig::max_bytes`) brings the modeled peak to
    /// ~592 MiB. That leaves ~240 MiB for the control plane's steady-state
    /// working set, response JSON serialization, and allocator fragmentation
    /// — MORE absolute margin than the previous 16 MiB default had under the
    /// old 512 MiB limit (~144 MiB), while admitting 3x the source bytes.
    ///
    /// Every one of those numbers is a fraction of the pod's memory limit, so
    /// they move together or not at all: raising this budget without raising
    /// `manifests/base/control-plane-deployment.yaml`'s `limits.memory` (or
    /// vice versa) invalidates the model above. A deployment that pins
    /// `POLYCHROME_QUERY_MAX_SOURCE_BYTES` explicitly — as
    /// `manifests/components/query-pilot/query-pilot-env.yaml` does — governs
    /// over this compiled-in default (`Config::query_max_source_bytes`,
    /// `crates/control-plane/src/config.rs`), and that env value is raised to
    /// match in the same change.
    ///
    /// This is a pilot-oriented starting point picked to give a real,
    /// legitimately large conversation (many turns, sizeable tool results)
    /// generous room before it trips this budget, not a measured optimum. An
    /// operator should confirm it against `polychrome_query_replayed_bytes`
    /// (`crate::metrics`) and real pod memory metrics once traffic exists.
    fn default() -> Self {
        Self {
            memory_bytes: 192 * 1024 * 1024,
            timeout: Duration::from_secs(30),
            row_cap: 10_000,
            max_source_events: 500_000,
            max_source_bytes: 48 * 1024 * 1024,
            spill_dir: std::env::temp_dir().join(DEFAULT_QUERY_SPILL_DIR_NAME),
            spill_quota_bytes: DEFAULT_QUERY_SPILL_QUOTA_BYTES,
        }
    }
}

/// Default [`QueryLimits::spill_dir`] directory name, created directly under
/// [`std::env::temp_dir`] — portable, always-writable, no Kubernetes
/// assumption. See that field's own doc for why a real deployment overrides
/// this through its own config layer instead of relying on it as-is.
const DEFAULT_QUERY_SPILL_DIR_NAME: &str = "polychrome-query-spill";

/// Default [`QueryLimits::spill_quota_bytes`] — 16 GiB, a generic backstop
/// sized for [`DEFAULT_QUERY_SPILL_DIR_NAME`]'s own portable default, not a
/// Kubernetes deployment's tier-ordered quota. See that field's own doc for
/// where the deployment-specific value lives instead.
const DEFAULT_QUERY_SPILL_QUOTA_BYTES: u64 = 16 * 1024 * 1024 * 1024;

/// The result of one [`QueryEngine::execute`] call.
#[derive(Debug, Clone)]
pub(crate) struct QueryOutput {
    /// The collected result batches, capped at [`QueryLimits::row_cap`] rows
    /// total.
    pub batches: Vec<RecordBatch>,
    /// `true` if the query's real result had more rows than
    /// [`QueryLimits::row_cap`] and was truncated to fit.
    pub truncated: bool,
    /// The `DataFrame`'s planned schema, read off it via `DataFrame::schema`
    /// BEFORE `.collect()` runs — the query's column list as `DataFusion`'s
    /// planner determined it, independent of how many batches (zero,
    /// including) execution actually produced. `crate::output::output_to_json`
    /// derives `columns` from this field rather than `batches.first()` so an
    /// inner join (or any other plan) that eliminates every row still reports
    /// its real column names instead of an empty list indistinguishable from
    /// "no schema could be determined" — see issue #1916.
    pub schema: SchemaRef,
}

/// Failure modes for [`QueryEngine::build`] and [`QueryEngine::execute`].
#[derive(Debug, thiserror::Error)]
pub(crate) enum QueryEngineError {
    /// Decoding a partition's events, or the supplied [`ReferenceData`],
    /// into the named table's Arrow batch failed (`"events_raw"`,
    /// `"usage_raw"`, `"model_call_raw"`, `"attribution_raw"`,
    /// `"turn_failed_raw"`, `"payments_raw"`, `"approvals_raw"`,
    /// `"handoffs_raw"`, `"grant_replays_raw"`, `"summary_raw"`,
    /// `"fires_raw"`, `"messages_raw"`, `"tool_calls_raw"`, `"personas"`,
    /// `"participations"`, `"persona_identities"`, or `"routines"`).
    #[error("failed to decode the `{table}` batch: {source}")]
    Decode {
        /// Which table's decode step failed.
        table: &'static str,
        /// The underlying Arrow error.
        #[source]
        source: ArrowError,
    },
    /// The submitted SQL was rejected before it ever reached the planner —
    /// wrong statement kind, `EXPLAIN` not opted into, or a parse failure.
    #[error(transparent)]
    Rejected(#[from] StatementRejected),
    /// `DataFusion` itself rejected planning or execution — includes an
    /// out-of-scope table failing to resolve (fail-closed catalog scoping,
    /// see the module docs) and memory-pool exhaustion surfaced by the
    /// shared `FairSpillPool` (see [`QueryLimits::memory_bytes`]).
    #[error(transparent)]
    DataFusion(#[from] DataFusionError),
    /// The query did not finish within [`QueryLimits::timeout`].
    #[error("query exceeded its {0:?} timeout")]
    Timeout(Duration),
}

/// The shape `DataFusion` gives an unresolvable table name: a
/// [`DataFusionError::Plan`] whose message is `table '<ref>' not found`,
/// built by its own session state when the catalog has no provider under
/// that name.
///
/// Matched as a prefix/suffix pair rather than a whole string because the
/// planner interpolates the fully-qualified reference between the two.
/// `table function '<name>' not found` deliberately does NOT match: this
/// prefix ends the quote immediately after `table `, and a caller who
/// invoked a missing table FUNCTION has a different problem than one who
/// guessed a table name.
const UNRESOLVED_TABLE_PREFIX: &str = "table '";
/// See [`UNRESOLVED_TABLE_PREFIX`].
const UNRESOLVED_TABLE_SUFFIX: &str = "' not found";

/// Whether `error` is `DataFusion` failing to resolve a table NAME — the one
/// engine failure that is the caller's own to fix (issue #2147), rather than
/// a decode, replay, memory, or execution fault.
///
/// Walks the error chain instead of testing only the outermost value: the
/// planner wraps its own `Plan` error in a
/// [`DataFusionError::Diagnostic`] before it leaves the SQL planner, and a
/// query with several unresolvable relations arrives as a
/// [`DataFusionError::Collection`]. `crate::authority::ScopedQuery::execute`
/// is the only caller; it turns a `true` here into
/// `crate::authority::ScopedQueryError::UnknownTable` and everything else
/// into `crate::authority::ScopedQueryError::Internal`.
pub(crate) fn is_unresolved_table_error(error: &QueryEngineError) -> bool {
    let QueryEngineError::DataFusion(error) = error else {
        return false;
    };
    let mut pending = vec![error];
    while let Some(current) = pending.pop() {
        match current {
            DataFusionError::Plan(message)
                if message.starts_with(UNRESOLVED_TABLE_PREFIX)
                    && message.ends_with(UNRESOLVED_TABLE_SUFFIX) =>
            {
                return true;
            }
            DataFusionError::Context(_, inner) | DataFusionError::Diagnostic(_, inner) => {
                pending.push(inner);
            }
            DataFusionError::Shared(inner) => pending.push(inner),
            DataFusionError::Collection(inner) => pending.extend(inner),
            _ => {}
        }
    }
    false
}

/// `DataFusion`'s own answer to an unresolvable COLUMN name, lifted out of
/// its error vocabulary into plain strings (issue #2206).
///
/// Produced by [`unresolved_column`]; rendered into a caller-facing message
/// by `crate::authority::unknown_column_message`. The two halves are split
/// so the message is BUILT from what the planner actually resolved against,
/// rather than scraped back out of a `Display` string that could change
/// wording under us — the same posture
/// `crate::authority::CONVERSATION_CATALOG` gives the table case.
pub(crate) struct UnresolvedColumn {
    /// The column name the caller wrote, with any qualifier stripped off —
    /// `"message"` for both `message` and `events.message`.
    pub(crate) name: String,
    /// The table the caller qualified that name with, if they qualified it
    /// at all — `Some("events")` for `events.message`, `None` for `message`.
    pub(crate) qualifier: Option<String>,
    /// Every column that WAS available at the point the planner tried to
    /// resolve this name, in the order it offered them, as
    /// `(qualifier, name)`.
    ///
    /// Read off `DataFusion`'s `valid_fields`, which is the resolution
    /// site's schema — NOT, in general, every relation the whole statement
    /// selects from. A name inside a correlated subquery resolves against
    /// the subquery's own relations, so that is all this carries; a miss in
    /// `GROUP BY`, `ORDER BY`, or `HAVING` resolves against the projection
    /// schema followed by the input schema, which is why
    /// [`unresolved_column`] deduplicates. Either way the schemas are those
    /// of relations that ALREADY RESOLVED, so for a redacted scope this
    /// already omits the columns that scope's views do not build (see this
    /// module's docs' redaction invariants).
    pub(crate) valid_fields: Vec<(Option<String>, String)>,
}

/// `DataFusion`'s structured account of a column name it could not resolve —
/// the second engine failure that is the caller's own to fix (issue #2206),
/// after the unresolvable table name [`is_unresolved_table_error`] detects.
///
/// Returns `None` for every other failure. Unlike the table case, which the
/// planner reports only as a [`DataFusionError::Plan`] message this crate has
/// to match by prefix and suffix, an unresolvable column arrives as a typed
/// `SchemaError::FieldNotFound` carrying both the name the caller wrote and
/// the columns the query's own schema does have, so nothing here parses text.
///
/// Walks the error chain for the same reason [`is_unresolved_table_error`]
/// does: the SQL planner wraps its own error in a
/// [`DataFusionError::Diagnostic`] (and a `Context`, and a `Collection` when
/// several expressions fail) before it leaves the planner, so testing only
/// the outermost value would let a wrapped error through as a generic
/// internal fault.
///
/// What keeps the columns named here inside the caller's own scope is the
/// construction of `valid_fields` itself, not any planning order: `DataFusion`
/// builds it from the `DFSchema`s of relations that ALREADY RESOLVED, so it
/// can only ever hold columns of tables this session registered — and, for a
/// redacted scope, only the columns that scope's views build. That holds
/// however the planner interleaves relation and expression resolution, which
/// it genuinely does interleave: a `UNION ALL` arm and a scalar subquery each
/// emit a `FieldNotFound` for the in-scope arm ALONGSIDE the out-of-scope
/// table's `Plan` error, in one [`DataFusionError::Collection`].
/// `crate::authority::ScopedQuery::execute_with_params` still asks
/// [`is_unresolved_table_error`] FIRST, so those mixed failures refuse as the
/// table they named; that ordering decides which refusal a caller reads, not
/// whether the offer is safe.
///
/// Deduplicates `valid_fields` on `(qualifier, name)`, keeping first-seen
/// order. A miss in `GROUP BY`, `ORDER BY`, or `HAVING` resolves against the
/// projection schema followed by the input schema, so every projected column
/// arrives twice; offering a caller `turn_id` twice reads as a bug and
/// doubles the width the caller-facing bound has to cover.
///
/// A statement with several bad columns arrives as a `Collection` of one
/// `FieldNotFound` each, and this reports the LAST of them — `SELECT nope1,
/// nope2` names `nope2`. Any of them is a real, correctable answer, and a
/// caller fixing one at a time reaches the same place.
pub(crate) fn unresolved_column(error: &QueryEngineError) -> Option<UnresolvedColumn> {
    let QueryEngineError::DataFusion(error) = error else {
        return None;
    };
    let mut pending = vec![error];
    while let Some(current) = pending.pop() {
        match current {
            DataFusionError::SchemaError(schema_error, _) => {
                if let SchemaError::FieldNotFound {
                    field,
                    valid_fields,
                } = schema_error.as_ref()
                {
                    let mut seen = HashSet::new();
                    return Some(UnresolvedColumn {
                        name: field.name.clone(),
                        qualifier: bare_table_name(field),
                        valid_fields: valid_fields
                            .iter()
                            .map(|column| (bare_table_name(column), column.name.clone()))
                            .filter(|field| seen.insert(field.clone()))
                            .collect(),
                    });
                }
            }
            DataFusionError::Context(_, inner) | DataFusionError::Diagnostic(_, inner) => {
                pending.push(inner);
            }
            DataFusionError::Shared(inner) => pending.push(inner),
            DataFusionError::Collection(inner) => pending.extend(inner),
            _ => {}
        }
    }
    None
}

/// `column`'s relation as the bare table name a caller would have typed —
/// `"events"`, never `"datafusion.public.events"`.
///
/// Every table this crate registers lives in the one default catalog and
/// schema, so the catalog and schema parts of a `TableReference` carry no
/// information a caller needs; dropping them is also what lets
/// `crate::authority::unknown_column_message` compare the qualifier the
/// caller wrote against the qualifiers the planner resolved.
fn bare_table_name(column: &Column) -> Option<String> {
    column
        .relation
        .as_ref()
        .map(|relation| relation.table().to_owned())
}

/// A scoped `DataFusion` `SessionContext`, built once by [`QueryEngine::build`]
/// and reused for every query the session runs.
///
/// See the module docs for how table registration enforces catalog scoping.
pub(crate) struct QueryEngine {
    ctx: SessionContext,
    limits: QueryLimits,
}

/// Test-only: counts every entry into this crate's ENGINE-ASSEMBLY step —
/// [`QueryEngine::build_from_tables`] (and, through it, the test-only
/// [`QueryEngine::build`] wrapper) — across the whole test binary. Before
/// Phase A's decode cache, this doubled as a decode-fan-out counter too
/// (`QueryEngine::build` decoded every partition inline, right before
/// assembling the engine); since `crate::cache` moved decode itself out to
/// [`decode_partition_tables`] — called from
/// `crate::authority::ScopedQuery::resolve_partitions` BEFORE this function
/// ever runs — this counter alone no longer proves decode never happened,
/// only that engine assembly didn't. `crate::engine::partition_tables::DECODE_CALL_COUNT`
/// is the counter that still proves that; see its own doc, and
/// `crate::authority::tests::fleet_query_over_source_budget_is_rejected_before_decode`,
/// which diffs THAT one instead of this one for exactly this reason. A
/// plain `AtomicUsize` (not a per-test reset) because `cargo nextest` runs
/// each test in its own process by default, so cross-test interference is
/// not a concern here the way it would be under a single shared-process
/// runner; the assertion this backs is always a same-process before/after
/// diff, never an absolute value.
#[cfg(test)]
pub(crate) static BUILD_CALL_COUNT: std::sync::atomic::AtomicUsize =
    std::sync::atomic::AtomicUsize::new(0);

impl QueryEngine {
    /// Build a `QueryEngine` scoped to `scope` over `partitions` and
    /// `reference`, under `limits`.
    ///
    /// Registers [`EVENTS_RAW_TABLE`]/[`ATTRIBUTION_RAW_TABLE`]/
    /// [`PAYMENTS_RAW_TABLE`]/[`MESSAGES_RAW_TABLE`]/[`TOOL_CALLS_RAW_TABLE`]/
    /// [`APPROVALS_RAW_TABLE`]/[`HANDOFFS_RAW_TABLE`]/[`USAGE_RAW_TABLE`]/
    /// [`MODEL_CALL_RAW_TABLE`]/[`TURN_FAILED_RAW_TABLE`]/[`SUMMARY_RAW_TABLE`]
    /// (Fleet-only after this call returns), [`EVENTS_VIEW`] (every scope,
    /// identical columns for every scope), [`ATTRIBUTION_TABLE`] (every scope,
    /// but with the four `identity_*` columns present only for
    /// [`QueryScope::Fleet`] — see the module docs' "Identity redaction
    /// invariant" section), [`PAYMENTS_TABLE`]
    /// (every scope, but with `signer_public_key` present only for
    /// [`QueryScope::Fleet`] — see the module docs' "Payment signer-key
    /// redaction invariant" section), [`MESSAGES_TABLE`]/[`TOOL_CALLS_TABLE`]
    /// (every scope, but with `internal_only = true` rows present only for
    /// [`QueryScope::Fleet`] — see the module docs' "Message-content
    /// redaction invariant" section), [`APPROVALS_TABLE`] (every scope, but
    /// with its nine Fleet-only columns present only for
    /// [`QueryScope::Fleet`] — see the module docs' "Approval redaction
    /// invariant" section), [`HANDOFFS_TABLE`] (every scope, but with
    /// `signed_by` present only for [`QueryScope::Fleet`] — see the module
    /// docs' "Handoff signed-by redaction invariant" section),
    /// [`USAGE_TABLE`]/[`MODEL_CALL_TABLE`]/[`TURN_FAILED_TABLE`] (every
    /// scope, identical columns, but filtered to committed turns — see the
    /// module docs' "Committed-turn filter invariant" section),
    /// [`SUMMARY_TABLE`] (Fleet-only — the one documented exception to that
    /// filter), and [`PERSONAS_TABLE`]/
    /// [`PARTICIPATIONS_TABLE`] (Fleet-only, from `reference`); see the
    /// module docs for the exact sequence and why raw rows, attribution's
    /// identity columns, payments' signer key, message-content's
    /// `internal_only` rows, approvals' Fleet-only columns, handoffs'
    /// signed-by column, uncommitted-turn rows, and the reference tables all
    /// stay hidden from non-Fleet scopes. `trusted_signers` is the
    /// deployment's approval-signer allow-list, threaded through to
    /// [`crate::decode::payments::decode_payments_events`] AND
    /// [`crate::decode::approvals::decode_approvals_events`] — the two
    /// typed tables this crate decodes through a signature-verification fold
    /// rather than a plain protobuf decode; see `crate::decode::payments`'s
    /// and `crate::decode::approvals`'s module docs.
    /// [`crate::decode::handoffs::decode_handoffs_events`] uses the handoff
    /// role's trust set (`#1124`). This test wrapper pins
    /// [`fixture_handoff_trust`].
    /// Also registers the `datafusion-functions-json` scalar UDFs
    /// (`json_get` and its typed siblings, plus the `->`/`->>` operator
    /// rewrite) for EVERY scope, Fleet included — see `crate::decode`'s
    /// module docs for the query-side contract. Unlike every table above,
    /// this registration has no scope-dependent variant: these are pure
    /// scalar functions over column values a query already selected, adding
    /// SQL expressiveness but no new data access.
    /// `information_schema` is enabled only for [`QueryScope::Fleet`], and
    /// `enable_url_table` is never called for any scope.
    ///
    /// # Errors
    ///
    /// Returns [`QueryEngineError::Decode`] if a partition's events, or
    /// `reference`'s persona/participation records, fail to decode into
    /// Arrow, and [`QueryEngineError::DataFusion`] if table/view construction
    /// itself fails (e.g. a schema mismatch across supplied partitions).
    ///
    /// # Shared base state, isolated catalog (A2 retrofit)
    ///
    /// `base_state` is [`crate::authority::QueryAuthority`]'s ONE process-wide
    /// prebuilt [`SessionState`] — the shared function/analyzer/optimizer
    /// registry (expensive to build: `SessionStateBuilder::with_default_features`
    /// constructs every built-in scalar/aggregate/window function) and the ONE
    /// shared, [`datafusion::execution::memory_pool::FairSpillPool`]-backed
    /// `RuntimeEnv` (docs/reference/datafusion-data-layer.md's 2026-07-21
    /// decision: a shared *greedy* pool lets one query starve its siblings
    /// into `ResourcesExhausted`; `FairSpillPool` is `DataFusion`'s own
    /// prescription for multiple concurrent spillable consumers sharing one
    /// ceiling). `SessionStateBuilder::new_from_existing(base_state.clone())`
    /// reuses both by CLONING `base_state` — cheap, because `SessionState`'s
    /// function registries are `HashMap`s of `Arc<dyn ..>`, so cloning bumps
    /// refcounts rather than reconstructing ~200 built-in functions from
    /// scratch every request.
    ///
    /// The one field that must NEVER be inherited from `base_state` is the
    /// catalog list: `SessionState` derives `Clone`, and `catalog_list` is an
    /// `Arc<dyn CatalogProviderList>` — cloning `SessionState` alone would
    /// hand every request the SAME mutable catalog/schema registry every
    /// OTHER request (past, present, and future) also registers tables into,
    /// which is exactly the cross-session leak this crate's whole catalog-
    /// scoping design exists to prevent. `.with_catalog_list(Arc::new(
    /// MemoryCatalogProviderList::new()))` overrides it with a FRESH, empty
    /// list before `.build()`, so this call's `ctx.register_table`/`CREATE
    /// VIEW`/`deregister_table` calls below land in a catalog only this one
    /// `QueryEngine` ever sees. `.with_config(..)` is likewise supplied fresh
    /// (not inherited) so this scope's own `information_schema` policy — not
    /// whatever the base template happened to carry — takes effect and so the
    /// fresh catalog list actually gets a default catalog/schema created into
    /// it (`SessionConfig::create_default_catalog_and_schema` defaults to
    /// `true`, but `new_from_existing` would otherwise compute `false` from
    /// `base_state`'s ALREADY-populated catalog, leaving the fresh list with
    /// nowhere for `register_table` to write).
    // Test-only since Phase A: `crate::authority::ScopedQuery::execute` now
    // calls `Self::build_from_tables` directly (resolving partitions through
    // `crate::cache` first) — this wrapper's only remaining caller is
    // `engine::tests`'s existing `PartitionEvents`-shaped fixtures, which
    // this wrapper lets keep calling `QueryEngine::build` unchanged rather
    // than rewriting every one of them to pre-decode into `PartitionTables`
    // by hand.
    #[cfg(test)]
    pub(crate) async fn build(
        base_state: &SessionState,
        scope: &QueryScope,
        partitions: Vec<PartitionEvents>,
        reference: ReferenceData,
        trusted_signers: &[Vec<u8>],
        limits: QueryLimits,
    ) -> Result<Self, QueryEngineError> {
        Self::build_scoped(
            base_state,
            scope,
            partitions,
            reference,
            trusted_signers,
            limits,
            false,
        )
        .await
    }

    /// The same as [`Self::build`], but with an explicit choice of whether
    /// this build registers owner-scoped `routines`/`fires` (issue #1882) —
    /// `crate::authority::tests`' own fixture for a persona-scoped session
    /// that owns a routine calls this directly rather than [`Self::build`],
    /// which always passes `false` (matching every pre-#1882 fixture's
    /// existing expectations unchanged).
    ///
    /// # Errors
    ///
    /// See [`Self::build`].
    #[cfg(test)]
    pub(crate) async fn build_scoped(
        base_state: &SessionState,
        scope: &QueryScope,
        partitions: Vec<PartitionEvents>,
        reference: ReferenceData,
        trusted_signers: &[Vec<u8>],
        limits: QueryLimits,
        register_owner_scoped_routines: bool,
    ) -> Result<Self, QueryEngineError> {
        let handoff_trust = fixture_handoff_trust();
        let partition_tables = partitions
            .iter()
            .map(|partition| {
                decode_partition_tables(
                    &partition.partition,
                    &partition.events,
                    trusted_signers,
                    &handoff_trust,
                )
            })
            .collect::<Result<Vec<_>, _>>()?;
        Self::build_from_tables(
            base_state,
            scope,
            partition_tables,
            reference,
            limits,
            register_owner_scoped_routines,
        )
        .await
    }

    /// Build a `QueryEngine`, the same as [`Self::build`], but from
    /// ALREADY-DECODED [`PartitionTables`] rather than raw
    /// [`PartitionEvents`] — the entry point `crate::authority::ScopedQuery`
    /// uses when `crate::cache`'s decode cache serves some or all of a
    /// scope's partitions from a prior decode (a hit, or a tail merged onto
    /// a hit) instead of decoding them again here. [`Self::build`] itself is
    /// now a thin wrapper: it decodes every supplied partition via
    /// [`decode_partition_tables`] (the identical decode
    /// [`crate::cache`]'s own miss/tail path calls, so a cold query and a
    /// cache-populated one decode through the SAME code, never two
    /// independently-maintained copies) and delegates here.
    ///
    /// `register_owner_scoped_routines` (issue #1882) is `true` iff this
    /// build is for a verified persona-scoped session (never Fleet, never a
    /// conversation grant — `crate::authority::ScopedQuery::execute` derives
    /// it from whether this session's scope is
    /// [`QueryScope::Conversations`] AND its own `caller_identity` resolved
    /// to a persona id). When `true`, `routines`/`fires` are registered for
    /// this otherwise-Fleet-only-reference-table build too, using
    /// `reference.routines` — already filtered to that persona's own rows by
    /// `crate::authority::ScopedQuery::resolve_routines` — joined against for
    /// `fires` exactly as [`crate::views::FIRES_OWNED_VIEW_SQL`] describes.
    /// Ignored when `scope` is [`QueryScope::Fleet`]: that scope already
    /// registers both tables unfiltered, regardless of this argument.
    ///
    /// # Errors
    ///
    /// Returns [`QueryEngineError::DataFusion`] if table/view construction
    /// itself fails (e.g. a schema mismatch across supplied partitions).
    pub(crate) async fn build_from_tables(
        base_state: &SessionState,
        scope: &QueryScope,
        partition_tables: Vec<PartitionTables>,
        reference: ReferenceData,
        limits: QueryLimits,
        register_owner_scoped_routines: bool,
    ) -> Result<Self, QueryEngineError> {
        // Test-only probe (item F, the QRY-3 hardening review): this is the
        // ONE entry point into this crate's engine-ASSEMBLY step —
        // `QueryLimits::max_source_events`'s doc names the decode fan-out
        // this gates. `crate::authority::tests` diffs this counter across a
        // source-budget rejection to prove engine assembly was never
        // reached. Decode itself may already have happened earlier, via
        // `crate::cache`'s Hit/Tail/Miss resolution inside
        // `crate::authority::ScopedQuery::resolve_partitions` — this counter
        // does NOT pin that; `crate::authority::ScopedQuery::execute` instead
        // enforces the QRY-3 event-count budget from an O(1)
        // `EventLogHost::partition_event_count` sum BEFORE `resolve_partitions`
        // (and therefore before any cache lookup or decode) ever runs, so a
        // rejected query never reaches decode either — see that method's own
        // doc, and `crate::engine::partition_tables::DECODE_CALL_COUNT`,
        // which is the counter that actually pins THAT invariant in a test.
        #[cfg(test)]
        BUILD_CALL_COUNT.fetch_add(1, std::sync::atomic::Ordering::SeqCst);

        let is_fleet = matches!(scope, QueryScope::Fleet);
        // `enable_url_table` is deliberately never called here, for either
        // scope — see the module docs.
        let config = SessionConfig::new().with_information_schema(is_fleet);
        let state = SessionStateBuilder::new_from_existing(base_state.clone())
            .with_config(config)
            .with_catalog_list(Arc::new(MemoryCatalogProviderList::new()))
            .build();
        let mut ctx = SessionContext::new_with_state(state);

        // In-SQL JSON navigation (`json_get`, `->`, `->>`, `json_contains`,
        // ...) over `payload_json` and every other JSON-string column
        // (`tool_calls.arguments`/`.result`) — see `crate::decode`'s module
        // docs for why the pinned `datafusion = "=54.0.0"` ships none of
        // this itself. Registered identically for EVERY scope, Fleet
        // included, unlike the table registrations above/below: these are
        // pure scalar functions over column values a query already has in
        // hand, not a new way to reach a row or column a scope couldn't
        // already select — there is no scope-dependent reason to withhold
        // them, and registering them before any table/view exists (so
        // before any user SQL can run) means every subsequent `ctx.sql`
        // call in this method, and every query a caller submits later,
        // already sees them.
        register_json_functions(&mut ctx)?;

        let raw_batches: Vec<RecordBatch> = partition_tables
            .iter()
            .map(|tables| tables.events_raw.clone())
            .collect();
        register_typed_table(
            &ctx,
            EVENTS_RAW_TABLE,
            EventsTableProvider::schema(),
            raw_batches,
        )?;

        register_typed_journal_tables(&ctx, &partition_tables)?;

        register_message_content_tables(&ctx, &partition_tables)?;

        let create_view_sql = format!("CREATE VIEW {EVENTS_VIEW} AS {COMMITTED_TURNS_VIEW_SQL}");
        ctx.sql(&create_view_sql).await?.collect().await?;

        // #1882: `routines` registers BEFORE `create_scope_dependent_views`
        // (rather than alongside every other reference table, further
        // below) whenever this build needs it at all — Fleet always, a
        // persona-scoped session when `register_owner_scoped_routines` is
        // set — because the persona-scoped `fires` view
        // (`crate::views::FIRES_OWNED_VIEW_SQL`) joins against `routines` by
        // uid, so `routines` must already be resolvable by the time that
        // `CREATE VIEW` runs. `register_reference_tables` (further below,
        // alongside `personas`/`participations`/...) no longer registers
        // `routines` at all — this is its ONLY registration point now, for
        // both Fleet and a persona-scoped session.
        //
        // `RegistrationScope` (finding 5 of #1882's review) folds `is_fleet`
        // and `register_owner_scoped_routines` into one three-variant enum
        // rather than a boolean pair whose one illegal combination (`is_fleet
        // && register_owner_scoped_routines`) used to need a defensive
        // re-derivation — see that type's own doc.
        let registration_scope =
            registration::RegistrationScope::new(scope, register_owner_scoped_routines);
        if registration_scope.is_fleet() || registration_scope.is_owner() {
            register_routines_table(&ctx, &reference)?;
        }

        create_scope_dependent_views(&ctx, registration_scope).await?;

        if !is_fleet {
            // Fail-closed hiding of the raw tables — see the module docs
            // for why the `events`/`attribution`/`payments`/`messages`/
            // `tool_calls`/`approvals`/`usage`/`model_call`/`turn_failed`
            // views keep working after this call. Deregistering
            // `turn_failed_raw` here (not just `usage_raw`/`model_call_raw`)
            // is what actually closes the QRY-1 gap for `turn_failed`: the
            // `turn_failed` VIEW's committed-turn semijoin is worthless if a
            // Conversations-scoped session can route around it with `SELECT
            // * FROM turn_failed_raw` directly, the exact bypass this
            // deregistration already prevents for every other filtered
            // typed table. `summary_raw` is deregistered here too even
            // though the `summary` VIEW itself was never created for this
            // scope in the first place (see `create_scope_dependent_views`)
            // — belt-and-suspenders, the same posture every other `*_raw`
            // table gets, so a future change that accidentally started
            // creating a non-Fleet `summary` view would still find
            // `summary_raw` gone.
            ctx.deregister_table(EVENTS_RAW_TABLE)?;
            ctx.deregister_table(ATTRIBUTION_RAW_TABLE)?;
            ctx.deregister_table(PAYMENTS_RAW_TABLE)?;
            // `refusals_raw` (`#2090`, INV-W5): the identical fail-closed
            // deregistration `payments_raw` gets just above — without this,
            // a non-Fleet session could route around
            // `REFUSALS_REDACTED_VIEW_SQL`'s redaction with `SELECT
            // signer_public_key FROM refusals_raw` directly. See
            // `engine::tests::conversations_scope_refusals_signer_public_key_is_unreachable`.
            ctx.deregister_table(REFUSALS_RAW_TABLE)?;
            // `wallet_link_lifecycle_raw` (`#2123`): the identical
            // fail-closed deregistration `refusals_raw` gets just above —
            // without this, a non-Fleet session could route around
            // `WALLET_LINK_LIFECYCLE_REDACTED_VIEW_SQL`'s redaction with
            // `SELECT signer_public_key FROM wallet_link_lifecycle_raw`
            // directly. This was the single highest-risk omission in
            // `#2090`'s own rollout (a review finding on that PR) — see
            // `engine::tests::conversations_scope_wallet_link_lifecycle_signer_public_key_is_unreachable`.
            ctx.deregister_table(WALLET_LINK_LIFECYCLE_RAW_TABLE)?;
            ctx.deregister_table(MESSAGES_RAW_TABLE)?;
            ctx.deregister_table(TOOL_CALLS_RAW_TABLE)?;
            ctx.deregister_table(APPROVALS_RAW_TABLE)?;
            ctx.deregister_table(HANDOFFS_RAW_TABLE)?;
            ctx.deregister_table(GRANT_REPLAYS_RAW_TABLE)?;
            ctx.deregister_table(USAGE_RAW_TABLE)?;
            ctx.deregister_table(MODEL_CALL_RAW_TABLE)?;
            ctx.deregister_table(TURN_FAILED_RAW_TABLE)?;
            ctx.deregister_table(TURN_DISPATCH_RAW_TABLE)?;
            ctx.deregister_table(SUMMARY_RAW_TABLE)?;
            // `fires_raw` is ALWAYS deregistered here, owner-scoped session
            // included (INV-OAF18 — a review finding on issue #1882's own
            // rollout): a persona-scoped session runs arbitrary SQL, so
            // leaving `fires_raw` resolvable would let `SELECT * FROM
            // fires_raw` return EVERY member's fires, unfiltered, bypassing
            // the owner-filtering join `FIRES_OWNED_VIEW_SQL` applies. This
            // is safe for the SAME reason deregistering `events_raw` never
            // breaks `events`: `create_scope_dependent_views` already built
            // this scope's own `fires` view (owner-filtered or absent) via
            // `CREATE VIEW`, which bakes the *resolved* `TableProvider`
            // straight into the view's stored `LogicalPlan` before this
            // deregister ever runs — deregistering the raw name afterward
            // does not disturb it. See
            // `engine::tests::owner_scoped_session_cannot_resolve_fires_raw_directly`.
            ctx.deregister_table(FIRES_RAW_TABLE)?;
            // Belt-and-suspenders, the same posture `fires_raw` gets just
            // above even though the `routine_lifecycle` VIEW itself was
            // never created for this scope in the first place (see
            // `create_scope_dependent_views`) — a future change that
            // accidentally started creating a non-Fleet `routine_lifecycle`
            // view would still find `routine_lifecycle_raw` gone (issue
            // #1593).
            ctx.deregister_table(ROUTINE_LIFECYCLE_RAW_TABLE)?;
            // `routine_setup_raw` : the identical fail-closed
            // deregistration `fires_raw` gets above — without this, a
            // persona-scoped session could route around the owner-filtering
            // join `ROUTINE_SETUP_OWNED_VIEW_SQL` applies with `SELECT *
            // FROM routine_setup_raw` directly.
            ctx.deregister_table(ROUTINE_SETUP_RAW_TABLE)?;
        }

        // Reference-data tables: registered ONLY for Fleet — never even
        // decoded/built for a non-Fleet scope, so a `Conversations` session
        // pays no cost for data it must never see (see the module docs'
        // "Reference data" section for why these tables are Fleet-only).
        // `routines` is the one exception (issue #1882) and is NOT
        // registered here — it already registered ABOVE, before
        // `create_scope_dependent_views` ran (for `RegistrationScope::Fleet`
        // always, and for `RegistrationScope::Owner`), so the owner-scoped
        // `fires` view could join against it; see the `registration_scope`
        // branch above and `register_reference_tables`'s own doc.
        if is_fleet {
            register_reference_tables(&ctx, &reference)?;
        }

        Ok(Self { ctx, limits })
    }

    /// Run `sql` against this session's catalog, honoring [`QueryLimits`].
    ///
    /// Pipeline: [`statement_gate::check_statement_allowed`] (rejecting
    /// anything but a single query statement, or a single plain `EXPLAIN` of
    /// one when `allow_explain` is `true` — see that function's docs) →
    /// `ctx.sql_with_options(sql, ..)` (belt-and-suspenders: `DataFusion`'s
    /// own DDL/DML/statement guard runs underneath the hand-written gate
    /// above — see below) → for a plain query (not `EXPLAIN`), push a
    /// logical `Limit` of [`QueryLimits::row_cap`] `+ 1` into the plan →
    /// collect under [`QueryLimits::timeout`] → the row cap applied across
    /// every collected batch combined.
    ///
    /// # Belt-and-suspenders: `SQLOptions` beneath the statement gate
    ///
    /// [`statement_gate::check_statement_allowed`] is the primary gate and
    /// stays exactly as strict — it does more than `SQLOptions` can express
    /// (multi-statement rejection, the `EXPLAIN`-opt-in policy, `EXPLAIN
    /// ANALYZE` banned unconditionally). This method also runs every query
    /// through `ctx.sql_with_options(sql, SQLOptions::new()
    /// .with_allow_ddl(false).with_allow_dml(false)
    /// .with_allow_statements(false))` rather than plain `ctx.sql(sql)` — a
    /// second, independent layer straight from `DataFusion` 54 itself
    /// (`SQLOptions::verify_plan`, `datafusion-54.0.0/src/execution/context/
    /// mod.rs`), so a bug in the hand-written gate above isn't the only
    /// thing standing between a submitted string and a DDL/DML plan actually
    /// executing.
    ///
    /// This is safe to apply unconditionally, including to an opted-in
    /// `EXPLAIN`: `SQLOptions::verify_plan` walks the planned
    /// [`datafusion::logical_expr::LogicalPlan`] and only rejects the
    /// `Ddl`, `Dml`/`Copy`, and `Statement` variants — `EXPLAIN` plans to
    /// its own distinct `LogicalPlan::Explain` variant, which none of those
    /// arms match, so `with_allow_statements(false)` does not reject it. No
    /// branching on [`AllowedStatement`] is needed here; the same
    /// `SQLOptions` value covers both `Query` and `Explain`.
    ///
    /// The `Limit` push (`DataFrame::limit(0, Some(row_cap.saturating_add(1)))`)
    /// bounds the number of rows `QueryEngine::execute` ever RETURNS to
    /// `row_cap + 1`, not just what it returns after collecting — collecting
    /// the full result set and slicing afterward, this method's prior
    /// behavior, held every row in memory regardless of `row_cap` before the
    /// cap was ever applied. `DataFusion` 54's `DataFrame::limit` prepends a
    /// `Limit` node to the logical plan (`LogicalPlanBuilder::limit`), and
    /// the `PushDownLimit` optimizer rule pushes that node toward the scan
    /// where the plan shape allows it, so a plain `SELECT *` over a huge
    /// history can skip scanning rows past the cap entirely. This does NOT
    /// bound every operator's intermediate state, though: a plan with an
    /// `ORDER BY`, `GROUP BY`, `JOIN`, or window function still builds that
    /// operator's own full intermediate result — the sort buffer, the hash
    /// table, the join's build side — before the `Limit` node downstream of
    /// it ever trims anything, because a limit cannot be pushed through an
    /// operator that has to see every input row to produce a correct output
    /// row. The `+ 1` (rather than exactly
    /// `row_cap`) is so the row-cap trim below can still distinguish
    /// "exactly `row_cap` rows" from "more than `row_cap`, truncated" — the
    /// final trim from `row_cap + 1` down to `row_cap` is unchanged from
    /// before this bound existed. `saturating_add` (rather than a bare `+`)
    /// is deliberate: a caller-configured `row_cap` of `usize::MAX` is a
    /// legitimate "no cap" configuration, and a bare `+ 1` would overflow
    /// (panicking in a debug/overflow-checked build) instead of saturating
    /// at `usize::MAX` — the trim below already treats that value as
    /// effectively unbounded either way.
    ///
    /// The `Limit` push is skipped for an opted-in `EXPLAIN`
    /// ([`AllowedStatement::Explain`]): `EXPLAIN`'s output is plan text, not
    /// the wrapped query's own row stream, so a `Limit` wrapped around an
    /// `Explain` plan node doesn't bound anything the row cap is meant to
    /// bound, and the plan text itself is always a handful of rows
    /// regardless of the wrapped query's size.
    ///
    /// A query that times out is abandoned, not cancelled mid-execution:
    /// `tokio::time::timeout` drops the collect future, which drops
    /// `DataFusion`'s execution stream and its share of the shared `FairSpillPool`
    /// reservations along with it — there is no separate cleanup step this
    /// method owes the caller.
    ///
    /// `params` are bound to `sql`'s `$1`, `$2`, ... placeholders as UTF-8
    /// literals; a caller with a fixed statement and nothing to bind passes
    /// an empty slice. Binding happens on the PLANNED logical plan
    /// (`DataFrame::with_param_values`), never by substituting text into
    /// `sql`: a bound value becomes one `Expr::Literal` node in a tree the
    /// planner already finished building, so no part of a caller-supplied
    /// value is ever parsed as SQL. That is what makes a fixed template plus
    /// bound values structurally injection-proof rather than
    /// escaping-dependent, which is what a surface offering a fixed-scope
    /// tool over untrusted text is built on. This crate's own
    /// `a_bound_parameter_is_a_value_and_never_sql` pins it against the
    /// engine, from both sides.
    ///
    /// A mismatch between `sql`'s placeholders and `params` is caught in ONE
    /// direction only. A placeholder with no value is refused
    /// ([`QueryEngineError::DataFusion`], `Placeholder '$1' was not provided
    /// a value for execution`); a value with no placeholder is SILENTLY
    /// IGNORED and the query runs. A caller that builds an optional
    /// predicate must therefore derive the clause and its value from one
    /// decision, or a dropped clause leaves a surplus value behind and the
    /// query returns every row unfiltered — the direction that fails open is
    /// the one nothing here checks. This crate's own
    /// `a_surplus_parameter_is_ignored_while_a_missing_one_is_refused` pins
    /// both halves against the engine.
    ///
    /// # Errors
    ///
    /// Returns [`QueryEngineError::Rejected`] if `sql` fails the statement
    /// gate, [`QueryEngineError::DataFusion`] if planning or execution fails
    /// (including an out-of-scope table failing to resolve, or the shared
    /// `FairSpillPool` refusing an allocation), and
    /// [`QueryEngineError::Timeout`] if the query is still running when
    /// [`QueryLimits::timeout`] elapses.
    pub(crate) async fn execute_with_params(
        &self,
        sql: &str,
        params: &[&str],
        allow_explain: bool,
    ) -> Result<QueryOutput, QueryEngineError> {
        let statement = statement_gate::check_statement_allowed(sql, allow_explain)?;

        // Second, independent layer beneath the gate above — see this
        // method's doc for why the same options apply to both `Query` and
        // `Explain` without branching.
        let sql_options = SQLOptions::new()
            .with_allow_ddl(false)
            .with_allow_dml(false)
            .with_allow_statements(false);
        let dataframe = self.ctx.sql_with_options(sql, sql_options).await?;
        let dataframe = if params.is_empty() {
            dataframe
        } else {
            let values: Vec<ScalarValue> = params
                .iter()
                .map(|param| ScalarValue::Utf8(Some((*param).to_owned())))
                .collect();
            dataframe.with_param_values(values)?
        };
        let dataframe = match statement {
            AllowedStatement::Query => {
                dataframe.limit(0, Some(self.limits.row_cap.saturating_add(1)))?
            }
            AllowedStatement::Explain => dataframe,
        };
        // Read the planned schema off the `DataFrame` before `.collect()`
        // consumes it — this is the column list `DataFusion`'s planner
        // determined for the query, and stays populated even when execution
        // produces zero batches (see `QueryOutput::schema`'s own doc).
        let schema: SchemaRef = dataframe.schema().inner().clone();
        let batches = tokio::time::timeout(self.limits.timeout, dataframe.collect())
            .await
            .map_err(|_elapsed| QueryEngineError::Timeout(self.limits.timeout))??;

        let (batches, truncated) = cap_rows(batches, self.limits.row_cap);
        Ok(QueryOutput {
            batches,
            truncated,
            schema,
        })
    }
}

/// Apply the row cap across `batches`, in order, keeping whole batches until
/// `cap` would be exceeded and slicing the batch that crosses it.
/// `truncated` is `true` iff the real total row count exceeded `cap`.
fn cap_rows(batches: Vec<RecordBatch>, cap: usize) -> (Vec<RecordBatch>, bool) {
    let mut kept = Vec::with_capacity(batches.len());
    let mut remaining = cap;
    let mut truncated = false;

    for batch in batches {
        if remaining == 0 {
            truncated |= batch.num_rows() > 0;
            continue;
        }
        if batch.num_rows() <= remaining {
            remaining -= batch.num_rows();
            kept.push(batch);
        } else {
            kept.push(batch.slice(0, remaining));
            remaining = 0;
            truncated = true;
        }
    }

    (kept, truncated)
}

#[cfg(test)]
mod tests;