nornir 0.5.3

//! Iceberg writer + reader for the funnel event log.
//!
//! Every funnel mutation (`IdeaSubmitted`, `NodeAdded`, …) lands as one
//! row in the `funnel_events` Iceberg table. Each append is its own
//! Iceberg snapshot, so the table at snapshot T == the funnel state at
//! time T (free time-travel).
//!
//! The schema is wide and typed (see [`super::iceberg_schema::funnel_events`]):
//!   * `refs`, `targets`, `produced_test_runs` → `list<string>`
//!   * `produced_commits` → `list<struct{repo, sha}>`
//! Only `params_json` is a JSON string, because `NodeAdded.params` is an
//! open-ended `serde_json::Map`.
//!
//! No row-level deletes, no upserts: a "node done" is a new
//! `NodeStatusChanged` row. Folding the rows back into a `Funnel`
//! struct happens in [`super::super::funnel::state::Funnel::apply`].

use std::sync::Arc;

use anyhow::{anyhow, Context, Result};
use arrow::array::{
    Array, ArrayBuilder, ListBuilder, RecordBatch, StringArray, StringBuilder,
    StructBuilder, TimestampMicrosecondArray,
};
use arrow::datatypes::{DataType, FieldRef, Fields, Schema as ArrowSchema, TimeUnit};
use chrono::{DateTime, TimeZone, Utc};
use iceberg::Catalog;
use iceberg::arrow::schema_to_arrow_schema;
use uuid::Uuid;

use super::iceberg::{IcebergWarehouse, TABLE_FUNNEL_EVENTS, append_batch, ensure_table_schema};
use crate::funnel::event::{
    CommitRef, Event, ItemKind, NodeStatus, PlanStatus, RunOutcome, TriageDecision,
};
use crate::funnel::ids::{IdeaId, NodeId, PlanId, RunId};

// Column indices match the `iceberg_schema::funnel_events` field order.
const COL_EVENT_ID: usize = 0;
const COL_TS_MICROS: usize = 1;
const COL_KIND: usize = 2;
const COL_IDEA_ID: usize = 3;
const COL_PLAN_ID: usize = 4;
const COL_NODE_ID: usize = 5;
const COL_RUN_ID: usize = 6;
const COL_FROM_NODE: usize = 7;
const COL_TO_NODE: usize = 8;
const COL_SOURCE: usize = 9;
const COL_TEXT: usize = 10;
const COL_REFS: usize = 11;
const COL_DECISION: usize = 12;
const COL_NODE_STATUS: usize = 13;
const COL_PLAN_STATUS: usize = 14;
const COL_WHY: usize = 15;
const COL_SUMMARY: usize = 16;
const COL_PLANNER: usize = 17;
const COL_NODE_KIND: usize = 18;
const COL_TARGETS: usize = 19;
const COL_PROMPT_EXCERPT: usize = 20;
const COL_PARAMS_JSON: usize = 21;
const COL_RAN_BY: usize = 22;
const COL_OUTCOME: usize = 23;
const COL_LOG_REF: usize = 24;
const COL_PRODUCED_COMMITS: usize = 25;
const COL_PRODUCED_TEST_RUNS: usize = 26;
const COL_ITEM_KIND: usize = 27;
// funnel-prompt-planner keystone (§4): five more OPTIONAL columns, added the
// same migration-safe way `item_kind` did. `IdeaModeSet`/`CardAssigned`/
// `CardStateChanged` reuse `COL_IDEA_ID`/`COL_NODE_ID` (a generic card id) and
// `COL_WHY` — only genuinely new fields get a new column.
const COL_MODE: usize = 28;
const COL_MEMBER: usize = 29;
const COL_ROLE: usize = 30;
const COL_FROM_LANE: usize = 31;
const COL_TO_LANE: usize = 32;

const KIND_IDEA_SUBMITTED: &str = "IdeaSubmitted";
const KIND_IDEA_TRIAGED: &str = "IdeaTriaged";
const KIND_PLAN_CREATED: &str = "PlanCreated";
const KIND_NODE_ADDED: &str = "NodeAdded";
const KIND_EDGE_ADDED: &str = "EdgeAdded";
const KIND_NODE_STATUS_CHANGED: &str = "NodeStatusChanged";
const KIND_RUN_RECORDED: &str = "RunRecorded";
const KIND_PLAN_STATUS_CHANGED: &str = "PlanStatusChanged";
const KIND_IDEA_MODE_SET: &str = "IdeaModeSet";
const KIND_CARD_ASSIGNED: &str = "CardAssigned";
const KIND_CARD_STATE_CHANGED: &str = "CardStateChanged";

/// Append one event. Each call produces a new Iceberg snapshot.
pub async fn append_event(wh: &IcebergWarehouse, event: &Event) -> Result<()> {
    let ident = wh.table_ident(TABLE_FUNNEL_EVENTS);
    let table = wh.catalog().load_table(&ident).await?;
    // A server whose `funnel_events` table predates the FI `item_kind` column
    // (field 28) carries only 27 columns; evolve it forward (Iceberg add-column)
    // so the full row appends cleanly. No-op once the table is current.
    let table = ensure_table_schema(
        wh.catalog(),
        &ident,
        table,
        &super::iceberg_schema::funnel_events()?,
    )
    .await?;
    let batch = build_batch(&table, std::slice::from_ref(event))?;
    append_batch(wh.catalog(), table, batch).await
}

/// Load every funnel event in original write order. Used to
/// reconstruct [`crate::funnel::state::Funnel`] on startup.
pub async fn load_all_events(wh: &IcebergWarehouse) -> Result<Vec<Event>> {
    // Blank/torn Iceberg metadata JSON → 0 rows, never a JSON-EOF crash (load+read).
    let batches: Vec<RecordBatch> =
        super::iceberg::load_and_read_all(wh, TABLE_FUNNEL_EVENTS).await?;

    let mut events: Vec<Event> = Vec::new();
    for batch in &batches {
        let chunk = parse_batch(batch)?;
        events.extend(chunk);
    }
    // Iceberg doesn't guarantee scan order. Sort by ts so replay is
    // deterministic.
    events.sort_by_key(|e| e.ts());
    Ok(events)
}

// ── Ragnar-fast point-in-time index over the funnel event log ────────────────
//
// "the funnel as of time T" is a fold of every event with `ts <= T`. Finding
// that replay cut is otherwise a linear scan of the (ts-sorted) log; the static
// index makes it `O(log n)`. Build once, query many point-in-time views.

/// A static [`RagnarLog`](super::ragnar::RagnarLog) over the funnel event log,
/// keyed by `ts_micros`. Its payload at a distinct timestamp is the **prefix
/// count**: the number of events with `ts <= that timestamp`. A point-in-time
/// query floors `T` to the newest covered timestamp and reads that prefix count
/// — the exact `&events[..cut]` slice a linear `filter(ts <= T)` would produce,
/// found in `O(log n)`.
pub struct FunnelLogIndex {
    idx: super::ragnar::RagnarLog<usize>,
}

impl FunnelLogIndex {
    /// Build from events **already sorted ascending by `ts`** (as
    /// [`load_all_events`] returns them). Collapses equal timestamps to one
    /// entry carrying the cumulative event count.
    pub fn build(events_sorted_by_ts: &[Event]) -> Self {
        // (ts_micros, prefix_end) where prefix_end = count of events with ts <=
        // this ts. Because the input is ts-sorted, the running index+1 at the
        // last event of each timestamp run is exactly that prefix count. Feeding
        // one row per event (last-writer-wins on equal ts) lands the highest
        // prefix_end for each timestamp, which is what we want.
        let rows: Vec<(i64, usize)> = events_sorted_by_ts
            .iter()
            .enumerate()
            .map(|(i, e)| (e.ts().timestamp_micros(), i + 1))
            .collect();
        Self {
            idx: super::ragnar::RagnarLog::build(rows),
        }
    }

    /// Number of events with `ts <= ts_micros` — the replay cut. `0` when the
    /// query precedes the whole log.
    #[inline]
    pub fn cut_as_of(&self, ts_micros: i64) -> usize {
        self.idx.as_of(ts_micros).copied().unwrap_or(0)
    }
}

/// The events in effect **as of** `ts_micros` — `events[..cut]` where `cut` is
/// found in `O(log n)` via a one-shot [`FunnelLogIndex`]. `events` must be
/// ts-sorted ascending (as [`load_all_events`] returns it). For repeated
/// point-in-time queries, build a [`FunnelLogIndex`] once and reuse it instead.
pub fn events_as_of(events_sorted_by_ts: &[Event], ts_micros: i64) -> &[Event] {
    let cut = FunnelLogIndex::build(events_sorted_by_ts).cut_as_of(ts_micros);
    &events_sorted_by_ts[..cut]
}

// ---------- write path ----------

fn build_batch(
    table: &iceberg::table::Table,
    events: &[Event],
) -> Result<RecordBatch> {
    let schema = Arc::new(schema_to_arrow_schema(table.metadata().current_schema())?);

    // Pull the iceberg-derived list element fields so our ListBuilders
    // produce arrays whose Field (name + nullability + PARQUET:field_id
    // metadata) matches the table schema exactly. Without this,
    // RecordBatch::try_new rejects the column.
    let refs_elem = list_element_field(&schema, COL_REFS)?;
    let targets_elem = list_element_field(&schema, COL_TARGETS)?;
    let test_runs_elem = list_element_field(&schema, COL_PRODUCED_TEST_RUNS)?;
    let commits_elem = list_element_field(&schema, COL_PRODUCED_COMMITS)?;
    let commit_struct_fields: Fields = match commits_elem.data_type() {
        DataType::Struct(fs) => fs.clone(),
        other => return Err(anyhow!("produced_commits element not Struct: {other:?}")),
    };

    // String column accumulators (None = null).
    let mut event_ids: Vec<String> = Vec::with_capacity(events.len());
    let mut ts_vals: Vec<i64> = Vec::with_capacity(events.len());
    let mut kinds: Vec<String> = Vec::with_capacity(events.len());
    let mut idea_ids: Vec<Option<String>> = Vec::with_capacity(events.len());
    let mut plan_ids: Vec<Option<String>> = Vec::with_capacity(events.len());
    let mut node_ids: Vec<Option<String>> = Vec::with_capacity(events.len());
    let mut run_ids: Vec<Option<String>> = Vec::with_capacity(events.len());
    let mut from_nodes: Vec<Option<String>> = Vec::with_capacity(events.len());
    let mut to_nodes: Vec<Option<String>> = Vec::with_capacity(events.len());
    let mut sources: Vec<Option<String>> = Vec::with_capacity(events.len());
    let mut texts: Vec<Option<String>> = Vec::with_capacity(events.len());
    let mut decisions: Vec<Option<String>> = Vec::with_capacity(events.len());
    let mut node_statuses: Vec<Option<String>> = Vec::with_capacity(events.len());
    let mut plan_statuses: Vec<Option<String>> = Vec::with_capacity(events.len());
    let mut whys: Vec<Option<String>> = Vec::with_capacity(events.len());
    let mut summaries: Vec<Option<String>> = Vec::with_capacity(events.len());
    let mut planners: Vec<Option<String>> = Vec::with_capacity(events.len());
    let mut node_kinds: Vec<Option<String>> = Vec::with_capacity(events.len());
    let mut prompt_excerpts: Vec<Option<String>> = Vec::with_capacity(events.len());
    let mut params_jsons: Vec<Option<String>> = Vec::with_capacity(events.len());
    let mut ran_bys: Vec<Option<String>> = Vec::with_capacity(events.len());
    let mut outcomes: Vec<Option<String>> = Vec::with_capacity(events.len());
    let mut log_refs: Vec<Option<String>> = Vec::with_capacity(events.len());
    let mut item_kinds: Vec<Option<String>> = Vec::with_capacity(events.len());
    // funnel-prompt-planner keystone.
    let mut modes: Vec<Option<String>> = Vec::with_capacity(events.len());
    let mut members: Vec<Option<String>> = Vec::with_capacity(events.len());
    let mut roles: Vec<Option<String>> = Vec::with_capacity(events.len());
    let mut from_lanes: Vec<Option<String>> = Vec::with_capacity(events.len());
    let mut to_lanes: Vec<Option<String>> = Vec::with_capacity(events.len());

    // List<string> builders, each wearing the iceberg-derived element field.
    let mut refs_b: ListBuilder<StringBuilder> =
        ListBuilder::new(StringBuilder::new()).with_field(refs_elem);
    let mut targets_b: ListBuilder<StringBuilder> =
        ListBuilder::new(StringBuilder::new()).with_field(targets_elem);
    let mut test_runs_b: ListBuilder<StringBuilder> =
        ListBuilder::new(StringBuilder::new()).with_field(test_runs_elem);

    // List<struct{repo,sha}> builder, struct fields driven by the schema.
    let commit_struct_builder = StructBuilder::new(
        commit_struct_fields.clone(),
        vec![
            Box::new(StringBuilder::new()) as Box<dyn ArrayBuilder>,
            Box::new(StringBuilder::new()) as Box<dyn ArrayBuilder>,
        ],
    );
    let mut commits_b: ListBuilder<StructBuilder> =
        ListBuilder::new(commit_struct_builder).with_field(commits_elem);

    for ev in events {
        event_ids.push(Uuid::new_v4().to_string());
        ts_vals.push(ev.ts().timestamp_micros());
        kinds.push(event_kind_str(ev).to_string());

        // Defaults: everything null, then fill per variant.
        let mut idea_id = None;
        let mut plan_id = None;
        let mut node_id = None;
        let mut run_id = None;
        let mut from_node = None;
        let mut to_node = None;
        let mut source = None;
        let mut text = None;
        let mut decision = None;
        let mut node_status = None;
        let mut plan_status = None;
        let mut why = None;
        let mut summary = None;
        let mut planner = None;
        let mut node_kind = None;
        let mut prompt_excerpt = None;
        let mut params_json = None;
        let mut ran_by = None;
        let mut outcome = None;
        let mut log_ref = None;
        let mut item_kind: Option<String> = None;
        let mut mode: Option<String> = None;
        let mut member: Option<String> = None;
        let mut role: Option<String> = None;
        let mut from_lane: Option<String> = None;
        let mut to_lane: Option<String> = None;
        let mut refs_list: Option<&Vec<String>> = None;
        let mut targets_list: Option<&Vec<String>> = None;
        let mut test_runs_list: Option<&Vec<String>> = None;
        let mut commits_list: Option<&Vec<CommitRef>> = None;

        match ev {
            Event::IdeaSubmitted { id, source: s, text: t, refs, item_kind: ik, .. } => {
                idea_id = Some(id.as_str().to_string());
                source = Some(s.clone());
                text = Some(t.clone());
                refs_list = Some(refs);
                item_kind = Some(ik.as_str().to_string());
            }
            Event::IdeaTriaged { idea_id: iid, decision: d, why: w, .. } => {
                idea_id = Some(iid.as_str().to_string());
                decision = Some(triage_str(*d).to_string());
                why = w.clone();
            }
            Event::PlanCreated { id, idea_id: iid, summary: s, planner: p, .. } => {
                plan_id = Some(id.as_str().to_string());
                idea_id = Some(iid.as_str().to_string());
                summary = Some(s.clone());
                planner = Some(p.clone());
            }
            Event::NodeAdded { plan_id: pid, node_id: nid, kind, params, targets, prompt_excerpt: pe, .. } => {
                plan_id = Some(pid.as_str().to_string());
                node_id = Some(nid.as_str().to_string());
                node_kind = Some(kind.clone());
                prompt_excerpt = pe.clone();
                targets_list = Some(targets);
                if !params.is_empty() {
                    params_json = Some(serde_json::to_string(params).unwrap_or_default());
                }
            }
            Event::EdgeAdded { plan_id: pid, from_node: f, to_node: t, .. } => {
                plan_id = Some(pid.as_str().to_string());
                from_node = Some(f.as_str().to_string());
                to_node = Some(t.as_str().to_string());
            }
            Event::NodeStatusChanged { plan_id: pid, node_id: nid, status, why: w, .. } => {
                plan_id = Some(pid.as_str().to_string());
                node_id = Some(nid.as_str().to_string());
                node_status = Some(node_status_str(*status).to_string());
                why = w.clone();
            }
            Event::RunRecorded { plan_id: pid, node_id: nid, run_id: rid, ran_by: rb, outcome: o, log_ref: lr, produced_commits, produced_test_runs, .. } => {
                plan_id = Some(pid.as_str().to_string());
                node_id = Some(nid.as_str().to_string());
                run_id = Some(rid.as_str().to_string());
                ran_by = Some(rb.clone());
                outcome = Some(run_outcome_str(*o).to_string());
                log_ref = lr.clone();
                commits_list = Some(produced_commits);
                test_runs_list = Some(produced_test_runs);
            }
            Event::PlanStatusChanged { plan_id: pid, status, why: w, .. } => {
                plan_id = Some(pid.as_str().to_string());
                plan_status = Some(plan_status_str(*status).to_string());
                why = w.clone();
            }
            Event::IdeaModeSet { idea_id: iid, mode: m, .. } => {
                idea_id = Some(iid.as_str().to_string());
                mode = Some(m.clone());
            }
            Event::CardAssigned { node, member: mem, role: r, why: w, .. } => {
                node_id = Some(node.clone());
                member = Some(mem.clone());
                role = r.clone();
                why = w.clone();
            }
            Event::CardStateChanged { node, from_lane: fl, to_lane: tl, why: w, .. } => {
                node_id = Some(node.clone());
                from_lane = fl.clone();
                to_lane = Some(tl.clone());
                why = w.clone();
            }
        }

        idea_ids.push(idea_id);
        plan_ids.push(plan_id);
        node_ids.push(node_id);
        run_ids.push(run_id);
        from_nodes.push(from_node);
        to_nodes.push(to_node);
        sources.push(source);
        texts.push(text);
        decisions.push(decision);
        node_statuses.push(node_status);
        plan_statuses.push(plan_status);
        whys.push(why);
        summaries.push(summary);
        planners.push(planner);
        node_kinds.push(node_kind);
        prompt_excerpts.push(prompt_excerpt);
        params_jsons.push(params_json);
        ran_bys.push(ran_by);
        outcomes.push(outcome);
        log_refs.push(log_ref);
        item_kinds.push(item_kind);
        modes.push(mode);
        members.push(member);
        roles.push(role);
        from_lanes.push(from_lane);
        to_lanes.push(to_lane);

        append_string_list(&mut refs_b, refs_list);
        append_string_list(&mut targets_b, targets_list);
        append_string_list(&mut test_runs_b, test_runs_list);
        append_commit_list(&mut commits_b, commits_list);
    }

    // Size the column vector to the table's ACTUAL field count. `append_event`
    // evolves a stale table to the full canonical schema before calling us, so
    // this is `N_COLS` in steady state; sizing off the schema keeps us correct
    // even if a caller hands us a table at a different schema version.
    let n_fields = schema.fields().len();
    let mut cols: Vec<Arc<dyn Array>> =
        vec![Arc::new(StringArray::from(Vec::<String>::new())); n_fields];
    cols[COL_EVENT_ID] = Arc::new(StringArray::from(event_ids));
    cols[COL_TS_MICROS] = Arc::new(
        TimestampMicrosecondArray::from(ts_vals).with_timezone("+00:00"),
    );
    cols[COL_KIND] = Arc::new(StringArray::from(kinds));
    cols[COL_IDEA_ID] = Arc::new(StringArray::from(idea_ids));
    cols[COL_PLAN_ID] = Arc::new(StringArray::from(plan_ids));
    cols[COL_NODE_ID] = Arc::new(StringArray::from(node_ids));
    cols[COL_RUN_ID] = Arc::new(StringArray::from(run_ids));
    cols[COL_FROM_NODE] = Arc::new(StringArray::from(from_nodes));
    cols[COL_TO_NODE] = Arc::new(StringArray::from(to_nodes));
    cols[COL_SOURCE] = Arc::new(StringArray::from(sources));
    cols[COL_TEXT] = Arc::new(StringArray::from(texts));
    cols[COL_REFS] = Arc::new(refs_b.finish());
    cols[COL_DECISION] = Arc::new(StringArray::from(decisions));
    cols[COL_NODE_STATUS] = Arc::new(StringArray::from(node_statuses));
    cols[COL_PLAN_STATUS] = Arc::new(StringArray::from(plan_statuses));
    cols[COL_WHY] = Arc::new(StringArray::from(whys));
    cols[COL_SUMMARY] = Arc::new(StringArray::from(summaries));
    cols[COL_PLANNER] = Arc::new(StringArray::from(planners));
    cols[COL_NODE_KIND] = Arc::new(StringArray::from(node_kinds));
    cols[COL_TARGETS] = Arc::new(targets_b.finish());
    cols[COL_PROMPT_EXCERPT] = Arc::new(StringArray::from(prompt_excerpts));
    cols[COL_PARAMS_JSON] = Arc::new(StringArray::from(params_jsons));
    cols[COL_RAN_BY] = Arc::new(StringArray::from(ran_bys));
    cols[COL_OUTCOME] = Arc::new(StringArray::from(outcomes));
    cols[COL_LOG_REF] = Arc::new(StringArray::from(log_refs));
    cols[COL_PRODUCED_COMMITS] = Arc::new(commits_b.finish());
    cols[COL_PRODUCED_TEST_RUNS] = Arc::new(test_runs_b.finish());
    // FI item_kind (field 28) — only set when the (evolved) table carries it.
    if n_fields > COL_ITEM_KIND {
        cols[COL_ITEM_KIND] = Arc::new(StringArray::from(item_kinds));
    }
    // funnel-prompt-planner keystone (fields 29-33) — same "only if the
    // (evolved) table carries it" migration-safety guard.
    if n_fields > COL_MODE {
        cols[COL_MODE] = Arc::new(StringArray::from(modes));
    }
    if n_fields > COL_MEMBER {
        cols[COL_MEMBER] = Arc::new(StringArray::from(members));
    }
    if n_fields > COL_ROLE {
        cols[COL_ROLE] = Arc::new(StringArray::from(roles));
    }
    if n_fields > COL_FROM_LANE {
        cols[COL_FROM_LANE] = Arc::new(StringArray::from(from_lanes));
    }
    if n_fields > COL_TO_LANE {
        cols[COL_TO_LANE] = Arc::new(StringArray::from(to_lanes));
    }

    // Use a TimestampMicrosecondArray with timezone; ensure schema matches.
    // The schema_to_arrow_schema call above already gives us the right field.
    let _ = TimeUnit::Microsecond; // suppress unused-import warning
    Ok(RecordBatch::try_new(schema, cols)?)
}

fn append_string_list(b: &mut ListBuilder<StringBuilder>, list: Option<&Vec<String>>) {
    match list {
        None => b.append(false),
        Some(v) => {
            for s in v {
                b.values().append_value(s);
            }
            b.append(true);
        }
    }
}

/// Pull the inner element `Field` of a list column from an Arrow schema.
/// Iceberg's schema_to_arrow_schema attaches `PARQUET:field_id` metadata
/// that RecordBatch::try_new validates; using this field on the
/// ListBuilder makes our column metadata match the table exactly.
fn list_element_field(schema: &ArrowSchema, idx: usize) -> Result<FieldRef> {
    let field = schema.field(idx);
    match field.data_type() {
        DataType::List(elem) | DataType::LargeList(elem) => Ok(elem.clone()),
        other => Err(anyhow!(
            "column `{}` (idx {idx}) expected List, got {other:?}",
            field.name()
        )),
    }
}

fn append_commit_list(b: &mut ListBuilder<StructBuilder>, list: Option<&Vec<CommitRef>>) {
    match list {
        None => b.append(false),
        Some(v) => {
            let struct_b: &mut StructBuilder = b.values();
            for c in v {
                struct_b
                    .field_builder::<StringBuilder>(0)
                    .expect("repo builder")
                    .append_value(&c.repo);
                struct_b
                    .field_builder::<StringBuilder>(1)
                    .expect("sha builder")
                    .append_value(&c.sha);
                struct_b.append(true);
            }
            b.append(true);
        }
    }
}

// ---------- read path ----------

fn parse_batch(batch: &RecordBatch) -> Result<Vec<Event>> {
    let kinds = string_col(batch, COL_KIND, "kind")?;
    let ts_arr = batch
        .column(COL_TS_MICROS)
        .as_any()
        .downcast_ref::<TimestampMicrosecondArray>()
        .ok_or_else(|| anyhow!("ts_micros column not Timestamp(us)"))?;
    let idea_ids = string_opt_col(batch, COL_IDEA_ID);
    let plan_ids = string_opt_col(batch, COL_PLAN_ID);
    let node_ids = string_opt_col(batch, COL_NODE_ID);
    let run_ids = string_opt_col(batch, COL_RUN_ID);
    let from_nodes = string_opt_col(batch, COL_FROM_NODE);
    let to_nodes = string_opt_col(batch, COL_TO_NODE);
    let sources = string_opt_col(batch, COL_SOURCE);
    let texts = string_opt_col(batch, COL_TEXT);
    let decisions = string_opt_col(batch, COL_DECISION);
    let node_statuses = string_opt_col(batch, COL_NODE_STATUS);
    let plan_statuses = string_opt_col(batch, COL_PLAN_STATUS);
    let whys = string_opt_col(batch, COL_WHY);
    let summaries = string_opt_col(batch, COL_SUMMARY);
    let planners = string_opt_col(batch, COL_PLANNER);
    let node_kinds = string_opt_col(batch, COL_NODE_KIND);
    let prompt_excerpts = string_opt_col(batch, COL_PROMPT_EXCERPT);
    let params_jsons = string_opt_col(batch, COL_PARAMS_JSON);
    let ran_bys = string_opt_col(batch, COL_RAN_BY);
    let outcomes = string_opt_col(batch, COL_OUTCOME);
    let log_refs = string_opt_col(batch, COL_LOG_REF);
    // FI `item_kind` (field 28) is migration-safe: a pre-FI table won't carry
    // the column, so probe by column count and default missing/null to `idea`.
    let item_kinds = string_opt_col_if_present(batch, COL_ITEM_KIND);
    // funnel-prompt-planner keystone (fields 29-33) — same tolerant probe; a
    // pre-keystone table simply won't carry these, and every value defaults to
    // `None` (mode unset / unassigned / no lane transition recorded).
    let modes = string_opt_col_if_present(batch, COL_MODE);
    let members = string_opt_col_if_present(batch, COL_MEMBER);
    let roles = string_opt_col_if_present(batch, COL_ROLE);
    let from_lanes = string_opt_col_if_present(batch, COL_FROM_LANE);
    let to_lanes = string_opt_col_if_present(batch, COL_TO_LANE);

    let refs_lists = string_list_col(batch, COL_REFS)?;
    let targets_lists = string_list_col(batch, COL_TARGETS)?;
    let test_runs_lists = string_list_col(batch, COL_PRODUCED_TEST_RUNS)?;
    let commits_lists = commit_list_col(batch, COL_PRODUCED_COMMITS)?;

    let mut out: Vec<Event> = Vec::with_capacity(batch.num_rows());
    for i in 0..batch.num_rows() {
        let ts = micros_to_dt(ts_arr.value(i));
        let kind = kinds.value(i);
        let ev = match kind {
            KIND_IDEA_SUBMITTED => Event::IdeaSubmitted {
                id: IdeaId::new(idea_ids[i].clone().context("IdeaSubmitted needs idea_id")?),
                source: sources[i].clone().unwrap_or_default(),
                text: texts[i].clone().unwrap_or_default(),
                refs: refs_lists[i].clone().unwrap_or_default(),
                // Missing column / null → `idea` (legacy rows are all ideas).
                item_kind: item_kinds
                    .get(i)
                    .and_then(|o| o.as_deref())
                    .map(ItemKind::parse)
                    .unwrap_or_default(),
                ts,
            },
            KIND_IDEA_TRIAGED => Event::IdeaTriaged {
                idea_id: IdeaId::new(idea_ids[i].clone().context("IdeaTriaged needs idea_id")?),
                decision: parse_triage(decisions[i].as_deref().unwrap_or(""))?,
                why: whys[i].clone(),
                ts,
            },
            KIND_PLAN_CREATED => Event::PlanCreated {
                id: PlanId::new(plan_ids[i].clone().context("PlanCreated needs plan_id")?),
                idea_id: IdeaId::new(idea_ids[i].clone().context("PlanCreated needs idea_id")?),
                summary: summaries[i].clone().unwrap_or_default(),
                planner: planners[i].clone().unwrap_or_default(),
                ts,
            },
            KIND_NODE_ADDED => Event::NodeAdded {
                plan_id: PlanId::new(plan_ids[i].clone().context("NodeAdded needs plan_id")?),
                node_id: NodeId::new(node_ids[i].clone().context("NodeAdded needs node_id")?),
                kind: node_kinds[i].clone().unwrap_or_default(),
                params: match params_jsons[i].as_deref() {
                    Some(s) if !s.is_empty() => serde_json::from_str(s).unwrap_or_default(),
                    _ => serde_json::Map::new(),
                },
                targets: targets_lists[i].clone().unwrap_or_default(),
                prompt_excerpt: prompt_excerpts[i].clone(),
                ts,
            },
            KIND_EDGE_ADDED => Event::EdgeAdded {
                plan_id: PlanId::new(plan_ids[i].clone().context("EdgeAdded needs plan_id")?),
                from_node: NodeId::new(from_nodes[i].clone().context("EdgeAdded needs from_node")?),
                to_node: NodeId::new(to_nodes[i].clone().context("EdgeAdded needs to_node")?),
                ts,
            },
            KIND_NODE_STATUS_CHANGED => Event::NodeStatusChanged {
                plan_id: PlanId::new(plan_ids[i].clone().context("NodeStatusChanged needs plan_id")?),
                node_id: NodeId::new(node_ids[i].clone().context("NodeStatusChanged needs node_id")?),
                status: parse_node_status(node_statuses[i].as_deref().unwrap_or(""))?,
                why: whys[i].clone(),
                ts,
            },
            KIND_RUN_RECORDED => Event::RunRecorded {
                plan_id: PlanId::new(plan_ids[i].clone().context("RunRecorded needs plan_id")?),
                node_id: NodeId::new(node_ids[i].clone().context("RunRecorded needs node_id")?),
                run_id: RunId::new(run_ids[i].clone().context("RunRecorded needs run_id")?),
                ran_by: ran_bys[i].clone().unwrap_or_default(),
                outcome: parse_run_outcome(outcomes[i].as_deref().unwrap_or(""))?,
                log_ref: log_refs[i].clone(),
                produced_commits: commits_lists[i].clone().unwrap_or_default(),
                produced_test_runs: test_runs_lists[i].clone().unwrap_or_default(),
                ts,
            },
            KIND_PLAN_STATUS_CHANGED => Event::PlanStatusChanged {
                plan_id: PlanId::new(plan_ids[i].clone().context("PlanStatusChanged needs plan_id")?),
                status: parse_plan_status(plan_statuses[i].as_deref().unwrap_or(""))?,
                why: whys[i].clone(),
                ts,
            },
            KIND_IDEA_MODE_SET => Event::IdeaModeSet {
                idea_id: IdeaId::new(idea_ids[i].clone().context("IdeaModeSet needs idea_id")?),
                // Missing column / null on a legacy row can't happen (this kind
                // didn't exist pre-keystone) but default to "" rather than panic.
                mode: modes.get(i).and_then(|o| o.clone()).unwrap_or_default(),
                ts,
            },
            KIND_CARD_ASSIGNED => Event::CardAssigned {
                node: node_ids[i].clone().context("CardAssigned needs node")?,
                member: members.get(i).and_then(|o| o.clone()).unwrap_or_default(),
                role: roles.get(i).and_then(|o| o.clone()),
                why: whys[i].clone(),
                ts,
            },
            KIND_CARD_STATE_CHANGED => Event::CardStateChanged {
                node: node_ids[i].clone().context("CardStateChanged needs node")?,
                from_lane: from_lanes.get(i).and_then(|o| o.clone()),
                to_lane: to_lanes.get(i).and_then(|o| o.clone()).unwrap_or_default(),
                why: whys[i].clone(),
                ts,
            },
            other => return Err(anyhow!("unknown funnel event kind `{other}`")),
        };
        out.push(ev);
    }
    Ok(out)
}

fn string_col<'a>(batch: &'a RecordBatch, idx: usize, name: &str) -> Result<&'a StringArray> {
    batch
        .column(idx)
        .as_any()
        .downcast_ref::<StringArray>()
        .ok_or_else(|| anyhow!("column `{name}` (idx {idx}) is not StringArray"))
}

fn string_opt_col(batch: &RecordBatch, idx: usize) -> Vec<Option<String>> {
    let arr = batch
        .column(idx)
        .as_any()
        .downcast_ref::<StringArray>()
        .expect("optional string column");
    (0..arr.len())
        .map(|i| if arr.is_null(i) { None } else { Some(arr.value(i).to_string()) })
        .collect()
}

/// Like [`string_opt_col`] but tolerant of a column that doesn't exist yet
/// (a pre-FI `funnel_events` table lacks `item_kind`). Returns an all-`None`
/// vec sized to the batch when the column is absent / not a StringArray, so
/// callers index it positionally and default missing values.
fn string_opt_col_if_present(batch: &RecordBatch, idx: usize) -> Vec<Option<String>> {
    if idx >= batch.num_columns() {
        return vec![None; batch.num_rows()];
    }
    match batch.column(idx).as_any().downcast_ref::<StringArray>() {
        Some(arr) => (0..arr.len())
            .map(|i| if arr.is_null(i) { None } else { Some(arr.value(i).to_string()) })
            .collect(),
        None => vec![None; batch.num_rows()],
    }
}

fn string_list_col(batch: &RecordBatch, idx: usize) -> Result<Vec<Option<Vec<String>>>> {
    use arrow::array::ListArray;
    let arr = batch
        .column(idx)
        .as_any()
        .downcast_ref::<ListArray>()
        .ok_or_else(|| anyhow!("column {idx} is not ListArray"))?;
    let mut out = Vec::with_capacity(arr.len());
    for i in 0..arr.len() {
        if arr.is_null(i) {
            out.push(None);
            continue;
        }
        let values = arr.value(i);
        let strs = values
            .as_any()
            .downcast_ref::<StringArray>()
            .ok_or_else(|| anyhow!("list element not StringArray"))?;
        let mut v = Vec::with_capacity(strs.len());
        for j in 0..strs.len() {
            if !strs.is_null(j) {
                v.push(strs.value(j).to_string());
            }
        }
        out.push(Some(v));
    }
    Ok(out)
}

fn commit_list_col(batch: &RecordBatch, idx: usize) -> Result<Vec<Option<Vec<CommitRef>>>> {
    use arrow::array::{ListArray, StructArray};
    let arr = batch
        .column(idx)
        .as_any()
        .downcast_ref::<ListArray>()
        .ok_or_else(|| anyhow!("column {idx} (produced_commits) is not ListArray"))?;
    let mut out = Vec::with_capacity(arr.len());
    for i in 0..arr.len() {
        if arr.is_null(i) {
            out.push(None);
            continue;
        }
        let values = arr.value(i);
        let st = values
            .as_any()
            .downcast_ref::<StructArray>()
            .ok_or_else(|| anyhow!("produced_commits element not StructArray"))?;
        let repos = st
            .column_by_name("repo")
            .and_then(|c| c.as_any().downcast_ref::<StringArray>())
            .ok_or_else(|| anyhow!("commit struct missing repo"))?;
        let shas = st
            .column_by_name("sha")
            .and_then(|c| c.as_any().downcast_ref::<StringArray>())
            .ok_or_else(|| anyhow!("commit struct missing sha"))?;
        let mut v = Vec::with_capacity(st.len());
        for j in 0..st.len() {
            v.push(CommitRef {
                repo: repos.value(j).to_string(),
                sha: shas.value(j).to_string(),
            });
        }
        out.push(Some(v));
    }
    Ok(out)
}

// ---------- enum string helpers ----------

fn event_kind_str(ev: &Event) -> &'static str {
    match ev {
        Event::IdeaSubmitted { .. } => KIND_IDEA_SUBMITTED,
        Event::IdeaTriaged { .. } => KIND_IDEA_TRIAGED,
        Event::PlanCreated { .. } => KIND_PLAN_CREATED,
        Event::NodeAdded { .. } => KIND_NODE_ADDED,
        Event::EdgeAdded { .. } => KIND_EDGE_ADDED,
        Event::NodeStatusChanged { .. } => KIND_NODE_STATUS_CHANGED,
        Event::RunRecorded { .. } => KIND_RUN_RECORDED,
        Event::PlanStatusChanged { .. } => KIND_PLAN_STATUS_CHANGED,
        Event::IdeaModeSet { .. } => KIND_IDEA_MODE_SET,
        Event::CardAssigned { .. } => KIND_CARD_ASSIGNED,
        Event::CardStateChanged { .. } => KIND_CARD_STATE_CHANGED,
    }
}

fn triage_str(d: TriageDecision) -> &'static str {
    match d {
        TriageDecision::Accept => "accept",
        TriageDecision::Drop => "drop",
    }
}

fn parse_triage(s: &str) -> Result<TriageDecision> {
    match s {
        "accept" => Ok(TriageDecision::Accept),
        "drop" => Ok(TriageDecision::Drop),
        other => Err(anyhow!("bad triage decision `{other}`")),
    }
}

fn node_status_str(s: NodeStatus) -> &'static str {
    match s {
        NodeStatus::Pending => "pending",
        NodeStatus::Ready => "ready",
        NodeStatus::InProgress => "in_progress",
        NodeStatus::Done => "done",
        NodeStatus::Blocked => "blocked",
        NodeStatus::Failed => "failed",
    }
}

fn parse_node_status(s: &str) -> Result<NodeStatus> {
    match s {
        "pending" => Ok(NodeStatus::Pending),
        "ready" => Ok(NodeStatus::Ready),
        "in_progress" => Ok(NodeStatus::InProgress),
        "done" => Ok(NodeStatus::Done),
        "blocked" => Ok(NodeStatus::Blocked),
        "failed" => Ok(NodeStatus::Failed),
        other => Err(anyhow!("bad node status `{other}`")),
    }
}

fn plan_status_str(s: PlanStatus) -> &'static str {
    match s {
        PlanStatus::Draft => "draft",
        PlanStatus::Active => "active",
        PlanStatus::Done => "done",
        PlanStatus::Abandoned => "abandoned",
    }
}

fn parse_plan_status(s: &str) -> Result<PlanStatus> {
    match s {
        "draft" => Ok(PlanStatus::Draft),
        "active" => Ok(PlanStatus::Active),
        "done" => Ok(PlanStatus::Done),
        "abandoned" => Ok(PlanStatus::Abandoned),
        other => Err(anyhow!("bad plan status `{other}`")),
    }
}

fn run_outcome_str(o: RunOutcome) -> &'static str {
    match o {
        RunOutcome::Ok => "ok",
        RunOutcome::Failed => "failed",
        RunOutcome::Aborted => "aborted",
    }
}

fn parse_run_outcome(s: &str) -> Result<RunOutcome> {
    match s {
        "ok" => Ok(RunOutcome::Ok),
        "failed" => Ok(RunOutcome::Failed),
        "aborted" => Ok(RunOutcome::Aborted),
        other => Err(anyhow!("bad run outcome `{other}`")),
    }
}

fn micros_to_dt(micros: i64) -> DateTime<Utc> {
    let secs = micros.div_euclid(1_000_000);
    let nanos = (micros.rem_euclid(1_000_000) * 1_000) as u32;
    Utc.timestamp_opt(secs, nanos).single().unwrap_or_else(Utc::now)
}

#[cfg(test)]
mod tests {
    use super::*;
    use chrono::TimeZone;

    /// Differential: the Ragnar-fast `events_as_of` returns exactly the slice a
    /// linear `filter(ts <= T)` would, across the whole query space — including
    /// duplicate timestamps and out-of-range queries.
    #[test]
    fn events_as_of_matches_linear_filter() {
        let base = Utc.with_ymd_and_hms(2026, 1, 1, 0, 0, 0).unwrap();
        // Build a ts-sorted log; several events share a timestamp (dup keys).
        let mut events: Vec<Event> = Vec::new();
        let secs = [0i64, 0, 1, 5, 5, 5, 8, 13, 21, 21, 34];
        for (i, &s) in secs.iter().enumerate() {
            events.push(Event::IdeaSubmitted {
                id: IdeaId::seq(i as u64 + 1),
                source: "t".into(),
                text: format!("idea {i}"),
                refs: vec![],
                item_kind: ItemKind::Idea,
                ts: base + chrono::Duration::seconds(s),
            });
        }
        // Already ts-sorted by construction.
        let idx = FunnelLogIndex::build(&events);

        for probe in -5..=40 {
            let t = (base + chrono::Duration::seconds(probe)).timestamp_micros();
            let fast = events_as_of(&events, t);
            let linear: Vec<&Event> =
                events.iter().filter(|e| e.ts().timestamp_micros() <= t).collect();
            assert_eq!(fast.len(), linear.len(), "cut differs at probe {probe}");
            assert_eq!(idx.cut_as_of(t), linear.len(), "index cut differs at {probe}");
            for (a, b) in fast.iter().zip(linear.iter()) {
                assert_eq!(a.ts(), b.ts());
            }
        }
    }

    #[test]
    fn roundtrip_all_event_kinds() {
        let dir = tempfile::tempdir().unwrap();
        let wh = IcebergWarehouse::open(dir.path()).unwrap();

        let t = Utc.with_ymd_and_hms(2026, 5, 31, 12, 0, 0).unwrap();
        let events = vec![
            Event::IdeaSubmitted {
                id: IdeaId::seq(1),
                source: "test".into(),
                text: "first idea".into(),
                refs: vec!["a".into(), "b".into()],
                item_kind: ItemKind::Idea,
                ts: t,
            },
            // FI: an error item must round-trip with item_kind = error.
            Event::IdeaSubmitted {
                id: IdeaId::seq(2),
                source: "test".into(),
                text: "panic in foo.rs".into(),
                refs: vec![],
                item_kind: ItemKind::Error,
                ts: t,
            },
            Event::PlanCreated {
                id: PlanId::seq(1),
                idea_id: IdeaId::seq(1),
                summary: "the plan".into(),
                planner: "tester".into(),
                ts: t,
            },
            Event::NodeAdded {
                plan_id: PlanId::seq(1),
                node_id: NodeId::seq(1),
                kind: "rust:impl".into(),
                params: serde_json::Map::new(),
                targets: vec!["src/foo.rs".into()],
                prompt_excerpt: Some("do the thing".into()),
                ts: t,
            },
            Event::EdgeAdded {
                plan_id: PlanId::seq(1),
                from_node: NodeId::seq(1),
                to_node: NodeId::seq(2),
                ts: t,
            },
            Event::NodeStatusChanged {
                plan_id: PlanId::seq(1),
                node_id: NodeId::seq(1),
                status: NodeStatus::Done,
                why: Some("done".into()),
                ts: t,
            },
            Event::RunRecorded {
                plan_id: PlanId::seq(1),
                node_id: NodeId::seq(1),
                run_id: RunId::seq(1),
                ran_by: "tester".into(),
                outcome: RunOutcome::Ok,
                log_ref: None,
                produced_commits: vec![CommitRef { repo: "nornir".into(), sha: "abc123".into() }],
                produced_test_runs: vec!["smoke-1".into()],
                ts: t,
            },
            Event::PlanStatusChanged {
                plan_id: PlanId::seq(1),
                status: PlanStatus::Active,
                why: None,
                ts: t,
            },
            // funnel-prompt-planner keystone: the three new event kinds.
            Event::IdeaModeSet { idea_id: IdeaId::seq(1), mode: "auto".into(), ts: t },
            Event::CardAssigned {
                node: "n-001".into(),
                member: "ada".into(),
                role: Some("coder".into()),
                why: Some("auto-assign: code:write -> Coder".into()),
                ts: t,
            },
            Event::CardStateChanged {
                node: "n-001".into(),
                from_lane: None,
                to_lane: "proposed".into(),
                why: Some("manual prompt decompose".into()),
                ts: t,
            },
        ];

        wh.block_on(async {
            for ev in &events {
                append_event(&wh, ev).await.unwrap();
            }
        });

        let loaded = wh.block_on(async { load_all_events(&wh).await.unwrap() });
        assert_eq!(loaded.len(), events.len(), "round-trip count");
        // Spot-check the IdeaSubmitted refs and RunRecorded commits made it.
        let mut saw_refs = false;
        let mut saw_commits = false;
        let mut saw_error_kind = false;
        let mut saw_idea_kind = false;
        let mut saw_mode = false;
        let mut saw_assigned = false;
        let mut saw_state_changed = false;
        for ev in &loaded {
            if let Event::IdeaSubmitted { refs, item_kind, id, .. } = ev {
                // FI: the item_kind round-trips per item.
                match (id.as_str(), item_kind) {
                    ("i-001", ItemKind::Idea) => {
                        assert_eq!(refs.len(), 2);
                        saw_refs = true;
                        saw_idea_kind = true;
                    }
                    ("i-002", ItemKind::Error) => saw_error_kind = true,
                    other => panic!("unexpected idea kind: {other:?}"),
                }
            }
            if let Event::RunRecorded { produced_commits, .. } = ev {
                assert_eq!(produced_commits.len(), 1);
                assert_eq!(produced_commits[0].repo, "nornir");
                saw_commits = true;
            }
            if let Event::IdeaModeSet { idea_id, mode, .. } = ev {
                assert_eq!(idea_id.as_str(), "i-001");
                assert_eq!(mode, "auto");
                saw_mode = true;
            }
            if let Event::CardAssigned { node, member, role, why, .. } = ev {
                assert_eq!(node, "n-001");
                assert_eq!(member, "ada");
                assert_eq!(role.as_deref(), Some("coder"));
                assert!(why.is_some());
                saw_assigned = true;
            }
            if let Event::CardStateChanged { node, from_lane, to_lane, .. } = ev {
                assert_eq!(node, "n-001");
                assert_eq!(*from_lane, None);
                assert_eq!(to_lane, "proposed");
                saw_state_changed = true;
            }
        }
        assert!(saw_mode && saw_assigned && saw_state_changed, "the 3 new event kinds round-trip");
        assert!(saw_refs && saw_commits, "refs and commits should round-trip");
        assert!(saw_idea_kind && saw_error_kind, "both item_kinds round-trip");
    }

    /// REGRESSION/migration: the oldest funnel_events shape — a 27-column table
    /// that predates BOTH the FI `item_kind` column (28) AND the 5 keystone
    /// columns (mode/member/role/from_lane/to_lane, 29-33) — must still evolve
    /// all the way to the canonical schema on the first write, and the written
    /// kind must read back. `append_event` calls `ensure_table_schema`.
    /// Reproduce the 27-col table, append an error item, assert it evolved.
    #[test]
    fn stale_27col_funnel_table_evolves_and_kind_round_trips() {
        use iceberg::Catalog;
        use iceberg::spec::Schema;
        use iceberg::TableCreation;

        let dir = tempfile::tempdir().unwrap();
        let wh = IcebergWarehouse::open(dir.path()).unwrap();
        let ident = wh.table_ident(TABLE_FUNNEL_EVENTS);

        // Recreate the table at the canonical schema MINUS item_kind (28) and
        // the 5 keystone columns (29-33) — the 27-column pre-everything shape
        // the oldest binary created.
        wh.block_on(async {
            let cat = wh.catalog();
            cat.drop_table(&ident).await.unwrap();
            let full = crate::warehouse::iceberg_schema::funnel_events().unwrap();
            let pre_fi_cols = ["item_kind", "mode", "member", "role", "from_lane", "to_lane"];
            let fields: Vec<_> = full
                .as_struct()
                .fields()
                .iter()
                .filter(|f| !pre_fi_cols.contains(&f.name.as_str()))
                .cloned()
                .collect();
            let stale = Schema::builder().with_schema_id(0).with_fields(fields).build().unwrap();
            assert_eq!(stale.as_struct().fields().len(), 27);
            let creation =
                TableCreation::builder().name(ident.name().to_string()).schema(stale).build();
            cat.create_table(ident.namespace(), creation).await.unwrap();
        });

        // Append an error item — the bug class: canonical cols vs stale fields.
        let ev = Event::IdeaSubmitted {
            id: IdeaId::seq(1),
            source: "test".into(),
            text: "stale-table error report".into(),
            refs: vec![],
            item_kind: ItemKind::Error,
            ts: Utc::now(),
        };
        wh.block_on(async { append_event(&wh, &ev).await.unwrap() });

        // Table evolved to the full canonical schema (33 cols) including item_kind.
        wh.block_on(async {
            let t = wh.catalog().load_table(&ident).await.unwrap();
            let names: Vec<&str> =
                t.metadata().current_schema().as_struct().fields().iter().map(|f| f.name.as_str()).collect();
            assert_eq!(names.len(), 33);
            assert!(names.contains(&"item_kind"));
        });

        // The kind reads back as error.
        let loaded = wh.block_on(async { load_all_events(&wh).await.unwrap() });
        assert_eq!(loaded.len(), 1);
        match &loaded[0] {
            Event::IdeaSubmitted { item_kind, .. } => assert_eq!(*item_kind, ItemKind::Error),
            other => panic!("expected IdeaSubmitted, got {other:?}"),
        }
    }

    /// REGRESSION/migration (funnel-prompt-planner keystone): a `funnel_events`
    /// table that predates the 5 new columns (mode/member/role/from_lane/
    /// to_lane, fields 29-33 — i.e. a pre-keystone 28-col table, exactly what
    /// `stale_27col_funnel_table_evolves_and_kind_round_trips` proves a real
    /// FI-era binary created) must evolve to 33 cols on the first write that
    /// needs them, AND every legacy row (no value in those columns) must keep
    /// replaying clean — never an error, never a wrong default. Reproduce the
    /// 28-col table, append a legacy-shaped `IdeaSubmitted` PLUS one of each new
    /// event kind, assert the table evolved and everything reads back right.
    #[test]
    fn stale_28col_funnel_table_evolves_and_new_events_round_trip() {
        use iceberg::Catalog;
        use iceberg::spec::Schema;
        use iceberg::TableCreation;

        let dir = tempfile::tempdir().unwrap();
        let wh = IcebergWarehouse::open(dir.path()).unwrap();
        let ident = wh.table_ident(TABLE_FUNNEL_EVENTS);

        // Recreate the table at the canonical schema MINUS the 5 keystone
        // columns — the exact 28-column shape `item_kind`-era nornir created.
        wh.block_on(async {
            let cat = wh.catalog();
            cat.drop_table(&ident).await.unwrap();
            let full = crate::warehouse::iceberg_schema::funnel_events().unwrap();
            let new_cols = ["mode", "member", "role", "from_lane", "to_lane"];
            let fields: Vec<_> = full
                .as_struct()
                .fields()
                .iter()
                .filter(|f| !new_cols.contains(&f.name.as_str()))
                .cloned()
                .collect();
            let stale = Schema::builder().with_schema_id(0).with_fields(fields).build().unwrap();
            assert_eq!(stale.as_struct().fields().len(), 28);
            let creation =
                TableCreation::builder().name(ident.name().to_string()).schema(stale).build();
            cat.create_table(ident.namespace(), creation).await.unwrap();
        });

        // First: a legacy-shaped write (an ordinary idea, nothing keystone-ish)
        // — must append cleanly against the stale 28-col table with NO evolution
        // yet triggered (append_event always evolves-to-canonical before every
        // write, so this also just proves that path doesn't choke on 28→33).
        let legacy = Event::IdeaSubmitted {
            id: IdeaId::seq(1),
            source: "test".into(),
            text: "a plain pre-keystone idea".into(),
            refs: vec![],
            item_kind: ItemKind::Idea,
            ts: Utc::now(),
        };
        wh.block_on(async { append_event(&wh, &legacy).await.unwrap() });

        // Then the 3 new event kinds — the actual keystone writes.
        let mode_ev = Event::IdeaModeSet { idea_id: IdeaId::seq(1), mode: "manual".into(), ts: Utc::now() };
        let assign_ev = Event::CardAssigned {
            node: "n-001".into(),
            member: "ada".into(),
            role: Some("tester".into()),
            why: None,
            ts: Utc::now(),
        };
        let state_ev = Event::CardStateChanged {
            node: "n-001".into(),
            from_lane: Some("inbox".into()),
            to_lane: "proposed".into(),
            why: Some("manual mode draft".into()),
            ts: Utc::now(),
        };
        wh.block_on(async {
            append_event(&wh, &mode_ev).await.unwrap();
            append_event(&wh, &assign_ev).await.unwrap();
            append_event(&wh, &state_ev).await.unwrap();
        });

        // Table evolved to 33 cols including all 5 new columns.
        wh.block_on(async {
            let t = wh.catalog().load_table(&ident).await.unwrap();
            let names: Vec<&str> =
                t.metadata().current_schema().as_struct().fields().iter().map(|f| f.name.as_str()).collect();
            assert_eq!(names.len(), 33);
            for c in ["mode", "member", "role", "from_lane", "to_lane"] {
                assert!(names.contains(&c), "evolved table must carry `{c}`");
            }
        });

        // Everything reads back: the legacy row untouched (no mode set on it —
        // proving old rows don't spuriously pick up a value from the new
        // columns), and the 3 new events with their fields intact.
        let loaded = wh.block_on(async { load_all_events(&wh).await.unwrap() });
        assert_eq!(loaded.len(), 4);

        let mut saw_legacy_untouched = false;
        let mut saw_mode = false;
        let mut saw_assigned = false;
        let mut saw_state = false;
        for ev in &loaded {
            match ev {
                Event::IdeaSubmitted { id, .. } if id.as_str() == "i-001" => {
                    saw_legacy_untouched = true;
                }
                Event::IdeaModeSet { idea_id, mode, .. } => {
                    assert_eq!(idea_id.as_str(), "i-001");
                    assert_eq!(mode, "manual");
                    saw_mode = true;
                }
                Event::CardAssigned { node, member, role, .. } => {
                    assert_eq!(node, "n-001");
                    assert_eq!(member, "ada");
                    assert_eq!(role.as_deref(), Some("tester"));
                    saw_assigned = true;
                }
                Event::CardStateChanged { node, from_lane, to_lane, .. } => {
                    assert_eq!(node, "n-001");
                    assert_eq!(from_lane.as_deref(), Some("inbox"));
                    assert_eq!(to_lane, "proposed");
                    saw_state = true;
                }
                other => panic!("unexpected event: {other:?}"),
            }
        }
        assert!(
            saw_legacy_untouched && saw_mode && saw_assigned && saw_state,
            "the pre-keystone row survives the evolution and all 3 new event kinds round-trip"
        );
    }
}