src/warehouse/funnel.rs
//! 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::{Context, Result, anyhow};
use arrow::array::{
Array, ArrayBuilder, Float64Array, 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;
// Plan-DAG ↔ Component-DAG wiring (`PlanComponentWired`): four more OPTIONAL
// columns, added the same migration-safe way. `plan_node` reuses `COL_NODE_ID`
// (a generic card id) and the note reuses `COL_WHY`.
const COL_COMPONENT: usize = 33;
const COL_WIRE_ORIGIN: usize = 34;
const COL_WIRE_STATUS: usize = 35;
const COL_WIRE_CONFIDENCE: usize = 36;
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";
const KIND_PLAN_COMPONENT_WIRED: &str = "PlanComponentWired";
/// Last-resort label for a `#[non_exhaustive]` variant whose serde tag could not
/// even be read. See [`event_kind`].
const KIND_UNKNOWN: &str = "UnknownFunnelEvent";
/// 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
}
/// Append MANY events in **one** Iceberg snapshot (one commit → one fsync),
/// instead of one snapshot per event. The row bytes are identical to calling
/// [`append_event`] once per event — only the snapshot count differs — so a
/// readback is byte-identical (see the roundtrip test). Amortizes the per-append
/// snapshot-commit cost that dominates bulk writes like `store_subtasks`
/// (mother `PlanCreated` + N `NodeAdded` + M `EdgeAdded`). No-op on an empty
/// slice (never cuts an empty snapshot).
pub async fn append_events(wh: &IcebergWarehouse, events: &[Event]) -> Result<()> {
if events.is_empty() {
return Ok(());
}
let ident = wh.table_ident(TABLE_FUNNEL_EVENTS);
let table = wh.catalog().load_table(&ident).await?;
let table = ensure_table_schema(
wh.catalog(),
&ident,
table,
&super::iceberg_schema::funnel_events()?,
)
.await?;
let batch = build_batch(&table, events)?;
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());
let mut components: Vec<Option<String>> = Vec::with_capacity(events.len());
let mut wire_origins: Vec<Option<String>> = Vec::with_capacity(events.len());
let mut wire_statuses: Vec<Option<String>> = Vec::with_capacity(events.len());
let mut wire_confidences: Vec<Option<f64>> = 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(ev));
// 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 component: Option<String> = None;
let mut wire_origin: Option<String> = None;
let mut wire_status: Option<String> = None;
let mut wire_confidence: Option<f64> = 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();
}
// Plan-DAG node ↔ Component-DAG component. The log stays append-only
// (last-write-wins per `(plan_node, component)`, retracted by a later
// `status = "rejected"` row), so this persists exactly like its
// siblings: the card id in the generic `node_id` column, the note in
// `why`, and the wire's own four fields in columns 34-37.
Event::PlanComponentWired {
plan_node,
component: c,
origin,
status,
confidence,
why: w,
..
} => {
node_id = Some(plan_node.clone());
component = Some(c.clone());
wire_origin = Some(origin.clone());
wire_status = Some(status.clone());
wire_confidence = *confidence;
why = w.clone();
}
// `funnel_core::Event` is `#[non_exhaustive]`, so a newer funnel-core
// can hand us a variant this binary was never compiled against. The
// envelope (`event_id`, `ts_micros`, `kind`) is still written by the
// code above — `event_kind` recovers the true kind from the serde tag,
// counts it in `UNKNOWN_FUNNEL_EVENTS` and warns — so the row is
// never LOST, only under-populated. Every per-field column stays NULL
// because nornir has no schema for fields it has never seen.
//
// This arm exists ONLY for variants that do not yet exist: every
// variant nornir knows is named above, and adding a named arm here is
// the fix the warning asks for.
_ => {}
}
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);
components.push(component);
wire_origins.push(wire_origin);
wire_statuses.push(wire_status);
wire_confidences.push(wire_confidence);
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));
}
// PlanComponentWired (fields 34-37) — same migration-safety guard. NOTE the
// confidence column is a Double, so the StringArray placeholder MUST be
// replaced whenever the table carries the field, or `RecordBatch::try_new`
// would reject the type mismatch.
if n_fields > COL_COMPONENT {
cols[COL_COMPONENT] = Arc::new(StringArray::from(components));
}
if n_fields > COL_WIRE_ORIGIN {
cols[COL_WIRE_ORIGIN] = Arc::new(StringArray::from(wire_origins));
}
if n_fields > COL_WIRE_STATUS {
cols[COL_WIRE_STATUS] = Arc::new(StringArray::from(wire_statuses));
}
if n_fields > COL_WIRE_CONFIDENCE {
cols[COL_WIRE_CONFIDENCE] = Arc::new(Float64Array::from(wire_confidences));
}
// 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 components = string_opt_col_if_present(batch, COL_COMPONENT);
let wire_origins = string_opt_col_if_present(batch, COL_WIRE_ORIGIN);
let wire_statuses = string_opt_col_if_present(batch, COL_WIRE_STATUS);
let wire_confidences = f64_opt_col_if_present(batch, COL_WIRE_CONFIDENCE);
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,
},
KIND_PLAN_COMPONENT_WIRED => Event::PlanComponentWired {
plan_node: node_ids[i]
.clone()
.context("PlanComponentWired needs plan_node")?,
component: components
.get(i)
.and_then(|o| o.clone())
.context("PlanComponentWired needs component")?,
// `origin`/`status` are free-form wire tokens read back typed via
// `Wire::{origin_kind,status_kind}` (which fail closed), so an
// empty cell round-trips as the empty token rather than an error.
origin: wire_origins
.get(i)
.and_then(|o| o.clone())
.unwrap_or_default(),
status: wire_statuses
.get(i)
.and_then(|o| o.clone())
.unwrap_or_default(),
confidence: wire_confidences.get(i).copied().flatten(),
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()],
}
}
/// `string_opt_col_if_present`'s Double twin, for `wire_confidence` (field 37).
/// A pre-wiring table simply doesn't carry the column ⇒ all-`None`.
fn f64_opt_col_if_present(batch: &RecordBatch, idx: usize) -> Vec<Option<f64>> {
if idx >= batch.num_columns() {
return vec![None; batch.num_rows()];
}
match batch.column(idx).as_any().downcast_ref::<Float64Array>() {
Some(arr) => (0..arr.len())
.map(|i| {
if arr.is_null(i) {
None
} else {
Some(arr.value(i))
}
})
.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 ----------
/// How many events this binary could not name — see [`event_kind`]. Public so a
/// health/introspection surface can read it: a non-zero count means the warehouse
/// is receiving funnel events from a NEWER funnel-core than this nornir knows, and
/// those rows carry only their envelope. Never resets; monotonic per process.
pub static UNKNOWN_FUNNEL_EVENTS: std::sync::atomic::AtomicU64 =
std::sync::atomic::AtomicU64::new(0);
/// The `kind` string a row is written under.
///
/// `funnel_core::Event` is `#[non_exhaustive]`, so a newer funnel-core can ship a
/// variant this binary has never compiled against. Rather than swallow it, the
/// wildcard is OBSERVABLE: it bumps [`UNKNOWN_FUNNEL_EVENTS`], warns once per
/// occurrence, and recovers the real kind from the event's own serde tag
/// (`#[serde(tag = "kind")]`) so the row is at least labelled TRUTHFULLY — a
/// later, wiser binary can then decode it. Only the envelope survives; the
/// variant's own fields have no columns here yet, which is precisely what the
/// warning is telling you to fix.
fn event_kind(ev: &Event) -> String {
if let Some(known) = known_event_kind(ev) {
return known.to_string();
}
let tag = serde_json::to_value(ev)
.ok()
.and_then(|v| v.get("kind").and_then(|k| k.as_str()).map(str::to_string))
.unwrap_or_else(|| KIND_UNKNOWN.to_string());
UNKNOWN_FUNNEL_EVENTS.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
eprintln!(
"nornir warehouse: funnel event `{tag}` is not known to this binary — the row is \
written with its envelope only (no per-field columns). Teach `warehouse::funnel` \
about it (funnel-core added a variant)."
);
tag
}
/// The kind of an event this binary DOES know, or `None` for a variant added by a
/// newer funnel-core. Every named arm here has a matching encoder arm above and a
/// decoder arm in `batch_to_events`.
fn known_event_kind(ev: &Event) -> Option<&'static str> {
Some(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,
Event::PlanComponentWired { .. } => KIND_PLAN_COMPONENT_WIRED,
// `Event` is `#[non_exhaustive]`: a newer funnel-core variant lands here,
// NOT in one of the named arms above. Reported (not swallowed) by the
// caller — see [`event_kind`].
_ => return None,
})
}
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 append_events_batch_is_byte_identical_to_single_appends() {
// The fsync-amortizing `append_events` (N events, ONE snapshot) must read
// back IDENTICALLY to N single `append_event` calls (N snapshots) — only
// the snapshot count differs, never the row bytes.
let t = Utc.with_ymd_and_hms(2026, 6, 1, 9, 0, 0).unwrap();
let events: Vec<Event> = (0..6)
.map(|i| Event::IdeaSubmitted {
id: IdeaId::seq(i + 1),
source: "batch-test".into(),
text: format!("idea {i}"),
refs: if i % 2 == 0 {
vec![]
} else {
vec!["r1".into(), "r2".into()]
},
item_kind: if i % 3 == 0 {
ItemKind::Error
} else {
ItemKind::Idea
},
ts: t + chrono::Duration::microseconds(i as i64),
})
.collect();
let dir_single = tempfile::tempdir().unwrap();
let wh_single = IcebergWarehouse::open(dir_single.path()).unwrap();
let single = wh_single.block_on(async {
for ev in &events {
append_event(&wh_single, ev).await.unwrap();
}
load_all_events(&wh_single).await.unwrap()
});
let dir_batch = tempfile::tempdir().unwrap();
let wh_batch = IcebergWarehouse::open(dir_batch.path()).unwrap();
let batch = wh_batch.block_on(async {
append_events(&wh_batch, &events).await.unwrap();
load_all_events(&wh_batch).await.unwrap()
});
assert_eq!(single.len(), events.len(), "single-append round-trip count");
// Event has no PartialEq; compare stable Debug reprs (byte-identical content).
assert_eq!(
format!("{batch:?}"),
format!("{single:?}"),
"batched append must read back identical to single appends"
);
// Empty batch is a clean no-op (never cuts an empty snapshot).
let empty = wh_batch.block_on(async {
append_events(&wh_batch, &[]).await.unwrap();
load_all_events(&wh_batch).await.unwrap()
});
assert_eq!(
format!("{empty:?}"),
format!("{single:?}"),
"empty append_events is a no-op"
);
}
#[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::TableCreation;
use iceberg::spec::Schema;
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",
// …and the 4 PlanComponentWired columns (34-37), which that
// 27-column binary equally never knew about.
"component",
"wire_origin",
"wire_status",
"wire_confidence",
];
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 (37 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(), 37);
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::TableCreation;
use iceberg::spec::Schema;
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",
"component",
"wire_origin",
"wire_status",
"wire_confidence",
];
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(),
};
// …and the Plan-DAG ↔ Component-DAG wire, whose 4 columns (34-37) this
// stale table is equally missing.
let wire_ev = Event::PlanComponentWired {
plan_node: "n-001".into(),
component: "funnel-core".into(),
origin: "inferred".into(),
status: "proposed".into(),
confidence: Some(0.75),
why: Some("mimir saw the crate in the node targets".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();
append_event(&wh, &wire_ev).await.unwrap();
});
// Table evolved to 37 cols including all 5 keystone + 4 wiring 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(), 37);
for c in [
"mode",
"member",
"role",
"from_lane",
"to_lane",
"component",
"wire_origin",
"wire_status",
"wire_confidence",
] {
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(), 5);
let mut saw_legacy_untouched = false;
let mut saw_mode = false;
let mut saw_assigned = false;
let mut saw_state = false;
let mut saw_wire = 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;
}
Event::PlanComponentWired {
plan_node,
component,
origin,
status,
confidence,
why,
..
} => {
assert_eq!(plan_node, "n-001");
assert_eq!(component, "funnel-core");
assert_eq!(origin, "inferred");
assert_eq!(status, "proposed");
assert_eq!(*confidence, Some(0.75));
assert_eq!(
why.as_deref(),
Some("mimir saw the crate in the node targets")
);
saw_wire = true;
}
other => panic!("unexpected event: {other:?}"),
}
}
assert!(
saw_legacy_untouched && saw_mode && saw_assigned && saw_state && saw_wire,
"the pre-keystone row survives the evolution and all 4 new event kinds round-trip"
);
}
}