polyc-query 2026.8.3

//! SQL view bodies `crate::engine::QueryEngine::build` compiles into
//! `CREATE VIEW` statements — the committed-turns view plus every
//! raw/redacted view pair a scope-dependent typed table needs.
//!
//! [`COMMITTED_TURNS_VIEW_SQL`]'s query shape was originally verbatim from
//! docs/reference/datafusion-data-layer.md's "The journal `TableProvider`"
//! section (a window-function reimplementation of the control plane's own
//! committed-turns filter, proven byte-for-byte against a deliberately
//! orphaned turn by the spike — a raw count of 3 rows for the orphaned
//! `turn_id`, 0 through this view). QRY-5 replaced that window-function body
//! with the correlated-`EXISTS` form below (see [`COMMITTED_TURNS_VIEW_SQL`]'s
//! own doc for why) — the 3-vs-0 proof itself is unchanged and still holds
//! (`engine::tests::real_journal_round_trip_reproduces_the_spikes_3_vs_0`).
//! One further adaptation [`QueryEngine::build`](crate::engine::QueryEngine::build)
//! needs that the spike's own ad hoc query didn't: the design doc's text
//! reads `FROM events`, naming the raw wide table `events` directly. This
//! crate instead reserves the name `events` for the view itself (the query
//! layer's **default surface** — see the crate's public API), so the `FROM`
//! clause here targets [`crate::engine::EVENTS_RAW_TABLE`] (`"events_raw"`) —
//! the raw registration `QueryEngine::build` builds this view from and then,
//! for every non-[`Fleet`](crate::session::QueryScope::Fleet) scope,
//! immediately deregisters. A caller opts into raw replay through that
//! explicit `events_raw` table, the single opt-in mechanism, to see
//! uncommitted/orphaned debris at all (docs/reference/datafusion-data-layer.md,
//! "The journal `TableProvider`"). `events_raw` stays maintainer-only;
//! persona-scoped sessions never see it (docs/reference/datafusion-data-layer.md,
//! "Access control").
//!
//! # Committed-turn filter invariant (QRY-1)
//!
//! [`COMMITTED_TURNS_VIEW_SQL`] hides orphaned turn rows from `events` at every
//! scope, including Fleet. A maintainer must opt into `events_raw` to read that
//! debris. Before this invariant, each typed public view selected its own raw
//! `MemTable` without a commit filter. A crash could therefore expose side
//! effects whose `turn_complete` marker never landed. The `events` view hid the
//! same turn. A conversation-scoped persona could read this inconsistency from
//! typed tables that were not Fleet-only.
//!
//! Each turn-coupled view adds a correlated `WHERE EXISTS` subquery. It matches
//! `events` on both `partition` and `turn_id`. The engine registers `events`
//! before every scope-dependent view. Existing identity, signer-key, and
//! `internal_only` redaction still applies. The commit check adds a filter; it
//! does not replace redaction. It also reads the existing `*_raw` table without
//! changing its contents. Raw tables remain Fleet-only and are deregistered at
//! every other scope. Fleet sessions can still inspect uncommitted rows through
//! those explicit raw tables.
//!
//! Handoffs are the exception. Production writes each signed handoff command
//! independently under a synthetic identifier. No turn boundary shares that
//! identifier. The public handoff views therefore keep every decodable record
//! instead of applying the turn-commit filter.
//!
//! # QRY-5: correlated on `(partition, turn_id)`, not `turn_id` alone
//!
//! Before QRY-5, every one of these filters read `WHERE turn_id IN (SELECT
//! turn_id FROM events)` — a plain semijoin keyed on `turn_id` ALONE, with no
//! `partition` correlation. Since `turn_id` is a client-minted UUID the
//! journal itself never enforces as globally unique, that plain semijoin let
//! ANY conversation's committed `turn_id` vouch for every OTHER conversation
//! sharing that same `turn_id` value — a typed row from conversation B's own
//! uncommitted/orphaned turn would pass the filter as long as SOME OTHER
//! conversation A happened to have a committed turn under the identical
//! `turn_id`. The `EXISTS` form above closes that: it correlates on BOTH
//! `t.partition = e.partition` AND `t.turn_id = e.turn_id`, so a typed row is
//! kept only when ITS OWN conversation's `events` view already proves that
//! row's OWN `(partition, turn_id)` pair committed — the identical
//! `(partition, turn_id)` correlation [`COMMITTED_TURNS_VIEW_SQL`]'s own
//! rewritten body now checks directly (see that constant's doc). `EXISTS`
//! rather than the row-value form `WHERE (partition, turn_id) IN (SELECT
//! partition, turn_id FROM events)` — both express the identical semantics,
//! but the row-value `IN` form is untested against this crate's pinned
//! `datafusion = "=54.0.0"`, while `EXISTS` is a long-established, routinely
//! decorrelated `DataFusion` construct.
//!
//! # NULL `turn_id` rows are dropped, intentionally
//!
//! Every `EXISTS` correlation above compares `t.turn_id = e.turn_id` — plain
//! SQL equality, which is `UNKNOWN` (never `TRUE`) whenever `t.turn_id IS
//! NULL`, the same three-valued-logic rule that made the ORIGINAL `turn_id IN
//! (SELECT turn_id FROM events)` semijoin drop a `NULL` `turn_id` too (`NULL
//! IN (...)` is never `TRUE` either). A typed row whose `turn_id` is `NULL` —
//! a bare, un-tagged event's kind carries no `turn_id` at all, see
//! `crate::decode`'s `decode_typed_kind_events`/`kinds::parse` — is therefore
//! DROPPED by every filtered view above, exactly like an orphaned/uncommitted
//! turn's row is: there is no `(partition, turn_id)` pair to prove committed
//! when `turn_id` itself is absent, so the fail-closed default is "hide it",
//! not "show it unconditionally". No production emitter tags a typed-table
//! event with a bare, un-tagged kind today — every real `usage`/`payments`/
//! `messages`/... emission carries a `:{turn_uuid}` suffix — so this path is
//! unexercised in production, but it is real `DataFusion` behavior this crate
//! relies on rather than an untested assumption:
//! `engine::tests::null_turn_id_typed_row_is_dropped_by_the_committed_turn_filter`
//! pins it directly (a bare-kind `usage` row is visible through `usage_raw`,
//! absent from `usage`).
//!
//! `turn_failed` carries this filter too, like every table named above — a
//! prior review pass exempted it on the theory that a failed turn's
//! `turn_id` never carries a `turn_complete` marker by construction. That
//! theory is false: the control plane's sole `turn_failed` emission
//! (`crates/control-plane/src/grpc/mod.rs`) pushes the optional
//! `turn_failed` event and then pushes `turn_complete` UNCONDITIONALLY, in
//! the SAME atomic batch, so a normally-failed turn carries BOTH markers and
//! the semijoin keeps its row; only an orphaned/uncommitted `turn_failed` row
//! is hidden, exactly like every other typed table. [`crate::engine::SUMMARY_TABLE`]
//! is the ONE typed table deliberately NOT filtered this way: its own
//! `turn_id` column is a synthetic tag minted for the summary event itself
//! (see `crate::decode::summary`'s module docs), never a real turn's id, so a
//! join against `events.turn_id` would zero the table unconditionally.
//! `summary` is instead scoped Fleet-only (see
//! [`crate::engine::SUMMARY_TABLE`]'s doc).

/// The committed-turns view: every row belonging to a `(partition, turn_id)`
/// pair that carries both a `turn_start` and a `turn_complete` marker WITHIN
/// THAT SAME PARTITION.
///
/// Selects from `events_raw`, not `events` — see the module docs for why.
/// The explicit `partition, position, kind, turn_id` projection also means
/// `events_raw`'s `payload` and `payload_json` columns
/// (`crate::provider::EventsTableProvider::schema`) never reach this view —
/// the redaction invariant `crate::engine`'s module docs describe.
///
/// # QRY-5: correlated on `(partition, turn_id)`, not `turn_id` alone
///
/// `turn_id` is a UUID minted client-side per turn, not a value the journal
/// itself guarantees globally unique — two DIFFERENT conversations
/// (`partition`s) can in principle carry the identical `turn_id` (a UUID
/// collision, a replayed/forked fixture, or a future caller that mints
/// `turn_id`s from a narrower space than a full UUID). This view's ORIGINAL
/// body (see the module docs) computed `start_count`/`complete_count` via
/// `SUM(...) OVER (PARTITION BY turn_id)` — windowed by `turn_id` ALONE, with
/// no `partition` correlation, which would let one conversation's
/// `turn_complete` marker satisfy another conversation's SAME-`turn_id`
/// `turn_start`: a committed turn in conversation A would make an orphaned,
/// same-`turn_id` turn in conversation B read as committed too, even though
/// B's own turn never completed. The two `EXISTS` subqueries below close
/// that: each correlates on BOTH `partition` AND `turn_id`, so a row is kept
/// only when THAT SAME conversation carries its own `turn_start`/
/// `turn_complete` marker for that `turn_id` — never a different
/// conversation's.
///
/// This `EXISTS` form also replaces an earlier `(partition, turn_id)`-
/// windowed rewrite of the ORIGINAL body (`SUM(...) OVER (PARTITION BY
/// partition, turn_id)`) that was correct but triggered a `DataFusion` 54
/// planner defect: `SELECT DISTINCT ... FROM events LIMIT n` (the shape
/// [`crate::engine::QueryEngine::execute`]'s row-cap `Limit` push produces
/// for every plain query) failed with an internal
/// `"Ordering direction required for DISTINCT with limit"` error once this
/// view's window operator partitioned by two columns instead of one — a
/// bug in `DataFusion`'s `LimitedDistinctAggregation`/`TopK` optimizer path,
/// not a semantic issue with the two-column partition itself. The
/// window-function shape is dropped entirely here rather than special-cased
/// around; `EXISTS` needs no window operator at all, so it cannot trip that
/// optimizer path.
pub(crate) const COMMITTED_TURNS_VIEW_SQL: &str = r"
SELECT partition, position, kind, turn_id
FROM events_raw t
WHERE turn_id IS NOT NULL
  AND EXISTS (
    SELECT 1 FROM events_raw s
    WHERE s.partition = t.partition AND s.turn_id = t.turn_id AND s.kind_base = 'turn_start'
  )
  AND EXISTS (
    SELECT 1 FROM events_raw c
    WHERE c.partition = t.partition AND c.turn_id = t.turn_id AND c.kind_base = 'turn_complete'
  )
";

/// The full `attribution` view, built only for
/// [`Fleet`](crate::session::QueryScope::Fleet).
///
/// Every column [`crate::decode::attribution::schema`] decodes, including
/// the four flattened `identity_*` columns, selected from
/// [`crate::engine::ATTRIBUTION_RAW_TABLE`], filtered to committed turns —
/// see the module docs' "Committed-turn filter invariant" section for why
/// every typed table's public view, this one included, joins against
/// `events` rather than selecting `attribution_raw` verbatim.
///
/// See [`ATTRIBUTION_REDACTED_VIEW_SQL`] for every other scope, and
/// `crate::engine`'s module docs ("Identity redaction invariant") for why
/// the two must never share a column list.
pub(crate) const ATTRIBUTION_VIEW_SQL: &str = r"
SELECT partition, position, turn_id, persona_id, role,
    identity_provider, identity_scope, identity_external_id, identity_display_name
FROM attribution_raw t
WHERE EXISTS (
    SELECT 1 FROM events e WHERE e.partition = t.partition AND e.turn_id = t.turn_id
)
";

/// The redacted `attribution` view built for every non-Fleet scope.
///
/// The fact model's uniform keys (`partition, position, turn_id`) plus the
/// RESOLVED, opaque `persona_id` and `role` a conversation-scoped session
/// already legitimately sees for its own conversation — with none of
/// `ExternalIdentity`'s four flattened `identity_*` columns, i.e. no raw
/// external identity for any participant (docs/reference/datafusion-data-layer.md,
/// "Three boundary conditions": "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"), AND filtered to committed
/// turns — see the module docs' "Committed-turn filter invariant" section.
pub(crate) const ATTRIBUTION_REDACTED_VIEW_SQL: &str = r"
SELECT partition, position, turn_id, persona_id, role
FROM attribution_raw t
WHERE EXISTS (
    SELECT 1 FROM events e WHERE e.partition = t.partition AND e.turn_id = t.turn_id
)
";

/// The full `payments` view, built only for
/// [`Fleet`](crate::session::QueryScope::Fleet).
///
/// Every column [`crate::decode::payments::schema`] decodes, including the
/// raw `signer_public_key`, selected from
/// [`crate::engine::PAYMENTS_RAW_TABLE`], filtered to committed turns — see
/// the module docs' "Committed-turn filter invariant" section.
///
/// See [`PAYMENTS_REDACTED_VIEW_SQL`] for every other scope, and
/// `crate::decode::payments`'s module docs ("Redaction") for why the two
/// must never share a column list.
pub(crate) const PAYMENTS_VIEW_SQL: &str = r"
SELECT partition, position, turn_id, direction, reference, amount, currency,
    recipient, method, timestamp, version, tool_call_id, approval_pos,
    approved_args_hash, subject, signer_public_key, payer_kind, paying_account
FROM payments_raw t
WHERE EXISTS (
    SELECT 1 FROM events e WHERE e.partition = t.partition AND e.turn_id = t.turn_id
)
";

/// The redacted `payments` view built for every non-Fleet scope.
///
/// Every column [`crate::decode::payments::schema`] decodes EXCEPT the raw
/// `signer_public_key` — docs/reference/datafusion-data-layer.md's access-
/// control boundary conditions: "raw signer keys ... stay out of
/// persona-scoped results. This redaction applies to every non-maintainer
/// registration." `subject` (the already-resolved, opaque principal the
/// spend is attributed to) is NOT redacted — see
/// `crate::decode::payments`'s module docs for why it carries the same
/// status `attribution.persona_id` has, not a raw identity. `payer_kind`/
/// `paying_account` are likewise NOT redacted: `paying_account` names the
/// reader's own wallet or the deployment's, the same non-redacted class as
/// `recipient` (a payment fact, not the deployment's own signing key) — see
/// `crate::decode::payments`'s module docs ("Payer attribution") for the
/// full argument. Also filtered to committed turns — see the module docs'
/// "Committed-turn filter invariant" section.
pub(crate) const PAYMENTS_REDACTED_VIEW_SQL: &str = r"
SELECT partition, position, turn_id, direction, reference, amount, currency,
    recipient, method, timestamp, version, tool_call_id, approval_pos,
    approved_args_hash, subject, payer_kind, paying_account
FROM payments_raw t
WHERE EXISTS (
    SELECT 1 FROM events e WHERE e.partition = t.partition AND e.turn_id = t.turn_id
)
";

/// The full `refusals` view (`#2090`, INV-W5), built only for
/// [`Fleet`](crate::session::QueryScope::Fleet).
///
/// Every column [`crate::decode::refusals::schema`] decodes, including the
/// raw `signer_public_key`, selected from
/// [`crate::engine::REFUSALS_RAW_TABLE`] VERBATIM — deliberately no
/// committed-turn filter; see [`crate::engine::REFUSALS_TABLE`]'s own doc for
/// the full reasoning.
///
/// See [`REFUSALS_REDACTED_VIEW_SQL`] for every other scope.
pub(crate) const REFUSALS_VIEW_SQL: &str = r"
SELECT partition, position, turn_id, reason, reason_detail, merchant_host,
    requested_base_units, permitted_base_units, tool_call_id, subject,
    timestamp, signer_public_key
FROM refusals_raw
";

/// The redacted `refusals` view built for every non-Fleet scope.
///
/// Every column [`crate::decode::refusals::schema`] decodes EXCEPT the raw
/// `signer_public_key` — the same redaction [`PAYMENTS_REDACTED_VIEW_SQL`]
/// applies. `subject` is NOT redacted, for the same reason
/// `payments.subject` isn't (already the resolved, opaque principal the
/// fold hands every consumer). No committed-turn filter, deliberately — see
/// [`crate::engine::REFUSALS_TABLE`]'s own doc.
pub(crate) const REFUSALS_REDACTED_VIEW_SQL: &str = r"
SELECT partition, position, turn_id, reason, reason_detail, merchant_host,
    requested_base_units, permitted_base_units, tool_call_id, subject,
    timestamp
FROM refusals_raw
";

/// The full `wallet_link_lifecycle` view (`#2123`), built only for
/// [`Fleet`](crate::session::QueryScope::Fleet).
///
/// Every column [`crate::decode::wallet_link_lifecycle::schema`] decodes,
/// including the raw `signer_public_key`, selected from
/// [`crate::engine::WALLET_LINK_LIFECYCLE_RAW_TABLE`] VERBATIM — deliberately
/// no committed-turn filter (structurally inapplicable, not merely skipped);
/// see [`crate::engine::WALLET_LINK_LIFECYCLE_TABLE`]'s own doc for the full
/// reasoning.
///
/// See [`WALLET_LINK_LIFECYCLE_REDACTED_VIEW_SQL`] for every other scope.
pub(crate) const WALLET_LINK_LIFECYCLE_VIEW_SQL: &str = r"
SELECT partition, position, turn_id, transition, subject, wallet_address,
    currency, chain_id, limit_base_units, limit_human, period_secs,
    expiry_unix, recipients, conversation_id, timestamp, signer_public_key
FROM wallet_link_lifecycle_raw
";

/// The redacted `wallet_link_lifecycle` view built for every non-Fleet scope.
///
/// Every column [`crate::decode::wallet_link_lifecycle::schema`] decodes
/// EXCEPT the raw `signer_public_key` — the same redaction
/// [`REFUSALS_REDACTED_VIEW_SQL`] applies. `subject` is NOT redacted, for the
/// same reason `refusals.subject` isn't. No committed-turn filter — see
/// [`crate::engine::WALLET_LINK_LIFECYCLE_TABLE`]'s own doc.
pub(crate) const WALLET_LINK_LIFECYCLE_REDACTED_VIEW_SQL: &str = r"
SELECT partition, position, turn_id, transition, subject, wallet_address,
    currency, chain_id, limit_base_units, limit_human, period_secs,
    expiry_unix, recipients, conversation_id, timestamp
FROM wallet_link_lifecycle_raw
";

/// The full `messages` view, built for every scope.
///
/// Every row [`crate::decode::message_content::messages_schema`] decodes,
/// `internal_only = true` rows included, selected from
/// [`crate::engine::MESSAGES_RAW_TABLE`], filtered to committed turns — see
/// the module docs' "Committed-turn filter invariant" section.
///
/// See [`MESSAGES_REDACTED_VIEW_SQL`] for every non-Fleet scope, and
/// `crate::decode::message_content`'s module docs ("Redaction") for why an
/// `internal_only` row must never reach a non-Fleet session. Unlike
/// [`ATTRIBUTION_VIEW_SQL`]/[`ATTRIBUTION_REDACTED_VIEW_SQL`], the two
/// `messages` view bodies share the SAME column list — this is a ROW
/// filter, not a column filter (there is no raw-identity-shaped column on
/// this table the way `attribution`'s `identity_*` columns are).
pub(crate) const MESSAGES_VIEW_SQL: &str = r"
SELECT partition, position, turn_id, role, internal_only, text, trust
FROM messages_raw t
WHERE EXISTS (
    SELECT 1 FROM events e WHERE e.partition = t.partition AND e.turn_id = t.turn_id
)
";

/// The redacted `messages` view built for every non-Fleet scope.
///
/// Same column list as [`MESSAGES_VIEW_SQL`], filtered to
/// `internal_only = false` — see that constant's doc and
/// `crate::decode::message_content`'s module docs ("Redaction") for why: an
/// `internal_only` message is, by its own wire doc, never emitted to any
/// client, so a non-Fleet session (persona/conversation-scoped, "the
/// conversation-scoped agent tool included") must never see one, the same
/// boundary condition [`ATTRIBUTION_REDACTED_VIEW_SQL`] enforces for
/// `identity_*`. Also filtered to committed turns, exactly like
/// [`MESSAGES_VIEW_SQL`] — see the module docs' "Committed-turn filter
/// invariant" section.
pub(crate) const MESSAGES_REDACTED_VIEW_SQL: &str = r"
SELECT partition, position, turn_id, role, internal_only, text, trust
FROM messages_raw t
WHERE internal_only = false AND EXISTS (
    SELECT 1 FROM events e WHERE e.partition = t.partition AND e.turn_id = t.turn_id
)
";

/// The full `tool_calls` view, built for every scope.
///
/// Every row [`crate::decode::message_content::tool_calls_schema`] decodes,
/// `internal_only = true` rows included, selected from
/// [`crate::engine::TOOL_CALLS_RAW_TABLE`], filtered to committed turns —
/// see the module docs' "Committed-turn filter invariant" section.
///
/// See [`TOOL_CALLS_REDACTED_VIEW_SQL`] for every non-Fleet scope — the same
/// ROW-filter (not column-filter) shape [`MESSAGES_VIEW_SQL`] uses, for the
/// identical reason.
pub(crate) const TOOL_CALLS_VIEW_SQL: &str = r"
SELECT partition, position, turn_id, tool_call_id, block_type, name,
    arguments, result, first_party, internal_only, trust
FROM tool_calls_raw t
WHERE EXISTS (
    SELECT 1 FROM events e WHERE e.partition = t.partition AND e.turn_id = t.turn_id
)
";

/// The redacted `tool_calls` view built for every non-Fleet scope.
///
/// Same column list as [`TOOL_CALLS_VIEW_SQL`], filtered to
/// `internal_only = false` — see [`MESSAGES_REDACTED_VIEW_SQL`]'s doc for
/// why an `internal_only` block must never reach a non-Fleet session; a
/// tool call/result folded from an `internal_only` message inherits the
/// same "never emitted to clients" contract its owning message carries.
/// Also filtered to committed turns, exactly like [`TOOL_CALLS_VIEW_SQL`] —
/// see the module docs' "Committed-turn filter invariant" section.
pub(crate) const TOOL_CALLS_REDACTED_VIEW_SQL: &str = r"
SELECT partition, position, turn_id, tool_call_id, block_type, name,
    arguments, result, first_party, internal_only, trust
FROM tool_calls_raw t
WHERE internal_only = false AND EXISTS (
    SELECT 1 FROM events e WHERE e.partition = t.partition AND e.turn_id = t.turn_id
)
";

/// The full `approvals` view, built only for
/// [`Fleet`](crate::session::QueryScope::Fleet).
///
/// Every column [`crate::decode::approvals::schema`] decodes, including the
/// nine Fleet-only columns, selected from
/// [`crate::engine::APPROVALS_RAW_TABLE`], filtered to committed turns —
/// see the module docs' "Committed-turn filter invariant" section.
///
/// See [`APPROVALS_REDACTED_VIEW_SQL`] for every other scope, and
/// `crate::decode::approvals`'s module docs ("Column selection and
/// redaction") for why the two must never share a column list.
pub(crate) const APPROVALS_VIEW_SQL: &str = r"
SELECT partition, position, turn_id, phase, request_id, tool_name, args_json,
    request_reason, request_sandbox_mode, approved, response_reason,
    signature_status, signer_public_key, modified_args_json,
    approved_for_session, caller, approver, response_sandbox_mode,
    injected_context, routine_grant, tool_descriptor_hash, grant_scope
FROM approvals_raw t
WHERE EXISTS (
    SELECT 1 FROM events e WHERE e.partition = t.partition AND e.turn_id = t.turn_id
)
";

/// The redacted `approvals` view built for every non-Fleet scope.
///
/// Exactly the participant-visible column set `forensics::collect_approvals`
/// already shows a participant — `tool_name`/`args_json` on the request,
/// `approved`/`response_reason`/`signature_status` on the response, plus the
/// uniform keys and the `phase`/`request_id` discriminators — with none of
/// the nine Fleet-only columns (docs/reference/datafusion-data-layer.md's
/// access-control boundary conditions: "raw signer keys ... stay out of
/// persona-scoped results"; `crate::decode::approvals`'s module docs explain
/// why the remaining response fields also exceed that participant-visible
/// set). The three routine-grant marker columns (`routine_grant`,
/// `tool_descriptor_hash`, `grant_scope`) ARE persona-visible:
/// none carries an external identity or signer key, and an owner reading
/// their own routine's grant ledger needs them to tell a grant from a
/// revocation — see `crate::decode::approvals`'s "Column selection and
/// redaction" section. Also filtered to committed turns, exactly like
/// [`APPROVALS_VIEW_SQL`] — see the module docs' "Committed-turn filter
/// invariant" section.
pub(crate) const APPROVALS_REDACTED_VIEW_SQL: &str = r"
SELECT partition, position, turn_id, phase, request_id, tool_name, args_json,
    approved, response_reason, signature_status, routine_grant,
    tool_descriptor_hash, grant_scope
FROM approvals_raw t
WHERE EXISTS (
    SELECT 1 FROM events e WHERE e.partition = t.partition AND e.turn_id = t.turn_id
)
";

/// The full `handoffs` view, built only for
/// [`Fleet`](crate::session::QueryScope::Fleet).
///
/// This selects every decoded column from
/// [`crate::engine::HANDOFFS_RAW_TABLE`], including raw `signed_by`.
/// Production commits these signed commands independently, so this view does
/// not apply the turn-commit filter.
///
/// See [`HANDOFFS_REDACTED_VIEW_SQL`] for every other scope, and
/// [`crate::decode::handoffs`]'s module docs ("Redaction") for why the two
/// must never share a column list.
pub(crate) const HANDOFFS_VIEW_SQL: &str = r"
SELECT partition, position, turn_id, phase, child_conversation_id,
    child_agent_id, carried_count, reason, parent_agent_id,
    denial_reason, allowed, signed_by, signature_status
FROM handoffs_raw t
";

/// The redacted `handoffs` view built for every non-Fleet scope.
///
/// This exposes every decoded column except raw `signed_by`. It applies the
/// same signer-key boundary as [`PAYMENTS_REDACTED_VIEW_SQL`]. Every other
/// column, including `signature_status`, remains visible. Like
/// [`HANDOFFS_VIEW_SQL`], it keeps independently committed handoff commands.
pub(crate) const HANDOFFS_REDACTED_VIEW_SQL: &str = r"
SELECT partition, position, turn_id, phase, child_conversation_id,
    child_agent_id, carried_count, reason, parent_agent_id,
    denial_reason, allowed, signature_status
FROM handoffs_raw t
";

/// The full `grant_replays` view, built only for
/// [`Fleet`](crate::session::QueryScope::Fleet).
///
/// Every column [`crate::decode::grant_replays::schema`] decodes, including
/// the raw `signer_public_key`, selected from
/// [`crate::engine::GRANT_REPLAYS_RAW_TABLE`], filtered to committed turns —
/// see the module docs' "Committed-turn filter invariant" section.
///
/// See [`GRANT_REPLAYS_REDACTED_VIEW_SQL`] for every other scope, and
/// [`crate::decode::grant_replays`]'s module docs ("Column selection and
/// redaction") for why the two must never share a column list.
pub(crate) const GRANT_REPLAYS_VIEW_SQL: &str = r"
SELECT partition, position, turn_id, tool, grant_ref, covered_capabilities,
    coverage_hash, signature_status, signer_public_key
FROM grant_replays_raw t
WHERE EXISTS (
    SELECT 1 FROM events e WHERE e.partition = t.partition AND e.turn_id = t.turn_id
)
";

/// The redacted `grant_replays` view built for every non-Fleet scope.
///
/// Exactly the participant-visible column set
/// `forensics::collect_grant_replays` already shows a participant — every
/// column except the raw `signer_public_key` (docs/reference/datafusion-data-layer.md's
/// access-control boundary conditions: "raw signer keys ... stay out of
/// persona-scoped results"). Also filtered to committed turns, exactly like
/// [`GRANT_REPLAYS_VIEW_SQL`] — see the module docs' "Committed-turn filter
/// invariant" section.
pub(crate) const GRANT_REPLAYS_REDACTED_VIEW_SQL: &str = r"
SELECT partition, position, turn_id, tool, grant_ref, covered_capabilities,
    coverage_hash, signature_status
FROM grant_replays_raw t
WHERE EXISTS (
    SELECT 1 FROM events e WHERE e.partition = t.partition AND e.turn_id = t.turn_id
)
";

/// The `usage` view, built identically for every scope (no identity/signer
/// redaction applies to `usage` — see [`crate::engine::USAGE_TABLE`]'s doc).
///
/// Every column [`crate::decode::usage::schema`] decodes, selected from
/// [`crate::engine::USAGE_RAW_TABLE`], filtered to committed turns — see the
/// module docs' "Committed-turn filter invariant" section. Unlike
/// `attribution`/`payments`/..., there is no full/redacted pair: `usage` has
/// no Fleet-only column, so one view body covers every scope, the same
/// single-body shape [`MODEL_CALL_VIEW_SQL`] uses.
pub(crate) const USAGE_VIEW_SQL: &str = r"
SELECT partition, position, turn_id, input_tokens, output_tokens
FROM usage_raw t
WHERE EXISTS (
    SELECT 1 FROM events e WHERE e.partition = t.partition AND e.turn_id = t.turn_id
)
";

/// The `model_call` view, built identically for every scope — see
/// [`USAGE_VIEW_SQL`]'s doc for why one body covers every scope here too.
///
/// Every column [`crate::decode::model_call::schema`] decodes, selected from
/// [`crate::engine::MODEL_CALL_RAW_TABLE`], filtered to committed turns —
/// see the module docs' "Committed-turn filter invariant" section.
pub(crate) const MODEL_CALL_VIEW_SQL: &str = r"
SELECT partition, position, turn_id, provider, model, captured_clock_unix_ms
FROM model_call_raw t
WHERE EXISTS (
    SELECT 1 FROM events e WHERE e.partition = t.partition AND e.turn_id = t.turn_id
)
";

/// The `turn_failed` view, built identically for every scope — see
/// [`USAGE_VIEW_SQL`]'s doc for why one body covers every scope here too (no
/// identity/signer-key column on this table either).
///
/// Every column [`crate::decode::turn_failed::schema`] decodes, selected from
/// [`crate::engine::TURN_FAILED_RAW_TABLE`], filtered to committed turns —
/// see the module docs' "Committed-turn filter invariant" section. A
/// normally-failed turn carries both `turn_failed` and `turn_complete` in the
/// same atomic batch (`crates/control-plane/src/grpc/mod.rs`), so this filter
/// hides only an orphaned/uncommitted `turn_failed` row, not every row —
/// see [`crate::engine::TURN_FAILED_TABLE`]'s doc.
pub(crate) const TURN_FAILED_VIEW_SQL: &str = r"
SELECT partition, position, turn_id, failure_kind, message
FROM turn_failed_raw t
WHERE EXISTS (
    SELECT 1 FROM events e WHERE e.partition = t.partition AND e.turn_id = t.turn_id
)
";

/// The `summary` view, built ONLY for
/// [`Fleet`](crate::session::QueryScope::Fleet) — see
/// [`crate::engine::SUMMARY_TABLE`]'s doc for why `summary` is scoped
/// Fleet-only rather than committed-turn-filtered like every other typed
/// table (its own `turn_id` column is a synthetic tag, not a real turn's
/// id, so the committed-turn semijoin every other view in this module uses
/// would zero this table unconditionally rather than hiding only
/// uncommitted rows — see the module docs' "Committed-turn filter
/// invariant" section).
///
/// Every column [`crate::decode::summary::schema`] decodes, selected from
/// [`crate::engine::SUMMARY_RAW_TABLE`] verbatim — no row filter, since
/// Fleet-only scoping is the whole redaction mechanism here, not a
/// second, row-level one.
pub(crate) const SUMMARY_VIEW_SQL: &str = r"
SELECT partition, position, turn_id, text, covers_through_position
FROM summary_raw
";

/// The `fires` view's [`Fleet`](crate::session::QueryScope::Fleet) body —
/// every column [`crate::decode::fires::schema`] decodes EXCEPT the internal
/// `routine_uid` join key (INV-OAF18's uid-churn fix is a persona-scope-only
/// concern; a Fleet session already sees every fire unfiltered, so it has no
/// need to filter or join on this column), selected from
/// [`crate::engine::FIRES_RAW_TABLE`] verbatim, no row filter: a Fleet
/// session sees every fire, the same "no row-level gate, the scope IS the
/// gate" shape [`SUMMARY_VIEW_SQL`] uses.
///
/// A persona-scoped session uses [`FIRES_OWNED_VIEW_SQL`] instead (issue
/// #1882) — see that constant's own doc.
pub(crate) const FIRES_VIEW_SQL: &str = r"
SELECT partition, position, turn_id, routine, occurrence, scheduled_at_ms, fired_at_ms, outcome,
    grant_drift_tools
FROM fires_raw
";

/// The `fires` view's persona-scoped body (issue #1882) — every column
/// [`crate::decode::fires::schema`] decodes EXCEPT the internal `routine_uid`
/// join key (see [`FIRES_VIEW_SQL`]'s own doc for why that column is never
/// user-visible), selected from [`crate::engine::FIRES_RAW_TABLE`],
/// inner-joined against [`crate::engine::ROUTINES_TABLE`] on `routine_uid =
/// routines.uid`.
///
/// This is the row-level owner filter, expressed as a join rather than a
/// `WHERE creator_persona = ...` predicate here: for a persona-scoped
/// session, [`crate::engine::ROUTINES_TABLE`] is ALREADY registered with
/// only that persona's own routines (`crate::authority::ScopedQuery::resolve_routines`
/// filters `reference.routines` before this view is ever created — see
/// `crate::engine::registration`'s registration-ordering doc for why
/// `routines` is registered before this view specifically), so a fire whose
/// `routine_uid` does not appear in that already-filtered table is dropped
/// by the join with no further predicate needed. `fires_raw`'s own source
/// partition (`"routine-scheduler"`) is admitted into a persona-scoped
/// replay narrowly and leniently for exactly this purpose — see
/// `crate::authority`'s module doc, "Admitting the scheduler partition"
/// section.
///
/// # Joins on the CR's `uid`, never its `name` (INV-OAF18)
///
/// A routine's NAME can be deleted and recreated by a different owner:
/// `fires` is append-only history keyed by whatever `routine` name the
/// firing CR had, while `routines` is a live, name-keyed snapshot of
/// CURRENTLY-existing CRs. A name-keyed join (`fires_raw.routine =
/// routines.name`) would let a member who creates a routine under a name a
/// DIFFERENT, earlier member used and deleted see that earlier member's own
/// historical fires — a cross-owner visibility leak on name reuse. Joining
/// on `uid` instead closes this: a CR's `uid` is minted once by the API
/// server and never reused, so two different owners' routines can never
/// collide on it even when they share a name. A fire recorded before this
/// field existed decodes with an empty `routine_uid`
/// (`crate::decode::fires`'s own doc) and simply never matches any owner's
/// `routines.uid` (also never empty for a real CR) — invisible in every
/// persona-scoped `fires` view, same as any other unmatched row, still
/// visible in the unfiltered Fleet `fires` view ([`FIRES_VIEW_SQL`]), which
/// applies no join at all. There is deliberately no name-based fallback for
/// these pre-uid rows: this repository ships no legacy-compatibility path.
///
/// # Fail-closed against an empty `routines.uid` too
///
/// `crate::routine_catalog`'s `to_status_record` derives `routines.uid` from
/// `Routine::uid()` via `.unwrap_or_default()` — an empty string, not an
/// error, for the (today theoretical, kube always assigns one before this
/// crate ever observes a CR) case where a CR has no uid yet. Without the
/// explicit `fires_raw.routine_uid <> ''` guard below, an empty owned-routine
/// uid would equality-match every pre-uid `routine_uid = ''` fire row under
/// plain SQL semantics (`'' = ''` is `TRUE`, unlike the `NULL` case
/// [`crate::views`]'s "NULL `turn_id` rows are dropped" section documents) —
/// resurrecting exactly the leak this join exists to close. The guard makes
/// that combination fail closed instead of silently matching.
pub(crate) const FIRES_OWNED_VIEW_SQL: &str = r"
SELECT fires_raw.partition, fires_raw.position, fires_raw.turn_id, fires_raw.routine,
    fires_raw.occurrence, fires_raw.scheduled_at_ms, fires_raw.fired_at_ms, fires_raw.outcome,
    fires_raw.grant_drift_tools
FROM fires_raw
JOIN routines ON fires_raw.routine_uid = routines.uid
WHERE fires_raw.routine_uid <> ''
";

/// The `turn_dispatch` view, registered identically for every scope — see
/// [`USAGE_VIEW_SQL`]'s doc for why one body covers every scope here too (no
/// identity on this table either).
///
/// Every column [`crate::decode::turn_dispatch::schema`] decodes, selected
/// from [`crate::engine::TURN_DISPATCH_RAW_TABLE`], filtered to committed
/// turns — see the module docs' "Committed-turn filter invariant" section
/// (issue #1593).
pub(crate) const TURN_DISPATCH_VIEW_SQL: &str = r"
SELECT partition, position, turn_id, occurrence, visibility, visibility_source,
    source_turn_id, edge_asserted_visibility
FROM turn_dispatch_raw t
WHERE EXISTS (
    SELECT 1 FROM events e WHERE e.partition = t.partition AND e.turn_id = t.turn_id
)
";

/// The `routine_lifecycle` view, built ONLY for
/// [`Fleet`](crate::session::QueryScope::Fleet) — the same Fleet-only gate
/// [`FIRES_VIEW_SQL`] uses, for the same underlying reason: this table's one
/// source partition (`"routine-scheduler"`) is itself admitted into replay
/// only for a Fleet scope — see [`crate::decode::routine_lifecycle`]'s
/// module docs (issue #1593).
///
/// Every column [`crate::decode::routine_lifecycle::schema`] decodes,
/// selected from [`crate::engine::ROUTINE_LIFECYCLE_RAW_TABLE`] verbatim —
/// no row filter, since Fleet-only scoping is the whole gate here, not a
/// second, row-level one (mirrors [`FIRES_VIEW_SQL`]'s own shape).
pub(crate) const ROUTINE_LIFECYCLE_VIEW_SQL: &str = r"
SELECT partition, position, turn_id, phase, routine, actor_persona, conversation_id, at_ms,
    reason, scope, signer_public_key
FROM routine_lifecycle_raw
";

/// The `routine_setup` view's [`Fleet`](crate::session::QueryScope::Fleet)
/// body — every column
/// [`crate::decode::routine_setup::schema`] decodes, selected from
/// [`crate::engine::ROUTINE_SETUP_RAW_TABLE`] verbatim, no row filter: the
/// same "the scope IS the gate" shape [`FIRES_VIEW_SQL`] uses.
pub(crate) const ROUTINE_SETUP_VIEW_SQL: &str = r"
SELECT partition, position, turn_id, routine_uid
FROM routine_setup_raw
";

/// The `routine_setup` view's persona-scoped body — the same
/// uid-keyed owner-filtering join [`FIRES_OWNED_VIEW_SQL`] applies, for the
/// same reason: for a persona-scoped session the `routines` table is
/// already filtered to that persona's own routines, so the join IS the
/// row-level owner filter, and the `<> ''` guard fails closed against an
/// empty-uid equality match exactly as `FIRES_OWNED_VIEW_SQL`'s doc
/// explains.
pub(crate) const ROUTINE_SETUP_OWNED_VIEW_SQL: &str = r"
SELECT routine_setup_raw.partition, routine_setup_raw.position, routine_setup_raw.turn_id,
    routine_setup_raw.routine_uid
FROM routine_setup_raw
JOIN routines ON routine_setup_raw.routine_uid = routines.uid
WHERE routine_setup_raw.routine_uid <> ''
";

/// The `routine_grants` view — the durable routine-fire
/// tool-grant LEDGER: every `approval_response` carrying the
/// `routine_grant` marker in a routine's fire conversation, joined to the
/// governed routine through `routines.fire_conversation_id` (the caller-
/// computed synthetic fire-conversation id — see
/// [`crate::routine_catalog::RoutineStatusRecord::fire_conversation_id`]).
///
/// ONE body for [`Fleet`](crate::session::QueryScope::Fleet) and a
/// routine-owning persona-scoped session: for a persona scope the
/// `routines` side is already filtered to the caller's own routines
/// (`crate::authority::ScopedQuery::resolve_routines`), so the join IS the
/// owner filter — the `FIRES_OWNED_VIEW_SQL` mechanism. A Fleet session's
/// `routines` holds every routine, so the same join yields every grant.
/// Never registered for a conversation-grant scope.
///
/// **Committed-turn filter: deliberate exception, not applied** — the same
/// class of exception [`crate::engine::REFUSALS_TABLE`]'s doc records. A
/// grant minted outside a paused turn (`append_routine_grant`: the
/// create-time blanket mint, POLY-32's allow-from-denial and revocation)
/// carries a synthetic turn id with NO `turn_dispatched`/`turn_complete`
/// events behind it, so the usual committed-turn semijoin would hide
/// exactly the records this ledger exists to show. Every grant record is
/// durable the instant its own atomic append commits.
///
/// Rows are the raw ledger, newest and oldest alike — a revocation is a
/// later `approved = false` row for the same key. The latest-wins fold is
/// [`ROUTINE_ACTIVE_GRANTS_VIEW_SQL`]'s job.
pub(crate) const ROUTINE_GRANTS_VIEW_SQL: &str = r"
SELECT r.name AS routine, r.uid AS routine_uid, a.partition, a.position, a.turn_id,
    a.tool_name, a.grant_scope, a.tool_descriptor_hash, a.approved,
    a.response_reason, a.caller, a.approver, a.signature_status
FROM approvals_raw a
JOIN routines r ON a.partition = concat('conv-', r.fire_conversation_id)
WHERE a.phase = 'response' AND a.routine_grant = true AND r.fire_conversation_id <> ''
";

/// The `routine_active_grants` view — the latest-wins fold over
/// [`ROUTINE_GRANTS_VIEW_SQL`], the SQL mirror of the control plane's own
/// `collect_routine_tool_grants` fold
/// (`crates/control-plane/src/grpc/approvals.rs`): keys are per-tool
/// (`grant_scope = 'tool'`, keyed by tool name) with every blanket scope
/// collapsed onto one shared key, ordered by journal position, and only a
/// key whose LATEST record approves survives — a latest disapproving
/// record (a revocation) removes its key. Same scope posture and body-
/// sharing as [`ROUTINE_GRANTS_VIEW_SQL`].
///
/// This mirrors the fold's PRE-verification candidate set. The harness
/// verifies each record's signature before the gate honors it, and this
/// view does not filter on `signature_status`. A row here is not proof
/// the gate enforces that grant.
pub(crate) const ROUTINE_ACTIVE_GRANTS_VIEW_SQL: &str = r"
SELECT routine, routine_uid, tool_name, grant_scope, tool_descriptor_hash, position
FROM (
    SELECT routine, routine_uid, tool_name, grant_scope, tool_descriptor_hash, approved,
        position,
        ROW_NUMBER() OVER (
            PARTITION BY routine_uid,
                CASE WHEN grant_scope = 'tool' THEN concat('tool:', tool_name)
                     ELSE 'blanket' END
            ORDER BY position DESC
        ) AS rn
    FROM routine_grants
)
WHERE rn = 1 AND approved = true
";

/// The `routine_overview` view — the per-routine aggregate, a
/// COMPOSITION over the typed tables this crate already registers (never a
/// second read path): one row per `routines` row, left-joined to
///
/// - fire stats over `fires_raw` (count + last fire instant), uid-keyed
///   with the `<> ''` fail-closed guard `FIRES_OWNED_VIEW_SQL` documents;
/// - the latest blanket record in `routine_active_grants`, from which the
///   grant MODE is DERIVED (`blanket_all` → `approve_all_dangerous`, any
///   other surviving blanket → `auto`, none → `individual`) — mode is not
///   its own event, it IS which blanket grant currently stands;
/// - the per-tool active-grant count in `routine_active_grants`;
/// - unattended denial facts (ordinary, non-grant-marked disapproving
///   `approval_response` records in the fire conversation — the signed
///   request/denied-response pairs a fire dispatch appends);
/// - unresolved `approval_request` records in the fire conversation (a
///   setup rehearsal's still-parked approvals), by anti-join on
///   `(partition, turn_id, request_id)`;
/// - `routine_setup_raw` presence, as the `setup_completed` flag.
///
/// The routine's fire OUTPUT is deliberately not aggregated here — output
/// is conversation content, reachable through the `fire_conversation_id`
/// column this view carries (join `turn_dispatch`/`messages` on that
/// conversation, exactly as the fires surface already does). Same
/// committed-turn exception and scope posture as
/// [`ROUTINE_GRANTS_VIEW_SQL`].
///
/// `denial_count` IS the refusal surface for a routine's fires: it counts
/// denied signed responses in the fire conversation. The separate
/// `refusals` table decodes a different event family (payment refusals) and
/// is deliberately not joined here.
///
/// # POLY-160 additions: the routine's own spec fields, latest-fire outcome,
/// and `stopped_tool`
///
/// `prompt`/`schedule_json`/`next_fires_json`/`suspended`/`paused_by`/
/// `paused_at_ms`/`pause_reason` are carried straight from `routines` (`r.*`)
/// — no new join, since [`crate::decode::routines::schema`] already decodes
/// every one of them from the CR's own spec/status. `last_fire_at_ms`/
/// `last_fire_outcome` replace the plain `MAX(fired_at_ms)` fold with a
/// `ROW_NUMBER()`-windowed pick of the single latest `fires_raw` row per
/// `routine_uid`, so `last_fire_outcome` names THAT row's own `outcome`
/// rather than an aggregate. Ordered `fired_at_ms DESC, position DESC`
/// (review finding) — a scheduler catch-up burst can plausibly record two
/// fires with the identical `fired_at_ms`, and `position` (the journal's own
/// monotonic append order) is the one column that always breaks that tie
/// deterministically, unlike `fired_at_ms` alone.
///
/// `stopped_tool` is deliberately the routine's LATEST unattended-fire
/// denial, not the denial scoped to that same latest fire's own turn: an
/// exact per-fire binding would need a THIRD correlation (fire → turn window
/// → the denial whose `turn_id` falls in that window), and this view already
/// derives `denial_count` from the identical unscoped "every denial in the
/// fire conversation" shape — binding `stopped_tool` any tighter than
/// `denial_count` already is would be inconsistent, not more correct, for no
/// caller-visible benefit today (a routine's fire conversation carries one
/// denial history, not per-fire-scoped ones, until a caller needs otherwise).
pub(crate) const ROUTINE_OVERVIEW_VIEW_SQL: &str = r"
SELECT r.name, r.uid, r.creator_persona, r.fire_conversation_id,
    r.prompt, r.schedule_json, r.next_fires_json, r.suspended, r.paused_by,
    r.paused_at_ms, r.pause_reason,
    CASE WHEN b.grant_scope = 'blanket_all' THEN 'approve_all_dangerous'
         WHEN b.grant_scope IS NOT NULL THEN 'auto'
         ELSE 'individual' END AS mode,
    s.routine_uid IS NOT NULL AS setup_completed,
    COALESCE(f.fire_count, 0) AS fire_count,
    lf.fired_at_ms AS last_fire_at_ms,
    lf.outcome AS last_fire_outcome,
    COALESCE(g.active_tool_grants, 0) AS active_tool_grants,
    COALESCE(d.denial_count, 0) AS denial_count,
    COALESCE(p.pending_setup_approvals, 0) AS pending_setup_approvals,
    ld.tool_name AS stopped_tool
FROM routines r
LEFT JOIN (
    SELECT routine_uid, COUNT(*) AS fire_count
    FROM fires_raw WHERE routine_uid <> '' GROUP BY routine_uid
) f ON f.routine_uid = r.uid
LEFT JOIN (
    SELECT routine_uid, fired_at_ms, outcome FROM (
        SELECT routine_uid, fired_at_ms, outcome,
            ROW_NUMBER() OVER (
                PARTITION BY routine_uid ORDER BY fired_at_ms DESC, position DESC
            ) AS rn
        FROM fires_raw WHERE routine_uid <> ''
    ) WHERE rn = 1
) lf ON lf.routine_uid = r.uid
LEFT JOIN (
    SELECT routine_uid, grant_scope FROM routine_active_grants WHERE grant_scope <> 'tool'
) b ON b.routine_uid = r.uid
LEFT JOIN (
    SELECT routine_uid, COUNT(*) AS active_tool_grants
    FROM routine_active_grants WHERE grant_scope = 'tool' GROUP BY routine_uid
) g ON g.routine_uid = r.uid
LEFT JOIN (
    SELECT r2.uid AS routine_uid, COUNT(*) AS denial_count
    FROM approvals_raw a
    JOIN routines r2 ON a.partition = concat('conv-', r2.fire_conversation_id)
    WHERE a.phase = 'response' AND a.approved = false
        AND COALESCE(a.routine_grant, false) = false
        AND r2.fire_conversation_id <> ''
    GROUP BY r2.uid
) d ON d.routine_uid = r.uid
LEFT JOIN (
    SELECT r3.uid AS routine_uid, COUNT(*) AS pending_setup_approvals
    FROM approvals_raw req
    JOIN routines r3 ON req.partition = concat('conv-', r3.fire_conversation_id)
    WHERE req.phase = 'request' AND r3.fire_conversation_id <> '' AND NOT EXISTS (
        SELECT 1 FROM approvals_raw resp
        WHERE resp.phase = 'response' AND resp.partition = req.partition
            AND resp.turn_id = req.turn_id AND resp.request_id = req.request_id
    )
    GROUP BY r3.uid
) p ON p.routine_uid = r.uid
LEFT JOIN (
    SELECT DISTINCT routine_uid FROM routine_setup_raw WHERE routine_uid <> ''
) s ON s.routine_uid = r.uid
LEFT JOIN (
    SELECT routine_uid, tool_name FROM (
        SELECT r4.uid AS routine_uid, a4.tool_name, a4.position,
            ROW_NUMBER() OVER (PARTITION BY r4.uid ORDER BY a4.position DESC) AS rn
        FROM approvals_raw a4
        JOIN routines r4 ON a4.partition = concat('conv-', r4.fire_conversation_id)
        WHERE a4.phase = 'response' AND a4.approved = false
            AND COALESCE(a4.routine_grant, false) = false
            AND r4.fire_conversation_id <> ''
    ) WHERE rn = 1
) ld ON ld.routine_uid = r.uid
";