polyc-query 2026.9.0

//! Per-kind protobuf-to-Arrow decode registry.
//!
//! Each of `kinds.rs`'s 64 kinds either gets its own typed table or stays an
//! opaque `payload` column in the wide `events` table until a query workload
//! justifies decoding it.
//! `usage` is the fact model's first typed table — the projection the spike
//! already proved, shipped to pin the typed-table path. `model_call`, `attribution`, and
//! `turn_failed` are the fact model's second, third, and fourth typed tables
//! (#1311/#1178's "keyed domain-fact-table" wave), added the same
//! deliberate way — see [`crate::decode::model_call`],
//! [`crate::decode::attribution`], and [`crate::decode::turn_failed`] for
//! their own column rationale. `payments` is the fifth, and the first
//! FOLD-COUPLED typed table — decoded through the shared
//! `polyc_facts::verified_receipts` fold rather than a per-kind protobuf
//! decode; see [`crate::decode::payments`]'s module docs for why it cannot
//! reuse [`decode_typed_kind_events`]. `message_content` —
//! `kinds::USER_MSG`/`kinds::OUTPUT_MSG`'s shared entry — is the sixth: the
//! fact model's "variant→block" content-block fold
//! ([`crate::decode::message_content`], reusing the same
//! `polyc_facts::fold_message_content` the trace projector and the approval
//! collector already call), whose output `crate::engine::QueryEngine::build`
//! registers as two sibling SQL tables, `messages` and `tool_calls` — see
//! that module's docs for the fold and [`Decode::Typed`]'s own doc for why
//! one `REGISTRY` label safely names a decode that fans out to more than
//! one physical table. `approvals` — `kinds::APPROVAL_REQUEST`/
//! `kinds::APPROVAL_RESPONSE`'s shared entry — is the seventh: the second
//! fold-coupled table. It decodes through
//! `polyc_facts::fold_approval_event`, the same fold
//! `forensics::classify_response_signature`/`parse_request_entry`/
//! `parse_response_entry` and `trace::response_signature_evidence` read
//! through — see [`crate::decode::approvals`]'s module docs for its
//! `phase`-discriminated single-table shape and, unlike `payments`, why an
//! unverified response is kept rather than dropped. `handoffs` uses the
//! `kinds::HANDOFF`/`kinds::HANDOFF_DENIED`'s shared entry is the eighth. It is
//! the fourth fold-coupled table (#1579). It decodes through
//! `polyc_facts::fold_handoff_event`, the same decode-and-verify
//! primitive the conversation-trace projector's own handoff steps
//! (`trace.rs`) read through, so trace's handoff events are fully decoded
//! rather than left opaque. See
//! [`crate::decode::handoffs`]'s module docs for its three-value `phase`
//! discriminator and why an unverified event is kept, like
//! `approvals`. `grant_replays` uses `kinds::GRANT_REPLAY` as its own entry. It
//! is the ninth table and the third fold-coupled table. It decodes through
//! `polyc_facts::fold_grant_replay_event`, the same
//! fold `forensics::parse_grant_replay_entry`/`trace::decode_grant_replay_fields`
//! read through. Unlike `approvals`/`handoffs`, it needs no `phase`/`direction`
//! discriminator at all: `grant_replay` is a single signed-JSON shape, not a
//! multi-kind family — see [`crate::decode::grant_replays`]'s module docs for
//! its column selection and why, like `approvals`, an unverified record is
//! kept rather than dropped. `summary` uses `kinds::SUMMARY` as its own entry.
//! It is the tenth table and a direct decode that is not fold-coupled. It uses
//! [`decode_typed_kind_events`], the same shared loop `usage`/`model_call`/
//! `turn_failed` call through — `usage`/`model_call` route their own
//! `decode` argument through the shared `polyc_facts::fold_usage_event`/
//! `fold_model_call_event` folds instead of a raw `try_decode_event_payload`
//! call (#1579's dedup: those two folds are also read by the control
//! plane's per-turn accounting and replay-determinism record), which is
//! exactly why this helper takes `decode` as a parameter rather than
//! resolving it from a `T: buffa::Message` bound. Decoded flat for every
//! scope, like `usage`/`model_call`/`turn_failed` — but, unlike those three, backs a
//! `summary` `CREATE VIEW` scoped to
//! [`QueryScope::Fleet`](crate::session::QueryScope::Fleet) only, not a
//! committed-turn-filtered view built for every scope: a persona/
//! conversation-scoped session must never see summary content, even though
//! a summary's text condenses only its own conversation's replayed
//! history and carries no external identity or signer key; see
//! [`crate::decode::summary`]'s module docs for its `turn_id` caveat (a
//! synthetic tag, never a real turn id, which is also why the usual
//! committed-turn filter cannot apply to this table) and, more importantly,
//! why `covers_through_position` must never be used for a position-based
//! cutoff query. `fires` — `kinds::ROUTINE_FIRED`'s own entry — is the
//! eleventh: a DIRECT decode (like `handoffs`/`summary`) of the routine
//! scheduler's durable `routine_fired` markers, through
//! [`decode_typed_kind_events`]. Like `summary`, `fires` backs a `CREATE
//! VIEW` scoped Fleet-only — see [`crate::decode::fires`]'s module docs for
//! why (the short version: its one source partition,
//! `"routine-scheduler"`, is itself admitted into replay only for a Fleet
//! scope, so a non-Fleet `fires` would only ever resolve to zero rows
//! anyway). `turn_dispatch` — `kinds::TURN_DISPATCHED`'s own entry — is the
//! twelfth (issue #1593): a DIRECT decode of the dispatch-intent marker's
//! `occurrence` field through [`decode_typed_kind_events`], committed-turn
//! filtered for every scope like `usage`/`model_call`/`turn_failed` — see
//! [`crate::decode::turn_dispatch`]'s module docs for why this table exists
//! (joining a routine's `fires` to the turns they opened, by occurrence).
//! `routine_lifecycle` — `kinds::ROUTINE_CREATED`/`ROUTINE_PAUSED`/
//! `ROUTINE_RESUMED`/`ROUTINE_DELETED`'s shared entry — is the thirteenth
//! (issue #1593): a DIRECT decode of the four signed routine lifecycle audit
//! events through `polyc_crypto::approval`'s own verify functions (not
//! [`decode_typed_kind_events`], since these are signed JSON, not protobuf),
//! Fleet-only like `fires` — see [`crate::decode::routine_lifecycle`]'s
//! module docs for its `phase` discriminator and drop-on-unverified posture.
//! `routines` — the CR-list-sourced reference table, alongside
//! `personas`/`participations` — is NOT in [`REGISTRY`] at all: it decodes
//! [`crate::routine_catalog::RoutineStatusRecord`], caller-supplied
//! reference data external to the event journal, never a journal kind; see
//! [`crate::decode::routines`]'s own module docs. [`REGISTRY`] is the
//! explicit typed-vs-opaque decision for every one of the 64 journal kinds,
//! and this module's `#[cfg(test)]` `registry_is_exhaustive_over_kinds_rs`
//! test is the design's own mandated structural test — a
//! registry-exhaustiveness test iterates every kind constant and fails when
//! a new kind lands without a decoder or an explicit opaque fallback.
//!
//! ## The schema-less residual: `payload_json` (#1313)
//!
//! Six kinds carry no proto schema at all (`approval_deferred`,
//! `tool_input_rewrite`, `tool_context_injection`, `tool_result_redaction`,
//! `taint_excision`, `admin_model_change`), and several more carry freeform
//! JSON payloads even where a schema exists elsewhere in the same table
//! (approval/payment `args_json`-style content). All of these are
//! [`Decode::Opaque`] — no typed table backs them — which left them
//! reachable only as raw `Binary` bytes, unqueryable by SQL. #1313 (part of
//! the #1178 schema-less-kinds epic) resolves this as **"opaque and
//! JSON-queryable now"**: rather than block on a proto schema for every one
//! of these kinds, [`events_batch`] additionally decodes each row's raw
//! payload into a `payload_json` `Utf8` column
//! ([`crate::provider::EventsTableProvider::schema`]) whenever the bytes
//! are valid UTF-8 that itself parses as JSON, `NULL` otherwise.
//!
//! `payload_json` is a uniform column on every `events_raw` row, not a
//! third [`Decode`] variant gated per kind. A kind's `Decode` disposition
//! (`Typed`/`Opaque`) answers "does a dedicated table exist for this
//! kind's proto shape"; whether a *specific row's* bytes happen to parse
//! as JSON is an orthogonal, per-payload property — a `Typed` kind's
//! payload is protobuf and simply decodes to `NULL` here, no different
//! from any other non-JSON payload. Branching `payload_json`'s
//! construction on [`Decode`] would reproduce that same `NULL` outcome
//! through an extra conditional, so [`events_batch`] computes it
//! unconditionally for every row instead — the least-special-cased shape,
//! and the one this module's registry-driven design already favors
//! (one explicit disposition per kind, no per-kind column logic).
//!
//! `payload_json`, and every other JSON-string column this crate exposes
//! (`tool_calls.arguments`/`tool_calls.result`), is queryable with real
//! in-SQL JSON navigation: [`crate::engine::QueryEngine::build`] registers
//! the community/contrib `datafusion-functions-json` crate's scalar UDFs —
//! pinned at the `=0.54.2` line tracking this crate's own `datafusion =
//! "=54.0.0"` pin (`crates/query/Cargo.toml`; that crate declares no
//! `arrow` of its own, so it inherits datafusion 54's arrow rather than
//! risking a second arrow version in the lockfile) — for EVERY scope,
//! Fleet included. `json_get(payload_json, 'field')` and its typed
//! siblings (`json_get_str`, `json_get_int`, `json_get_bool`, ...) pull a
//! field out directly, and the crate's registered operator rewrite gives
//! you `payload_json -> 'field'` (returns a JSON-typed value) and
//! `payload_json ->> 'field'` (returns text) as well. `payload_json` still
//! also works like any other `Utf8` column for `LIKE`, `regexp_like`,
//! `substr`, equality, and every other string function `DataFusion` 54
//! ships — the JSON functions are additive, not a replacement. Querying
//! `NULL` (a `Typed` kind's payload, or any non-JSON/non-UTF8 payload)
//! through any of these functions returns `NULL`, never an error (see
//! [`crate::engine`]'s
//! `events_raw_payload_json_extraction_is_null_for_a_typed_payload` test).
//! [`crate::engine`]'s
//! `events_raw_payload_json_is_queryable_via_json_get_for_a_schemaless_kind`
//! and `tool_calls_arguments_is_queryable_via_json_get` tests pin the
//! extraction contract over `payload_json` and `tool_calls.arguments`
//! respectively.

use std::sync::Arc;

use arrow::array::{ArrayRef, BinaryBuilder, StringBuilder, UInt64Builder};
use arrow::error::ArrowError;
use arrow::record_batch::RecordBatch;
use polyc_eventlog::Event;
use polyc_proto::kinds;

pub(crate) mod approvals;
pub(crate) mod attribution;
pub(crate) mod dashboard;
pub(crate) mod fires;
pub(crate) mod grant_replays;
pub(crate) mod handoffs;
pub(crate) mod message_content;
pub(crate) mod model_call;
pub(crate) mod payments;
pub(crate) mod persona;
pub(crate) mod refusals;
pub(crate) mod routine_lifecycle;
pub(crate) mod routine_setup;
pub(crate) mod routines;
pub(crate) mod summary;
pub(crate) mod turn_dispatch;
pub(crate) mod turn_failed;
pub(crate) mod usage;
pub(crate) mod wallet_link_lifecycle;

/// A kind-base's registered decode disposition.
///
/// Exactly one of these covers every kind-base in `polyc_proto::kinds` —
/// see [`REGISTRY`] and its exhaustiveness test.
// This module's runtime decode path (`events_batch`, below) never
// consults `Decode`/`REGISTRY`/`decode_decision` — they exist solely to
// drive `registry_is_exhaustive_over_kinds_rs`, the structural
// exhaustiveness test the design doc mandates (see this module's own
// doc). Sealing this module to `pub(crate)` unmasked that as `dead_code`
// outside `#[cfg(test)]` builds; `allow` rather than delete, since the
// test itself is real and required.
#[allow(dead_code)]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum Decode {
    /// The kind has its own typed table, decoded straight from the proto
    /// payload rather than left in the wide table's opaque `payload`
    /// column. The `&'static str` names the decode this kind drives —
    /// USUALLY exactly one SQL-visible typed table (`"usage"`,
    /// `"model_call"`, …), but not always: `kinds::USER_MSG`/
    /// `kinds::OUTPUT_MSG` both carry `Typed("message_content")`, naming the
    /// ONE shared content-block fold ([`crate::decode::message_content`])
    /// their payloads decode through, whose output
    /// `crate::engine::QueryEngine::build` registers as TWO sibling SQL
    /// tables (`messages`, `tool_calls`) — see that module's docs for why
    /// one kind fans out to more than one physical table. This is safe
    /// because [`REGISTRY`] never drives table registration at runtime (see
    /// the note above this enum) — it exists purely to pin the
    /// typed-vs-opaque decision per kind for
    /// `registry_is_exhaustive_over_kinds_rs`, so a label naming a shared
    /// decode rather than a single table name changes nothing about how
    /// `crate::engine` actually registers tables.
    Typed(&'static str),
    /// No typed table exists yet — a deliberate, reviewed "not yet" rather
    /// than an omission. Rows of this kind stay in the wide `events` table
    /// with an opaque `payload` column.
    Opaque,
}

/// Every kind-base's decode disposition, covering all 73 constants declared
/// in `crates/proto/src/kinds.rs`. Order mirrors that file's declaration
/// order; order carries no meaning beyond readability.
///
/// The count is measured against `kinds.rs`, never carried forward: it read 64
/// while the file declared 70, so it had already drifted by six before
/// `TURN_AMBIGUOUS` and `OBSERVED_IDENTITY` made it eight.
/// `registry_is_exhaustive_over_kinds_rs` is what actually enforces the
/// correspondence; this number is prose.
///
/// Count the ROWS, with comments stripped. Several entries carry a `kinds::`
/// name in their own explanatory comment, so a bare grep over this block
/// over-counts — which is how the first correction to this number arrived at
/// the right answer for the wrong reason.
///
/// This is the "deliberate, reviewed addition" the design doc requires:
/// every typed table is a deliberate, reviewed addition, not a byproduct of
/// a codegen step — a
/// new kind constant must land here in the same review as `kinds.rs`, or
/// this module's `registry_is_exhaustive_over_kinds_rs` test fails the
/// build.
#[allow(dead_code)] // see the `allow` above `Decode`
pub(crate) const REGISTRY: &[(&str, Decode)] = &[
    (kinds::TURN_START, Decode::Opaque),
    (kinds::TURN_COMPLETE, Decode::Opaque),
    // `#743` under D5: a content-free marker saying a paused turn's narration
    // is withheld. No payload to decode, and no query workload reads it — the
    // redaction it drives runs in the control plane's read path.
    (kinds::TURN_TEXT_WITHHELD, Decode::Opaque),
    // POLY-257: a content-free marker saying the turn used the provider's
    // native search grounding, so outside content entered the conversation.
    // No payload to decode. Its whole content is its presence at a position
    // plus its quarantined trust tag, which the wide `events` table already
    // carries; the taint derivation that reads it runs in the control plane.
    (kinds::GROUNDED_CONTENT, Decode::Opaque),
    (kinds::TURN_FAILED, Decode::Typed("turn_failed")),
    // POLY-210: the turn's outcome is not known. Opaque for the same reason
    // `step_commit` is — no query workload reads it as SQL yet. A person reads
    // it through forensics, which decodes the payload by kind. Promote it to a
    // typed table when a query consumer needs the fields.
    (kinds::TURN_AMBIGUOUS, Decode::Opaque),
    (kinds::TURN_DISPATCHED, Decode::Typed("turn_dispatch")),
    (kinds::USER_MSG, Decode::Typed("message_content")),
    (kinds::OUTPUT_MSG, Decode::Typed("message_content")),
    (kinds::USAGE, Decode::Typed("usage")),
    // #1565 D5: State's durable commit marker for one accepted Execution
    // outcome. The marker holds identities, a fence, and a content digest —
    // no model or tool body, because the bodies are the adjacent `output_msg`
    // records the `message_content` decode above already covers. Opaque
    // because no query workload reads step metadata as SQL yet; the resume
    // path reads it through the control plane's own verified projection
    // (`crate::step_commit::last_committed_step`), never through this crate.
    // Promote it to a typed table when a forensics consumer needs the fields.
    (kinds::STEP_COMMIT, Decode::Opaque),
    (kinds::ADMISSION_REFUSED, Decode::Opaque),
    // #1691: the conversation's tenancy binding and the binding check's
    // refusal record. Opaque for the same reason `admission_refused` is —
    // both are operator forensics reached by kind, and a typed view over a
    // refusal family is an explicit non-goal of ADR 0011.
    (kinds::CONVERSATION_NAMESPACE, Decode::Opaque),
    (kinds::NAMESPACE_REFUSED, Decode::Opaque),
    // The conversation's durable multi-party floor. Opaque: the marker's
    // whole content is its presence plus a timestamp, and the only consumer
    // is the control plane's own memory-scope decision, which reads the
    // journal directly. A typed table would also hand SQL a "who is in this
    // room" shape the payload deliberately refuses to carry.
    (kinds::CONVERSATION_MULTIPARTY_FLOOR, Decode::Opaque),
    (kinds::MODEL_CALL, Decode::Typed("model_call")),
    (kinds::SUMMARY, Decode::Typed("summary")),
    // #1783/INV-C13: the mint-time retention gate's audit trail for a
    // rejected/force-admitted candidate. Structured payloads exist
    // (`SummaryGateRejectedEvent`/`SummaryGateAdmittedEvent`, both already
    // dispatch-table-decoded in `events_decode.rs`), but — like
    // `conversation_summary`/`query_audit`/`erasure_recorded` below — a
    // proto schema alone is not the bar for a typed table here (see this
    // module's docs: "every typed table is a deliberate, reviewed
    // addition... not a byproduct of a codegen step"). The gate's own
    // consecutive-reject count is derived by the control plane replaying
    // the raw eventlog directly (`consecutive_gate_rejects`), never through
    // this crate, so there is no query-workload consumer yet to justify
    // one; Opaque until a forensics/query consumer needs these fields in
    // SQL.
    (kinds::SUMMARY_GATE_REJECTED, Decode::Opaque),
    (kinds::SUMMARY_GATE_ADMITTED, Decode::Opaque),
    (kinds::COMPACTION_CHECKPOINT, Decode::Opaque),
    (kinds::APPROVAL_REQUEST, Decode::Typed("approvals")),
    (kinds::APPROVAL_RESPONSE, Decode::Typed("approvals")),
    (kinds::APPROVAL_DEFERRED, Decode::Opaque),
    // #1660: ask_question's pending-question request/response pair — a
    // sibling of the approval request/response entries above, deliberately
    // NOT folded into the `approvals` typed table (a question is not a
    // danger/permission decision). Opaque for now, same as
    // `APPROVAL_DEFERRED` above: a reviewed, not-yet decision, not an
    // oversight — a `Typed("questions")` fold is a later slice once a
    // forensics consumer needs one.
    (kinds::QUESTION_REQUEST, Decode::Opaque),
    (kinds::QUESTION_RESPONSE, Decode::Opaque),
    (kinds::TOOL_INPUT_REWRITE, Decode::Opaque),
    (kinds::TOOL_CONTEXT_INJECTION, Decode::Opaque),
    (kinds::TOOL_RESULT_REDACTION, Decode::Opaque),
    (kinds::PAYMENT_RECEIPT, Decode::Typed("payments")),
    (kinds::CALLER, Decode::Typed("attribution")),
    (kinds::PARTICIPANT, Decode::Typed("attribution")),
    // POLY-232: an edge-observed identity, bare kind (no turn suffix).
    // Deliberately Opaque, and deliberately NOT folded into `attribution`
    // above even though both source kinds carry a similar
    // identity+persona_id shape: `attribution`'s decode routes through
    // `polyc_facts::attribution_events_with_positions`/`AttributionScope`,
    // the exact fold the lethal-trifecta taint/completeness read path is
    // built on (see `kinds::OBSERVED_IDENTITY`'s own doc). Widening that
    // scope to cover this kind — even read-only, even in this crate, which
    // never itself computes taint — would be the kind of "it's just a
    // decode, what's the harm" step that quietly re-couples the two. No
    // query workload needs this kind's fields as SQL yet; the forensics
    // conversation trace (`crates/control-plane/src/trace.rs`, a different
    // crate) is its one reader today. Opaque until a real query consumer
    // justifies a dedicated typed table, decoded through its own function —
    // never through `attribution`'s.
    (kinds::OBSERVED_IDENTITY, Decode::Opaque),
    (kinds::OUTBOUND_PAYMENT_RECEIPT, Decode::Typed("payments")),
    (kinds::OUTBOUND_PAYMENT_ATTEMPT, Decode::Opaque),
    // `#2090`, INV-W5: durable payment-refusal events, decoded through the
    // shared `polyc_facts::verified_refusals` fold into the `refusals`
    // typed table (`crate::decode::refusals`).
    (kinds::PAYMENT_REFUSAL, Decode::Typed("refusals")),
    // `#2123`: durable wallet-link lifecycle events (link/renew/revoke),
    // decoded through the shared
    // `polyc_facts::verified_wallet_link_lifecycle_events` fold into the
    // `wallet_link_lifecycle` typed table
    // (`crate::decode::wallet_link_lifecycle`).
    (
        kinds::WALLET_LINK_LIFECYCLE,
        Decode::Typed("wallet_link_lifecycle"),
    ),
    (kinds::TAINT_EXCISION, Decode::Opaque),
    (kinds::ENROLLMENT, Decode::Opaque),
    (kinds::ROUTINE_GRANT, Decode::Opaque),
    (kinds::ROUTINE_FIRED, Decode::Typed("fires")),
    // #1656: a fire's classified outcome, folded into the SAME `fires`
    // table by `(routine, occurrence)` — see `crate::decode::fires`'s
    // module doc's "outcome" section. Not a separate table: this constant
    // shares `ROUTINE_FIRED`'s `Decode::Typed("fires")` label rather than
    // minting a second one, matching how `message_content`/`approvals`
    // already share one table label across more than one kind.
    (kinds::ROUTINE_FIRE_OUTCOME, Decode::Typed("fires")),
    // `#1497`/`#1495`/`#1593`: all four routine lifecycle audit kinds now
    // share one typed table (`routine_lifecycle`), added together as the
    // REGISTRY comment they replaced always planned. `#1806` adds
    // `ROUTINE_SCOPE_CHANGED` as a fifth phase of the same table — same
    // partition, same signed-JSON construction, see
    // `crate::decode::routine_lifecycle`'s module doc.
    (kinds::ROUTINE_CREATED, Decode::Typed("routine_lifecycle")),
    (kinds::ROUTINE_PAUSED, Decode::Typed("routine_lifecycle")),
    (kinds::ROUTINE_RESUMED, Decode::Typed("routine_lifecycle")),
    (kinds::ROUTINE_DELETED, Decode::Typed("routine_lifecycle")),
    (
        kinds::ROUTINE_SCOPE_CHANGED,
        Decode::Typed("routine_lifecycle"),
    ),
    // The scheduler's setup-completion marker, decoded into the
    // `routine_setup` typed table so the read surface can answer "has this
    // routine's setup rehearsal completed" — see
    // `crate::decode::routine_setup`'s module doc.
    (
        kinds::ROUTINE_SETUP_COMPLETED,
        Decode::Typed("routine_setup"),
    ),
    (kinds::GRANT_REVOCATION, Decode::Opaque),
    (kinds::GRANT_SUSPENSION, Decode::Opaque),
    (kinds::GRANT_REPLAY, Decode::Typed("grant_replays")),
    (kinds::NUDGE_SENT, Decode::Opaque),
    (kinds::MEMORY_ADDED, Decode::Opaque),
    (kinds::MEMORY_INVALIDATED, Decode::Opaque),
    (kinds::MEMORY_CORROBORATED, Decode::Opaque),
    (kinds::PROFILE_DOC_UPDATED, Decode::Opaque),
    (kinds::CONVERSATION_SUMMARY, Decode::Opaque),
    (kinds::INCOGNITO_SET, Decode::Opaque),
    (kinds::MEMORY_EXTRACTED, Decode::Opaque),
    (kinds::ERASURE_RECORDED, Decode::Opaque),
    (kinds::CONVERSATION_ERASURE_INTENT, Decode::Opaque),
    (kinds::QUERY_AUDIT, Decode::Opaque),
    // The audit tier stays opaque to the fact tables: a `read_audit` row is
    // forensic evidence about a read, never an input to another read.
    (kinds::READ_AUDIT, Decode::Opaque),
    (kinds::ADMIN_MODEL_CHANGE, Decode::Opaque),
    // Historical Control-owned credential mutation records. C2.3 moved
    // current lifecycle audit into State's atomic semantic feed. These remain
    // opaque so older journal history stays decodable. Like every other
    // `admin-audit` kind: the payload is hand-built canonical JSON, not a
    // proto message, and it carries verifier metadata that has no fact-table
    // consumer.
    (kinds::CREDENTIAL_ENROLLED, Decode::Opaque),
    (kinds::CREDENTIAL_KEY_ADDED, Decode::Opaque),
    (kinds::CREDENTIAL_KEY_RETIRED, Decode::Opaque),
    (kinds::CREDENTIAL_REVOKED, Decode::Opaque),
    (kinds::HANDOFF, Decode::Typed("handoffs")),
    (kinds::HANDOFF_DENIED, Decode::Typed("handoffs")),
    (kinds::SUBAGENT_SPAWN, Decode::Opaque),
    (kinds::SUBAGENT_RESULT, Decode::Opaque),
    (kinds::SUBAGENT_MODEL_CALL, Decode::Opaque),
    (kinds::INGRESS_DIRECTIVE, Decode::Opaque),
];

/// Look up `kind_base`'s registered decode disposition.
///
/// Returns `None` for a kind-base absent from [`REGISTRY`] — which, per the
/// module's `registry_is_exhaustive_over_kinds_rs` test, only ever happens
/// for a string that is not one of `kinds.rs`'s own constants.
#[must_use]
#[allow(dead_code)] // see the `allow` above `Decode`
pub(crate) fn decode_decision(kind_base: &str) -> Option<Decode> {
    REGISTRY
        .iter()
        .find(|(base, _)| *base == kind_base)
        .map(|(_, decision)| *decision)
}

/// Build one `events` wide-table [`RecordBatch`] from `partition`'s events,
/// each paired with its journal position.
///
/// Position is not carried on `Event` itself — the journal assigns it on
/// append/replay (`crates/eventlog-model/src/event.rs`) — so callers pair each
/// event with its position before calling this function (e.g. zipping
/// `EventLog::replay`'s yielded order against a running counter).
///
/// `kind_base` and `turn_id` are derived per row via
/// [`polyc_proto::kinds::parse`] — the platform's single kind-grammar owner,
/// reused here rather than re-derived — so a `base:{turn_uuid}` kind splits
/// into `kind_base = base` and `turn_id = Some(turn_uuid)`; a bare kind (no
/// `:` suffix) yields `turn_id = None`. The `turn_id` column stores the
/// `Uuid`'s canonical hyphenated `to_string()` form; `kinds::tagged` itself
/// writes the compact `as_simple()` form into the stored `kind` string, but
/// nothing downstream keys off preserving that exact formatting through
/// this round trip, so the more common hyphenated rendering is used for the
/// column a human or a `WHERE turn_id = '...'` query actually reads.
///
/// # Errors
///
/// Returns [`ArrowError`] if Arrow batch construction fails.
pub(crate) fn events_batch(
    partition: &str,
    events: &[(u64, Event)],
) -> Result<RecordBatch, ArrowError> {
    let mut partition_b = StringBuilder::with_capacity(events.len(), events.len() * 8);
    let mut position_b = UInt64Builder::with_capacity(events.len());
    let mut kind_b = StringBuilder::with_capacity(events.len(), events.len() * 16);
    let mut kind_base_b = StringBuilder::with_capacity(events.len(), events.len() * 16);
    let mut turn_id_b = StringBuilder::with_capacity(events.len(), events.len() * 36);
    let mut trust_b = StringBuilder::with_capacity(events.len(), events.len() * 20);
    let mut payload_b = BinaryBuilder::with_capacity(events.len(), events.len() * 32);
    let mut payload_json_b = StringBuilder::with_capacity(events.len(), events.len() * 32);

    for (position, event) in events {
        let (base, turn_id) = kinds::parse(&event.kind);

        partition_b.append_value(partition);
        position_b.append_value(*position);
        kind_b.append_value(&event.kind);
        kind_base_b.append_value(base);
        match turn_id {
            Some(id) => turn_id_b.append_value(id.to_string()),
            None => turn_id_b.append_null(),
        }
        trust_b.append_value(event.trust.as_str());
        payload_b.append_value(&event.payload);
        match decode_payload_json(&event.payload) {
            Some(text) => payload_json_b.append_value(text),
            None => payload_json_b.append_null(),
        }
    }

    let columns: Vec<ArrayRef> = vec![
        Arc::new(partition_b.finish()),
        Arc::new(position_b.finish()),
        Arc::new(kind_b.finish()),
        Arc::new(kind_base_b.finish()),
        Arc::new(turn_id_b.finish()),
        Arc::new(trust_b.finish()),
        Arc::new(payload_b.finish()),
        Arc::new(payload_json_b.finish()),
    ];

    RecordBatch::try_new(crate::provider::EventsTableProvider::schema(), columns)
}

/// Best-effort JSON decode of one raw event payload, backing the uniform
/// `payload_json` column every `events_raw` row carries (see the module
/// docs' "The schema-less residual" section).
///
/// `payload` bytes that are valid UTF-8 *and* parse as JSON return `Some`
/// of that UTF-8 text verbatim — not re-serialized, so a SQL caller reads
/// exactly the bytes the journal stored (modulo the UTF-8 decode itself),
/// not a canonicalized re-encoding. Anything else — non-UTF-8 bytes, an
/// empty payload (`serde_json::from_str("")` is not valid JSON), or UTF-8
/// text that fails to parse — returns `None`, which [`events_batch`] writes
/// as a SQL `NULL` rather than an `events_batch` decode error: this column
/// has no failure mode, only "this payload happens to be JSON" or not.
///
/// RFC 8259 also accepts a bare JSON scalar (e.g. the two bytes `42`), not
/// just objects/arrays; in the vanishingly unlikely case a `Typed` kind's
/// protobuf bytes are themselves valid UTF-8 that happens to parse as a
/// bare scalar, `payload_json` populates with that meaningless value — an
/// accepted trade-off, not a bug.
fn decode_payload_json(payload: &[u8]) -> Option<String> {
    let text = std::str::from_utf8(payload).ok()?;
    serde_json::from_str::<serde_json::Value>(text).ok()?;
    Some(text.to_string())
}

/// The framing/key-derivation loop shared by every per-kind typed-table
/// decoder: filter `events` to rows whose kind-base is one of `kind_bases`,
/// decode each row's payload via the caller-supplied `decode` function, and
/// pair a successful decode with the fact model's uniform key columns —
/// `partition` (stored verbatim), `position` (the event's own paired
/// journal position), and `turn_id` (derived via
/// [`polyc_proto::kinds::parse`], the platform's single kind-grammar owner,
/// exactly as [`events_batch`] derives it) — via the caller-supplied
/// `build` closure.
///
/// `kind_bases` is a slice, not a single `&str`, because a typed table's
/// proto shape does not always map to exactly one kind: `usage`/
/// `model_call`/`turn_failed`/`summary` each match one kind and pass a
/// single-element slice, but a multi-kind table (`attribution` was the
/// original example: `caller` and `participant` both decode as
/// `polychrome.events.v1.AttributionEvent`, see
/// `crates/proto/proto/events.proto`'s kind-to-message table) can match on
/// `kind_bases.contains(base)` here, once, rather than concatenating two
/// separate decode passes by hand. `attribution` itself moved off this
/// helper in #1579 (see [`attribution`]'s module docs' "Shared decode"
/// section) once it needed to decode through the shared
/// `polyc_facts::attribution_events_with_positions` fold instead — the
/// slice-typed parameter stays regardless, for the next multi-kind table
/// that does keep using this loop.
///
/// `decode` is a plain `Fn(&[u8]) -> Result<T, E>` — not a `T: buffa::Message`
/// bound resolved internally via
/// [`polyc_proto::events_decode::try_decode_event_payload`] — precisely so
/// a caller can pass a shared `polyc_facts` fold (`usage`/`model_call`, since
/// #1579) instead of the raw wire decode when a second consumer needs the
/// exact same semantic fold; `turn_failed`/`summary`/`fires`/`turn_dispatch`,
/// which have no such second consumer yet, simply pass
/// `try_decode_event_payload::<T>` itself
/// (a function item already has this shape). Either way this helper owns
/// only the loop shape — kind filtering, key derivation, the corrupt-payload
/// skip/warn policy below — never a row's own field-by-field shape, which
/// stays the caller's `build` closure.
///
/// A NON-empty payload that fails to decode is skipped — logged via
/// `tracing::warn!` naming `table_name` — never surfaced as an error. An
/// EMPTY payload decodes cleanly to `T`'s all-defaults value (proto3 elides
/// all-default scalar fields, and every `polyc_facts` fold used here
/// preserves that same contract) and is a legitimate row, not a hole in the
/// batch; see [`usage`]'s module docs for the fuller rationale, which
/// applies unchanged to every caller of this helper.
fn decode_typed_kind_events<T, R, E>(
    partition: &str,
    events: &[(u64, Event)],
    kind_bases: &[&str],
    table_name: &'static str,
    decode: impl Fn(&[u8]) -> Result<T, E>,
    build: impl Fn(String, u64, Option<String>, T) -> R,
) -> Vec<R>
where
    E: std::fmt::Display,
{
    events
        .iter()
        .filter_map(|(position, event)| {
            let (base, turn_id) = kinds::parse(&event.kind);
            if !kind_bases.contains(&base) {
                return None;
            }
            match decode(&event.payload) {
                Ok(decoded) => Some(build(
                    partition.to_string(),
                    *position,
                    turn_id.map(|id| id.to_string()),
                    decoded,
                )),
                Err(e) => {
                    if !event.payload.is_empty() {
                        tracing::warn!(
                            error = %e,
                            len = event.payload.len(),
                            table = table_name,
                            "corrupt event payload; skipping row"
                        );
                    }
                    None
                }
            }
        })
        .collect()
}

#[cfg(test)]
mod tests {
    use arrow::array::Array as _;
    use polyc_eventlog::TrustTag;
    use uuid::Uuid;

    use super::*;

    /// Extract every `pub const FOO: &str = "value";` declared in
    /// `crates/proto/src/kinds.rs`, by parsing its own checked-in source —
    /// not a hand-maintained duplicate list. This is the drift-detection
    /// mechanism: a new kind constant changes the source file, which changes
    /// what this function returns, which this test then compares against
    /// [`REGISTRY`]. See [`extract_pub_const_str_values`] for how it stays
    /// correct even when rustfmt wraps a declaration across more than one
    /// physical line.
    fn kind_constants_from_source() -> Vec<&'static str> {
        // Path is relative to this file (`crates/query/src/decode/mod.rs`),
        // per `include_str!`'s own resolution rule — three levels up reaches
        // `crates/`, then back down into the sibling `proto` crate. Framed
        // workspace-relative, this is `crates/query/../proto/src/kinds.rs`.
        const KINDS_SRC: &str = include_str!("../../../proto/src/kinds.rs");
        extract_pub_const_str_values(KINDS_SRC)
    }

    /// Extract every `pub const NAME: &str = "value";` declaration's string
    /// value out of `source`.
    ///
    /// Locates each declaration by its `pub const ` keyword and its
    /// terminating `;`, then matches the `: &str = "..."` shape by skipping
    /// whitespace between tokens rather than assuming the whole declaration
    /// sits on one physical line. That tolerance matters:
    /// **rustfmt.toml sets `max_width = 100`**, so a sufficiently long
    /// constant name gets wrapped by rustfmt onto a second line (name and
    /// `:` on one line, `&str = "value";` on the next) — a parser that
    /// required the whole thing on one line would silently skip that
    /// constant, and the test below proves it does not.
    fn extract_pub_const_str_values(source: &'static str) -> Vec<&'static str> {
        let mut out = Vec::new();
        let mut cursor = source;
        while let Some(start) = cursor.find("pub const ") {
            let after_keyword = &cursor[start + "pub const ".len()..];
            let Some(statement_end) = after_keyword.find(';') else {
                break;
            };
            let statement = &after_keyword[..statement_end];
            cursor = &after_keyword[statement_end + 1..];

            if let Some(value) = str_value_after_colon(statement) {
                out.push(value);
            }
        }
        out
    }

    /// Within one declaration's body — already sliced to just after
    /// `pub const NAME` up to (not including) the terminating `;` — find the
    /// `: &str = "..."` marker and return the quoted value.
    ///
    /// Whitespace (including a newline, the wrapped case this exists for)
    /// between the `:`, `&str`, `=`, and the opening `"` is skipped rather
    /// than matched literally; the value itself (a plain kind-base name) is
    /// never expected to contain a newline, so it is read verbatim up to the
    /// closing `"`.
    fn str_value_after_colon(statement: &'static str) -> Option<&'static str> {
        let (_name, rest) = statement.split_once(':')?;
        let rest = rest.trim_start().strip_prefix("&str")?.trim_start();
        let rest = rest.strip_prefix('=')?.trim_start();
        let rest = rest.strip_prefix('"')?;
        let (value, _) = rest.split_once('"')?;
        Some(value)
    }

    /// The design's own mandated structural test: [`REGISTRY`] must cover
    /// exactly the kind constants `kinds.rs` declares today — no more, no
    /// fewer. A new kind landing in `kinds.rs` without a matching `REGISTRY`
    /// entry fails this test, and a stale `REGISTRY` entry naming a kind
    /// `kinds.rs` no longer declares fails it too.
    #[test]
    fn registry_is_exhaustive_over_kinds_rs() {
        let mut from_source = kind_constants_from_source();
        from_source.sort_unstable();
        from_source.dedup();

        let mut from_registry: Vec<&str> = REGISTRY.iter().map(|(base, _)| *base).collect();
        from_registry.sort_unstable();

        assert_eq!(
            from_registry.len(),
            REGISTRY.len(),
            "REGISTRY has a duplicate kind-base entry"
        );
        assert_eq!(
            from_source, from_registry,
            "kinds.rs and decode::REGISTRY disagree — a kind constant landed \
             without a typed/opaque decode decision, or REGISTRY names a kind \
             kinds.rs no longer declares"
        );
    }

    /// The regression this parser exists to fix: a rustfmt-wrapped
    /// declaration (constant name and `:` on one line, `&str = "value";` on
    /// the next — exactly what `rustfmt.toml`'s `max_width = 100` produces
    /// for a long enough name) must still be extracted, not silently
    /// dropped. A one-line-per-declaration parser would return zero values
    /// for this fixture; [`extract_pub_const_str_values`] must return one.
    #[test]
    fn extract_pub_const_str_values_tolerates_a_rustfmt_wrapped_declaration() {
        const FIXTURE: &str = concat!(
            "/// A constant whose name alone is long enough that rustfmt wraps\n",
            "/// the declaration across two physical lines.\n",
            "pub const A_DELIBERATELY_VERY_LONG_KIND_CONSTANT_NAME_TO_FORCE_WRAPPING:\n",
            "    &str = \"a_deliberately_very_long_kind_constant_name_to_force_wrapping\";\n",
            "\n",
            "/// A normal, single-line declaration, to prove both shapes coexist.\n",
            "pub const SHORT: &str = \"short\";\n",
        );

        let values = extract_pub_const_str_values(FIXTURE);

        assert_eq!(
            values,
            vec![
                "a_deliberately_very_long_kind_constant_name_to_force_wrapping",
                "short",
            ],
            "both the wrapped and single-line declarations must be extracted"
        );
    }

    /// [`REGISTRY`]'s typed kinds today: `usage`, `model_call`
    /// (#1311's second typed table), `caller`/`participant` (#1311's third
    /// typed table, `attribution` — the first typed table backed by more
    /// than one kind, since both decode as
    /// `polychrome.events.v1.AttributionEvent`; see
    /// `crate::decode::attribution`'s module docs), `turn_failed` (#1311's
    /// fourth typed table; see `crate::decode::turn_failed`'s module docs),
    /// `payment_receipt`/`outbound_payment_receipt` (`payments`, the fifth
    /// typed table and the first FOLD-COUPLED one — see
    /// `crate::decode::payments`'s module docs), `user_msg`/`output_msg`
    /// (`message_content`, the sixth typed table and the first whose decode
    /// fans out to more than one SQL table — see
    /// `crate::decode::message_content`'s module docs and
    /// [`Decode::Typed`]'s own doc for why one `REGISTRY` label still
    /// covers it), `approval_request`/`approval_response` (`approvals`,
    /// the seventh typed table and the second FOLD-COUPLED one — see
    /// `crate::decode::approvals`'s module docs), `handoff`/
    /// `handoff_denied` (`handoffs`, the eighth typed table — a DIRECT
    /// decode, not fold-coupled; see `crate::decode::handoffs`'s module
    /// docs), `grant_replay` (`grant_replays`, the ninth typed table and
    /// the THIRD FOLD-COUPLED one — see `crate::decode::grant_replays`'s
    /// module docs), `summary` (the tenth typed table — a DIRECT decode,
    /// like `handoffs`; see `crate::decode::summary`'s module docs), and
    /// `routine_fired` (`fires`, the eleventh typed table — a DIRECT decode
    /// of the routine scheduler's durable fire markers, Fleet-only like
    /// `summary`; see `crate::decode::fires`'s module docs), `turn_dispatched`
    /// (`turn_dispatch`, the twelfth typed table, issue #1593 — see
    /// `crate::decode::turn_dispatch`'s module docs), and
    /// `routine_created`/`routine_paused`/`routine_resumed`/`routine_deleted`
    /// (`routine_lifecycle`, the thirteenth typed table and the first backed
    /// by FOUR kinds, issue #1593 — see
    /// `crate::decode::routine_lifecycle`'s module docs). A new
    /// kind must be added here deliberately, in the same
    /// review as its `REGISTRY` entry — this test pins the exact set rather
    /// than just its count, so a stray/omitted entry fails loudly instead of
    /// silently changing `typed.len()`.
    #[test]
    #[allow(clippy::too_many_lines)] // one assertion per typed kind, mechanical
    fn usage_model_call_attribution_turn_failed_payments_message_content_approvals_handoffs_grant_replays_summary_fires_turn_dispatch_routine_lifecycle_and_routine_setup_kinds_are_the_only_typed_kinds()
     {
        let typed: Vec<&str> = REGISTRY
            .iter()
            .filter_map(|(base, decision)| match decision {
                Decode::Typed(_) => Some(*base),
                Decode::Opaque => None,
            })
            .collect();
        assert_eq!(
            typed,
            vec![
                kinds::TURN_FAILED,
                kinds::TURN_DISPATCHED,
                kinds::USER_MSG,
                kinds::OUTPUT_MSG,
                kinds::USAGE,
                kinds::MODEL_CALL,
                kinds::SUMMARY,
                kinds::APPROVAL_REQUEST,
                kinds::APPROVAL_RESPONSE,
                kinds::PAYMENT_RECEIPT,
                kinds::CALLER,
                kinds::PARTICIPANT,
                kinds::OUTBOUND_PAYMENT_RECEIPT,
                kinds::PAYMENT_REFUSAL,
                kinds::WALLET_LINK_LIFECYCLE,
                kinds::ROUTINE_FIRED,
                kinds::ROUTINE_FIRE_OUTCOME,
                kinds::ROUTINE_CREATED,
                kinds::ROUTINE_PAUSED,
                kinds::ROUTINE_RESUMED,
                kinds::ROUTINE_DELETED,
                kinds::ROUTINE_SCOPE_CHANGED,
                kinds::ROUTINE_SETUP_COMPLETED,
                kinds::GRANT_REPLAY,
                kinds::HANDOFF,
                kinds::HANDOFF_DENIED,
            ]
        );
        assert_eq!(
            decode_decision(kinds::TURN_FAILED),
            Some(Decode::Typed("turn_failed"))
        );
        assert_eq!(
            decode_decision(kinds::USER_MSG),
            Some(Decode::Typed("message_content"))
        );
        assert_eq!(
            decode_decision(kinds::OUTPUT_MSG),
            Some(Decode::Typed("message_content"))
        );
        assert_eq!(
            decode_decision(kinds::TURN_DISPATCHED),
            Some(Decode::Typed("turn_dispatch"))
        );
        assert_eq!(decode_decision(kinds::USAGE), Some(Decode::Typed("usage")));
        assert_eq!(
            decode_decision(kinds::MODEL_CALL),
            Some(Decode::Typed("model_call"))
        );
        assert_eq!(
            decode_decision(kinds::SUMMARY),
            Some(Decode::Typed("summary"))
        );
        assert_eq!(
            decode_decision(kinds::APPROVAL_REQUEST),
            Some(Decode::Typed("approvals"))
        );
        assert_eq!(
            decode_decision(kinds::APPROVAL_RESPONSE),
            Some(Decode::Typed("approvals"))
        );
        assert_eq!(
            decode_decision(kinds::CALLER),
            Some(Decode::Typed("attribution"))
        );
        assert_eq!(
            decode_decision(kinds::PARTICIPANT),
            Some(Decode::Typed("attribution"))
        );
        assert_eq!(
            decode_decision(kinds::PAYMENT_RECEIPT),
            Some(Decode::Typed("payments"))
        );
        assert_eq!(
            decode_decision(kinds::OUTBOUND_PAYMENT_RECEIPT),
            Some(Decode::Typed("payments"))
        );
        assert_eq!(
            decode_decision(kinds::PAYMENT_REFUSAL),
            Some(Decode::Typed("refusals"))
        );
        assert_eq!(
            decode_decision(kinds::WALLET_LINK_LIFECYCLE),
            Some(Decode::Typed("wallet_link_lifecycle"))
        );
        assert_eq!(
            decode_decision(kinds::ROUTINE_FIRED),
            Some(Decode::Typed("fires"))
        );
        assert_eq!(
            decode_decision(kinds::ROUTINE_FIRE_OUTCOME),
            Some(Decode::Typed("fires"))
        );
        assert_eq!(
            decode_decision(kinds::ROUTINE_CREATED),
            Some(Decode::Typed("routine_lifecycle"))
        );
        assert_eq!(
            decode_decision(kinds::ROUTINE_PAUSED),
            Some(Decode::Typed("routine_lifecycle"))
        );
        assert_eq!(
            decode_decision(kinds::ROUTINE_RESUMED),
            Some(Decode::Typed("routine_lifecycle"))
        );
        assert_eq!(
            decode_decision(kinds::ROUTINE_DELETED),
            Some(Decode::Typed("routine_lifecycle"))
        );
        assert_eq!(
            decode_decision(kinds::ROUTINE_SCOPE_CHANGED),
            Some(Decode::Typed("routine_lifecycle"))
        );
        assert_eq!(
            decode_decision(kinds::ROUTINE_SETUP_COMPLETED),
            Some(Decode::Typed("routine_setup"))
        );
        assert_eq!(
            decode_decision(kinds::GRANT_REPLAY),
            Some(Decode::Typed("grant_replays"))
        );
        assert_eq!(
            decode_decision(kinds::HANDOFF),
            Some(Decode::Typed("handoffs"))
        );
        assert_eq!(
            decode_decision(kinds::HANDOFF_DENIED),
            Some(Decode::Typed("handoffs"))
        );
    }

    #[test]
    fn unknown_kind_base_has_no_decision() {
        assert_eq!(decode_decision("totally_invented_kind"), None);
    }

    fn sample_events() -> Vec<(u64, Event)> {
        let turn = Uuid::from_u128(0x0195_abcd_ef01_2345_6789_abcd_ef01_2345);
        vec![
            (10, Event::new(kinds::TURN_START, Vec::new())),
            (
                11,
                Event::trusted(kinds::tagged(kinds::USER_MSG, &turn), b"hello".to_vec()),
            ),
            (
                12,
                Event::quarantined(kinds::tagged(kinds::USAGE, &turn), b"usage-bytes".to_vec()),
            ),
            (
                // A schema-less kind (no proto schema exists for
                // `approval_deferred`) whose payload happens to be JSON —
                // the case `payload_json` exists for.
                13,
                Event::new(
                    kinds::APPROVAL_DEFERRED,
                    br#"{"reason":"awaiting reviewer"}"#.to_vec(),
                ),
            ),
        ]
    }

    #[test]
    fn events_batch_round_trips_every_column() {
        let events = sample_events();
        let batch = events_batch("conv-42", &events).expect("batch build");

        assert_eq!(batch.num_rows(), 4);
        assert_eq!(
            batch.schema(),
            crate::provider::EventsTableProvider::schema()
        );

        let partition = batch
            .column(0)
            .as_any()
            .downcast_ref::<arrow::array::StringArray>()
            .unwrap();
        for i in 0..4 {
            assert_eq!(partition.value(i), "conv-42");
        }

        let position = batch
            .column(1)
            .as_any()
            .downcast_ref::<arrow::array::UInt64Array>()
            .unwrap();
        assert_eq!(position.values(), &[10, 11, 12, 13]);

        let kind = batch
            .column(2)
            .as_any()
            .downcast_ref::<arrow::array::StringArray>()
            .unwrap();
        assert_eq!(kind.value(0), kinds::TURN_START);
        assert_eq!(
            kind.value(1),
            kinds::tagged(
                kinds::USER_MSG,
                &Uuid::from_u128(0x0195_abcd_ef01_2345_6789_abcd_ef01_2345)
            )
        );

        let kind_base = batch
            .column(3)
            .as_any()
            .downcast_ref::<arrow::array::StringArray>()
            .unwrap();
        assert_eq!(kind_base.value(0), kinds::TURN_START);
        assert_eq!(kind_base.value(1), kinds::USER_MSG);
        assert_eq!(kind_base.value(2), kinds::USAGE);
        assert_eq!(kind_base.value(3), kinds::APPROVAL_DEFERRED);

        let turn_id = batch
            .column(4)
            .as_any()
            .downcast_ref::<arrow::array::StringArray>()
            .unwrap();
        assert!(turn_id.is_null(0));
        let expected_turn = Uuid::from_u128(0x0195_abcd_ef01_2345_6789_abcd_ef01_2345).to_string();
        assert_eq!(turn_id.value(1), expected_turn);
        assert_eq!(turn_id.value(2), expected_turn);
        assert!(
            turn_id.is_null(3),
            "the bare approval_deferred kind tags no turn"
        );

        let trust = batch
            .column(5)
            .as_any()
            .downcast_ref::<arrow::array::StringArray>()
            .unwrap();
        assert_eq!(trust.value(0), TrustTag::Unspecified.as_str());
        assert_eq!(trust.value(1), TrustTag::TrustedUser.as_str());
        assert_eq!(trust.value(2), TrustTag::QuarantinedContent.as_str());
        assert_eq!(trust.value(3), TrustTag::Unspecified.as_str());

        let payload = batch
            .column(6)
            .as_any()
            .downcast_ref::<arrow::array::BinaryArray>()
            .unwrap();
        assert_eq!(payload.value(0), b"");
        assert_eq!(payload.value(1), b"hello");
        assert_eq!(payload.value(2), b"usage-bytes");
        assert_eq!(payload.value(3), br#"{"reason":"awaiting reviewer"}"#);

        let payload_json = batch
            .column(7)
            .as_any()
            .downcast_ref::<arrow::array::StringArray>()
            .unwrap();
        assert!(
            payload_json.is_null(0),
            "an empty payload is not valid JSON"
        );
        assert!(
            payload_json.is_null(1),
            "plain text (\"hello\") is not valid JSON"
        );
        assert!(
            payload_json.is_null(2),
            "opaque non-JSON bytes decode to NULL, not a decode error"
        );
        assert_eq!(
            payload_json.value(3),
            r#"{"reason":"awaiting reviewer"}"#,
            "a schema-less kind's JSON payload round-trips as text verbatim"
        );
    }

    #[test]
    fn events_batch_empty_input_has_zero_rows_and_the_full_schema() {
        let batch = events_batch("conv-empty", &[]).expect("batch build");
        assert_eq!(batch.num_rows(), 0);
        assert_eq!(
            batch.schema(),
            crate::provider::EventsTableProvider::schema()
        );
    }

    #[test]
    fn decode_payload_json_accepts_a_json_object() {
        assert_eq!(
            decode_payload_json(br#"{"a":1,"b":"two"}"#),
            Some(r#"{"a":1,"b":"two"}"#.to_string())
        );
    }

    #[test]
    fn decode_payload_json_accepts_a_bare_json_scalar() {
        // JSON's grammar allows a top-level scalar, not just an
        // object/array — `serde_json::from_str` (and therefore this
        // column) accepts it too, so a schema-less kind whose payload is
        // e.g. a bare JSON string or number still round-trips.
        assert_eq!(
            decode_payload_json(br#""just a string""#),
            Some(r#""just a string""#.to_string())
        );
        assert_eq!(decode_payload_json(b"42"), Some("42".to_string()));
    }

    #[test]
    fn decode_payload_json_rejects_empty_payload() {
        assert_eq!(decode_payload_json(b""), None);
    }

    #[test]
    fn decode_payload_json_rejects_non_json_utf8_text() {
        assert_eq!(decode_payload_json(b"hello, not json"), None);
    }

    #[test]
    fn decode_payload_json_rejects_non_utf8_bytes() {
        assert_eq!(decode_payload_json(&[0xFF, 0xFE, 0x00, 0x01]), None);
    }
}