khive_db/stores/

event.rs

//! SQL-backed `EventStore` implementation.
//!
//! FILE SIZE JUSTIFICATION: Event store covers append, query-by-filter,
//! observation recording, and paginated listing with shared row-mapping and
//! timestamp serialization helpers. The event schema has complex JSON data
//! columns (observations, referent kinds, outcomes) whose parsing is shared
//! across all read paths, making a split impractical without duplicating the
//! deserialization logic.

use std::sync::Arc;

use async_trait::async_trait;
use uuid::Uuid;

use khive_storage::error::StorageError;
#[cfg(test)]
use khive_storage::error::WriterTaskRequestState;
use khive_storage::event::{
    Event, EventAppendDisposition, EventFilter, EventObservation, IdempotentEventBatchResult,
    ObservationRole, ReferentKind,
};
use khive_storage::types::{BatchWriteSummary, Page, PageRequest, SqlStatement, SqlValue};
use khive_storage::EventStore;
use khive_storage::SqlWriter;
use khive_storage::StorageCapability;
use khive_types::{EventKind, EventOutcome, SubstrateKind};

use crate::pool::ConnectionPool;
use crate::writer_task::WriterTaskHandle;

#[path = "event_cursor.rs"]
mod cursor;

fn map_err(e: rusqlite::Error, op: &'static str) -> StorageError {
    StorageError::driver(StorageCapability::Events, op, e)
}

// Preserve the existing error-classification fixtures at their original seam.
#[cfg(test)]
fn mark_unknown_append_usage(error: &StorageError) {
    khive_storage::usage::account_event_write(Err(error));
}

/// An EventStore backed by SQLite tables.
pub struct SqlEventStore {
    pool: Arc<ConnectionPool>,
    is_file_backed: bool,
    namespace: String,
    writer_task: Option<WriterTaskHandle>,
    #[cfg(test)]
    lose_next_fallback_reply: std::sync::atomic::AtomicBool,
}

impl SqlEventStore {
    /// Create a new store scoped to one namespace.
    pub fn new_scoped(
        pool: Arc<ConnectionPool>,
        is_file_backed: bool,
        namespace: impl Into<String>,
    ) -> Self {
        // Enabled by default for file-backed pools; explicit off/degraded
        // construction remains synchronous (ADR-067 Component A, mirrors
        // entity.rs policy): a missing writer task is cached without failing
        // construction. Every write re-resolves it and applies
        // strict/compatibility policy then.
        let writer_task = pool.writer_task_handle().ok().flatten();
        Self {
            pool,
            is_file_backed,
            namespace: namespace.into(),
            writer_task,
            #[cfg(test)]
            lose_next_fallback_reply: std::sync::atomic::AtomicBool::new(false),
        }
    }

    #[cfg(test)]
    fn after_fallback_write_for_test<T>(
        &self,
        result: Result<T, StorageError>,
    ) -> Result<T, StorageError> {
        // Lose only the reply: the real fallback transaction already committed.
        if result.is_ok()
            && self
                .lose_next_fallback_reply
                .swap(false, std::sync::atomic::Ordering::SeqCst)
        {
            return Err(StorageError::writer_task_terminated(
                WriterTaskRequestState::SideEffectsUnknown,
            ));
        }
        result
    }

    fn current_writer_task(
        &self,
        operation: &'static str,
    ) -> Result<Option<WriterTaskHandle>, StorageError> {
        self.pool
            .writer_task_for_write(self.writer_task.as_ref(), operation)
    }

    /// Route a single-row write through the pool-wide `WriterTask` when
    /// the write queue is enabled and a handle is available. Strict mode
    /// refuses a missing handle; compatibility mode falls back to the legacy
    /// standalone-connection / pool-mutex path (ADR-067 Component A, Fork C
    /// slice 2).
    ///
    /// `append_event`/`append_events` perform the same write-time lookup
    /// first; a non-strict `None` then falls through this helper, which
    /// records the actual compatibility fallback. Strict mode returns before
    /// the direct-writer seam.
    /// `f` must be DML-only on the flag-on path (no bare `BEGIN IMMEDIATE`)
    /// since it runs inside the WriterTask's own transaction.
    async fn with_writer<F, R>(&self, op: &'static str, f: F) -> Result<R, StorageError>
    where
        F: FnOnce(&rusqlite::Connection) -> Result<R, rusqlite::Error> + Send + 'static,
        R: Send + 'static,
    {
        if let Some(writer_task) = self.current_writer_task(op)? {
            return writer_task
                .send_bounded(move |conn| f(conn).map_err(|e| map_err(e, op)))
                .await;
        }

        self.pool
            .record_direct_route(crate::timeout_sink::Site::DirectRouteEventGeneralWrite);
        // Atomic SQL units release an in-memory connection between statements;
        // share their unit budget so this transaction cannot overlap one.
        let unit_slot = if self.is_file_backed {
            None
        } else {
            Some(crate::sql_bridge::acquire_in_memory_write_unit(&self.pool, op).await?)
        };
        let pool = Arc::clone(&self.pool);
        let db = crate::timeout_sink::db_label(&pool);
        let is_file_backed = self.is_file_backed;
        tokio::task::spawn_blocking(move || {
            // Cancellation of the caller must not release the slot before this job ends.
            let _unit_slot = unit_slot;
            let result =
                pool.execute_direct_transaction(StorageCapability::Events, op, move |conn| {
                    f(conn).map_err(|error| map_err(error, op))
                });
            if is_file_backed {
                if let Err(error) = &result {
                    crate::timeout_sink::maybe_emit_busy_storage_error(
                        &db,
                        crate::timeout_sink::Site::StandaloneEvent,
                        error,
                    );
                }
            }
            result
        })
        .await
        .map_err(|e| StorageError::driver(StorageCapability::Events, op, e))?
    }

    async fn with_reader<F, R>(&self, op: &'static str, f: F) -> Result<R, StorageError>
    where
        F: FnOnce(&rusqlite::Connection) -> Result<R, rusqlite::Error> + Send + 'static,
        R: Send + 'static,
    {
        super::run_pooled_store_read(
            Arc::clone(&self.pool),
            StorageCapability::Events,
            op,
            move |conn| f(conn).map_err(|error| map_err(error, op)),
        )
        .await
    }
}

// =============================================================================
// Helpers: parse SubstrateKind / EventOutcome / EventKind from DB strings
// =============================================================================

fn substrate_from_str(s: &str) -> Result<SubstrateKind, rusqlite::Error> {
    s.parse::<SubstrateKind>().map_err(|_| {
        rusqlite::Error::FromSqlConversionFailure(
            0,
            rusqlite::types::Type::Text,
            format!("unknown SubstrateKind: {s}").into(),
        )
    })
}

fn outcome_from_str(s: &str) -> Result<EventOutcome, rusqlite::Error> {
    match s {
        "success" => Ok(EventOutcome::Success),
        "denied" => Ok(EventOutcome::Denied),
        "error" => Ok(EventOutcome::Error),
        other => Err(rusqlite::Error::FromSqlConversionFailure(
            0,
            rusqlite::types::Type::Text,
            format!("unknown EventOutcome: {other}").into(),
        )),
    }
}

fn kind_from_str(s: &str) -> Result<EventKind, rusqlite::Error> {
    s.parse::<EventKind>().map_err(|_| {
        rusqlite::Error::FromSqlConversionFailure(
            0,
            rusqlite::types::Type::Text,
            format!("unknown EventKind: {s}").into(),
        )
    })
}

fn referent_kind_from_str(s: &str) -> Result<ReferentKind, rusqlite::Error> {
    match s {
        "entity" => Ok(ReferentKind::Entity),
        "note" => Ok(ReferentKind::Note),
        "edge" => Ok(ReferentKind::Edge),
        other => Err(rusqlite::Error::FromSqlConversionFailure(
            0,
            rusqlite::types::Type::Text,
            format!("unknown ReferentKind: {other}").into(),
        )),
    }
}

fn observation_role_from_str(s: &str) -> Result<ObservationRole, rusqlite::Error> {
    match s {
        "candidate" => Ok(ObservationRole::Candidate),
        "selected" => Ok(ObservationRole::Selected),
        "target" => Ok(ObservationRole::Target),
        "signal" => Ok(ObservationRole::Signal),
        other => Err(rusqlite::Error::FromSqlConversionFailure(
            0,
            rusqlite::types::Type::Text,
            format!("unknown ObservationRole: {other}").into(),
        )),
    }
}

fn parse_uuid(s: &str) -> Result<Uuid, rusqlite::Error> {
    Uuid::parse_str(s).map_err(|e| {
        rusqlite::Error::FromSqlConversionFailure(0, rusqlite::types::Type::Text, Box::new(e))
    })
}

// Column order: id(0), namespace(1), verb(2), substrate(3), actor(4),
//               kind(5), outcome(6), payload(7), payload_schema_version(8),
//               profile_state_version(9), duration_us(10), target_id(11),
//               session_id(12), aggregate_kind(13), aggregate_id(14), created_at(15),
//               op_index(16), ref_resolution(17)
fn read_event(row: &rusqlite::Row<'_>) -> Result<Event, rusqlite::Error> {
    let id_str: String = row.get(0)?;
    let namespace: String = row.get(1)?;
    let verb: String = row.get(2)?;
    let substrate_str: String = row.get(3)?;
    let actor: String = row.get(4)?;
    let kind_str: String = row.get(5)?;
    let outcome_str: String = row.get(6)?;
    let payload_str: String = row.get(7)?;
    let payload_schema_version: i64 = row.get(8)?;
    let profile_state_version: Option<i64> = row.get(9)?;
    let duration_us: i64 = row.get(10)?;
    let target_str: Option<String> = row.get(11)?;
    let session_str: Option<String> = row.get(12)?;
    let aggregate_kind: Option<String> = row.get(13)?;
    let aggregate_str: Option<String> = row.get(14)?;
    let created_at: i64 = row.get(15)?;
    let op_index = row.get::<_, Option<u32>>(16)?;
    let ref_resolution = row
        .get::<_, Option<String>>(17)?
        .map(|value| {
            value
                .parse::<khive_types::RefResolution>()
                .map_err(|error| {
                    rusqlite::Error::FromSqlConversionFailure(
                        17,
                        rusqlite::types::Type::Text,
                        Box::new(error),
                    )
                })
        })
        .transpose()?;
    if op_index.is_some() != ref_resolution.is_some() {
        return Err(rusqlite::Error::FromSqlConversionFailure(
            16,
            rusqlite::types::Type::Integer,
            "event operation attribution must be present or absent together".into(),
        ));
    }

    let id = parse_uuid(&id_str)?;
    let substrate = substrate_from_str(&substrate_str)?;
    let kind = kind_from_str(&kind_str)?;
    let outcome = outcome_from_str(&outcome_str)?;
    let payload: serde_json::Value = serde_json::from_str(&payload_str).map_err(|e| {
        rusqlite::Error::FromSqlConversionFailure(7, rusqlite::types::Type::Text, Box::new(e))
    })?;
    let target_id = target_str.as_deref().map(parse_uuid).transpose()?;
    let session_id = session_str.as_deref().map(parse_uuid).transpose()?;
    let aggregate_id = aggregate_str.as_deref().map(parse_uuid).transpose()?;
    let payload_schema_version_u32: u32 = payload_schema_version.try_into().map_err(|_| {
        rusqlite::Error::FromSqlConversionFailure(
            8,
            rusqlite::types::Type::Integer,
            format!("payload_schema_version {payload_schema_version} out of u32 range").into(),
        )
    })?;
    let profile_state_version_u64: Option<u64> = profile_state_version
        .map(|v| {
            u64::try_from(v).map_err(|_| {
                rusqlite::Error::FromSqlConversionFailure(
                    9,
                    rusqlite::types::Type::Integer,
                    format!("profile_state_version {v} out of u64 range").into(),
                )
            })
        })
        .transpose()?;

    Ok(Event {
        id,
        namespace,
        verb,
        substrate,
        actor,
        kind,
        outcome,
        payload,
        payload_schema_version: payload_schema_version_u32,
        profile_state_version: profile_state_version_u64,
        duration_us,
        target_id,
        session_id,
        aggregate_kind,
        aggregate_id,
        created_at,
        op_index,
        ref_resolution,
    })
}

// =============================================================================
// Helpers: observation projection write path
// =============================================================================

fn insert_event_with_observations(
    conn: &rusqlite::Connection,
    event: &Event,
) -> Result<(), rusqlite::Error> {
    validate_operation_pair(event)?;
    let id_str = event.id.to_string();
    let substrate_str = event.substrate.name().to_string();
    let kind_str = event.kind.name().to_string();
    let outcome_str = event.outcome.name().to_string();
    let payload_str = event.payload.to_string();
    let target_str = event.target_id.map(|u| u.to_string());
    let session_str = event.session_id.map(|u| u.to_string());
    let aggregate_str = event.aggregate_id.map(|u| u.to_string());
    let profile_state_version = profile_state_version_to_sql(event)?;

    conn.execute(
        "INSERT INTO events \
         (id, namespace, verb, substrate, actor, kind, outcome, payload, payload_schema_version, \
          profile_state_version, duration_us, target_id, session_id, aggregate_kind, aggregate_id, created_at, op_index, ref_resolution) \
         VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16, ?17, ?18)",
        rusqlite::params![
            id_str,
            &event.namespace,
            &event.verb,
            substrate_str,
            &event.actor,
            kind_str,
            outcome_str,
            payload_str,
            event.payload_schema_version as i64,
            profile_state_version,
            event.duration_us,
            target_str,
            session_str,
            &event.aggregate_kind,
            aggregate_str,
            event.created_at,
            event.op_index,
            event.ref_resolution.map(|value| value.name()),
        ],
    )?;

    for observation in decode_event_observations(event)? {
        conn.execute(
            "INSERT INTO event_observations \
             (event_id, entity_id, referent_kind, role, position) \
             VALUES (?1, ?2, ?3, ?4, ?5)",
            rusqlite::params![
                observation.event_id.to_string(),
                observation.entity_id.to_string(),
                observation.referent_kind.name(),
                observation.role.name(),
                observation.position as i64,
            ],
        )?;
    }

    Ok(())
}

/// Append an event on the caller's existing SQLite transaction, including the
/// same observation projection as `SqlEventStore::append_event`. This does
/// not open or commit a transaction: the caller must roll back its domain
/// changes if this insert fails.
pub fn append_event_in_transaction(
    conn: &rusqlite::Connection,
    event: &Event,
) -> Result<(), rusqlite::Error> {
    insert_event_with_observations(conn, event)
}

/// DML-only batch append loop shared by both the legacy (flag-off) and
/// WriterTask-routed (flag-on) `append_events` paths (ADR-067 Component A).
///
/// Issues no `BEGIN` / `COMMIT` / `ROLLBACK` itself — the caller owns the
/// enclosing transaction. All-or-nothing: the first failed insert returns
/// `Err` immediately (matching the pre-existing `append_events` contract) —
/// the caller's transaction wrapper issues the ROLLBACK.
fn batch_append_events_dml(
    conn: &rusqlite::Connection,
    events: &[Event],
    attempted: u64,
) -> Result<BatchWriteSummary, rusqlite::Error> {
    let mut affected = 0u64;
    for event in events {
        insert_event_with_observations(conn, event)?;
        affected += 1;
    }
    Ok(BatchWriteSummary {
        attempted,
        affected,
        ..BatchWriteSummary::default()
    })
}

fn fetch_event_by_id(
    conn: &rusqlite::Connection,
    id: Uuid,
) -> Result<Option<Event>, rusqlite::Error> {
    let id_str = id.to_string();
    let mut stmt = conn.prepare(
        "SELECT id, namespace, verb, substrate, actor, kind, outcome, payload, \
                payload_schema_version, profile_state_version, duration_us, target_id, \
                session_id, aggregate_kind, aggregate_id, created_at, op_index, ref_resolution \
         FROM events WHERE id = ?1",
    )?;
    let mut rows = stmt.query(rusqlite::params![id_str])?;
    match rows.next()? {
        Some(row) => Ok(Some(read_event(row)?)),
        None => Ok(None),
    }
}

/// Fetch `event_id`'s persisted observation projection in the same order it
/// was written: `event_observations`'s primary key is `(event_id, role,
/// position)`, and every `decode_*_observations` builder emits one role
/// group at a time in a fixed sequence that sorts identically by `role`
/// text, so `ORDER BY role, position` reproduces the original emission
/// order without a separate sequence column.
fn fetch_event_observations(
    conn: &rusqlite::Connection,
    event_id: Uuid,
) -> Result<Vec<EventObservation>, rusqlite::Error> {
    let id_str = event_id.to_string();
    let mut stmt = conn.prepare(
        "SELECT event_id, entity_id, referent_kind, role, position \
         FROM event_observations WHERE event_id = ?1 ORDER BY role, position",
    )?;
    let rows = stmt.query_map(rusqlite::params![id_str], |row| {
        let event_id: String = row.get(0)?;
        let entity_id: String = row.get(1)?;
        let referent_kind: String = row.get(2)?;
        let role: String = row.get(3)?;
        let position: i64 = row.get(4)?;
        Ok((event_id, entity_id, referent_kind, role, position))
    })?;

    let mut out = Vec::new();
    for row in rows {
        let (event_id, entity_id, referent_kind, role, position) = row?;
        let position_u32: u32 = position.try_into().map_err(|_| {
            rusqlite::Error::FromSqlConversionFailure(
                4,
                rusqlite::types::Type::Integer,
                format!("position {position} out of u32 range").into(),
            )
        })?;
        out.push(EventObservation {
            event_id: parse_uuid(&event_id)?,
            entity_id: parse_uuid(&entity_id)?,
            referent_kind: referent_kind_from_str(&referent_kind)?,
            role: observation_role_from_str(&role)?,
            position: position_u32,
        });
    }
    Ok(out)
}

/// DML-only idempotent batch loop shared by both the writer-task-routed and
/// standalone-transaction `append_events_idempotent` paths, mirroring
/// [`batch_append_events_dml`]'s split. A row whose id is unseen is
/// inserted; a row whose id already exists is compared column-for-column
/// plus its ordered observation projection against the submitted event —
/// exact equality reports `AlreadyPresentIdentical` and skips the write,
/// any mismatch reports `IdentityConflict` and skips the write for that row
/// alone. A genuine store failure on a fresh insert still aborts the whole
/// batch via `Err`, matching `append_events`'s all-or-nothing contract; only
/// identity comparison is per-row.
fn idempotent_batch_dml(
    conn: &rusqlite::Connection,
    events: &[Event],
) -> Result<IdempotentEventBatchResult, rusqlite::Error> {
    let mut rows = Vec::with_capacity(events.len());
    for event in events {
        validate_operation_pair(event)?;
        match fetch_event_by_id(conn, event.id)? {
            None => {
                insert_event_with_observations(conn, event)?;
                rows.push(EventAppendDisposition::Inserted);
            }
            Some(existing) => {
                let existing_observations = fetch_event_observations(conn, event.id)?;
                let submitted_observations = decode_event_observations(event)?;
                if existing == *event && existing_observations == submitted_observations {
                    rows.push(EventAppendDisposition::AlreadyPresentIdentical);
                } else {
                    rows.push(EventAppendDisposition::IdentityConflict);
                }
            }
        }
    }
    Ok(IdempotentEventBatchResult { rows })
}

/// Pure statement builder (ADR-099 B3 r6 structural cut): the exact
/// `events` row insert plus every derived `event_observations` insert for
/// `event`, as plain [`SqlStatement`]s with no I/O. This is the ONE place
/// that knows the `events`/`event_observations` insert shape at the
/// statement-text level; both [`append_event_on_writer`] below (the async
/// `SqlWriter` execution path) and `khive-runtime`'s ADR-099 `--atomic`
/// prepare path (which needs plain-data statements it can fold into a
/// synchronous commit-phase plan, not an async writer call) build on this
/// function rather than each hand-writing the INSERT text — the divergence
/// that produced the drift this cut fixes.
pub fn event_insert_statements(event: &Event) -> Result<Vec<SqlStatement>, rusqlite::Error> {
    validate_operation_pair(event)?;
    let id_str = event.id.to_string();
    let substrate_str = event.substrate.name().to_string();
    let kind_str = event.kind.name().to_string();
    let outcome_str = event.outcome.name().to_string();
    let payload_str = event.payload.to_string();
    let target_str = event.target_id.map(|u| u.to_string());
    let session_str = event.session_id.map(|u| u.to_string());
    let aggregate_str = event.aggregate_id.map(|u| u.to_string());
    let profile_state_version = profile_state_version_to_sql(event)?;

    let mut statements = vec![SqlStatement {
        sql: "INSERT INTO events \
              (id, namespace, verb, substrate, actor, kind, outcome, payload, payload_schema_version, \
               profile_state_version, duration_us, target_id, session_id, aggregate_kind, aggregate_id, created_at, op_index, ref_resolution) \
              VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16, ?17, ?18)"
            .into(),
        params: vec![
            SqlValue::Text(id_str),
            SqlValue::Text(event.namespace.clone()),
            SqlValue::Text(event.verb.clone()),
            SqlValue::Text(substrate_str),
            SqlValue::Text(event.actor.clone()),
            SqlValue::Text(kind_str),
            SqlValue::Text(outcome_str),
            SqlValue::Text(payload_str),
            SqlValue::Integer(event.payload_schema_version as i64),
            profile_state_version
                .map(SqlValue::Integer)
                .unwrap_or(SqlValue::Null),
            SqlValue::Integer(event.duration_us),
            target_str.map(SqlValue::Text).unwrap_or(SqlValue::Null),
            session_str.map(SqlValue::Text).unwrap_or(SqlValue::Null),
            event
                .aggregate_kind
                .clone()
                .map(SqlValue::Text)
                .unwrap_or(SqlValue::Null),
            aggregate_str.map(SqlValue::Text).unwrap_or(SqlValue::Null),
            SqlValue::Integer(event.created_at),
            event.op_index.map(|value| SqlValue::Integer(i64::from(value))).unwrap_or(SqlValue::Null),
            event.ref_resolution.map(|value| SqlValue::Text(value.name().into())).unwrap_or(SqlValue::Null),
        ],
        label: Some("event_insert_on_writer".into()),
    }];

    for observation in decode_event_observations(event)? {
        statements.push(SqlStatement {
            sql: "INSERT INTO event_observations \
                  (event_id, entity_id, referent_kind, role, position) \
                  VALUES (?1, ?2, ?3, ?4, ?5)"
                .into(),
            params: vec![
                SqlValue::Text(observation.event_id.to_string()),
                SqlValue::Text(observation.entity_id.to_string()),
                SqlValue::Text(observation.referent_kind.name().to_string()),
                SqlValue::Text(observation.role.name().to_string()),
                SqlValue::Integer(observation.position as i64),
            ],
            label: Some("event_observation_insert_on_writer".into()),
        });
    }

    Ok(statements)
}

fn validate_operation_pair(event: &Event) -> Result<(), rusqlite::Error> {
    if event.op_index.is_some() != event.ref_resolution.is_some() {
        return Err(rusqlite::Error::ToSqlConversionFailure(
            "event operation attribution must be present or absent together".into(),
        ));
    }
    profile_state_version_to_sql(event)?;
    Ok(())
}

fn profile_state_version_to_sql(event: &Event) -> Result<Option<i64>, rusqlite::Error> {
    event
        .profile_state_version
        .map(|version| {
            i64::try_from(version).map_err(|_| {
                rusqlite::Error::ToSqlConversionFailure(
                    format!("profile_state_version {version} exceeds i64::MAX").into(),
                )
            })
        })
        .transpose()
}

/// Build commit-time warning inserts for lineage-sensitive incident edges
/// removed by a hard-delete cascade (ADR-002).
///
/// Each statement handles one protected relation and reads `graph_edges`
/// when it executes, inside the caller's delete transaction. This is
/// intentionally not a prepare-time edge query: a guarded edge write that
/// commits immediately before the delete must be represented in the warning
/// payload before the subsequent cascade statement removes it. Relations
/// with no matching incident edge insert no event.
pub fn hard_delete_lineage_warning_statements(
    namespace: &str,
    actor: &str,
    target_id: Uuid,
    substrate: SubstrateKind,
) -> Vec<SqlStatement> {
    const WARNINGS: [(&str, &str); 5] = [
        ("derived_from", "provenance_loss"),
        ("supersedes", "replacement_lineage_loss"),
        ("precedes", "temporal_sequence_loss"),
        ("supports", "evidential_link_loss"),
        ("refutes", "evidential_link_loss"),
    ];

    let target_id = target_id.to_string();
    let created_at = chrono::Utc::now().timestamp_micros();
    let operation = khive_storage::operation_context::current_operation_attribution();
    WARNINGS
        .into_iter()
        .map(|(relation, warning)| SqlStatement {
            sql: "INSERT INTO events \
                  (id, namespace, verb, substrate, actor, kind, outcome, payload, \
                   payload_schema_version, profile_state_version, duration_us, target_id, \
                   session_id, aggregate_kind, aggregate_id, created_at, op_index, ref_resolution) \
                  SELECT ?1, ?2, 'delete', ?3, ?4, ?5, ?6, \
                         json_object( \
                           'severity', 'warning', \
                           'warning', ?7, \
                           'deleted_id', ?8, \
                           'relation', ?9, \
                           'edge_count', count(*), \
                           'edges', json_group_array(json_object( \
                             'id', incident.id, \
                             'namespace', incident.namespace, \
                             'source_id', incident.source_id, \
                             'target_id', incident.target_id, \
                             'deleted_at', incident.deleted_at)) \
                         ), \
                         1, NULL, 0, ?8, NULL, NULL, NULL, ?10, ?11, ?12 \
                  FROM ( \
                    SELECT id, namespace, source_id, target_id, deleted_at \
                    FROM graph_edges \
                    WHERE (source_id = ?8 OR target_id = ?8) AND relation = ?9 \
                    ORDER BY namespace, id \
                  ) AS incident \
                  HAVING count(*) > 0"
                .into(),
            params: vec![
                SqlValue::Text(Uuid::new_v4().to_string()),
                SqlValue::Text(namespace.to_string()),
                SqlValue::Text(substrate.name().to_string()),
                SqlValue::Text(actor.to_string()),
                SqlValue::Text(EventKind::Audit.name().to_string()),
                SqlValue::Text(EventOutcome::Success.name().to_string()),
                SqlValue::Text(warning.to_string()),
                SqlValue::Text(target_id.clone()),
                SqlValue::Text(relation.to_string()),
                SqlValue::Integer(created_at),
                operation
                    .map(|value| SqlValue::Integer(i64::from(value.op_index)))
                    .unwrap_or(SqlValue::Null),
                operation
                    .map(|value| SqlValue::Text(value.ref_resolution.name().into()))
                    .unwrap_or(SqlValue::Null),
            ],
            label: Some(format!("hard-delete-{relation}-warning")),
        })
        .collect()
}

/// Insert `event` (and any derived `event_observations` rows) through the
/// `khive-storage` `SqlWriter` seam, on a transaction the CALLER already
/// opened and controls the commit/rollback boundary for. Issues no
/// `BEGIN`/`COMMIT`/`ROLLBACK` of its own.
///
/// This exists alongside `insert_event_with_observations` (the raw
/// `rusqlite::Connection` path `SqlEventStore::append_event` uses, whose
/// caller now owns an admitted transaction for the ordinary case) rather
/// than replacing it: the two run on
/// different connection abstractions — a standalone `rusqlite::Connection`
/// vs. a `Box<dyn SqlWriter>` — so they cannot share one function body. Both
/// build on [`event_insert_statements`] for the actual insert shape.
/// Callers that need the event append to be part of a larger atomic unit —
/// e.g. ADR-081's brain fold gate (`khive-pack-brain/src/fold_gate.rs`),
/// which holds its own `BEGIN IMMEDIATE` transaction on a `SqlWriter` for
/// the dedup claim + mass fold and needs the feedback event to land in that
/// same transaction — call this instead of duplicating the insert shape
/// into their own crate.
pub async fn append_event_on_writer(
    writer: &mut dyn SqlWriter,
    event: &Event,
) -> Result<(), StorageError> {
    let statements =
        event_insert_statements(event).map_err(|e| map_err(e, "decode_event_observations"))?;
    for statement in statements {
        writer.execute(statement).await?;
    }
    Ok(())
}

fn decode_event_observations(event: &Event) -> Result<Vec<EventObservation>, rusqlite::Error> {
    match event.kind {
        EventKind::RerankExecuted => decode_rerank_observations(event),
        EventKind::RecallExecuted => decode_recall_observations(event),
        EventKind::SearchExecuted => decode_search_observations(event),
        EventKind::LinkCreated => decode_link_observations(event),
        EventKind::EdgeUpdated | EventKind::EdgeDeleted => decode_edge_target_observation(event),
        EventKind::EntityCreated
        | EventKind::EntityUpdated
        | EventKind::EntityDeleted
        | EventKind::NoteCreated
        | EventKind::NoteUpdated
        | EventKind::NoteDeleted
        | EventKind::TaskTransitioned => decode_target_observation(event),
        EventKind::FeedbackExplicit => decode_signal_observation(event),
        _ => Ok(Vec::new()),
    }
}

fn payload_uuid_array(event: &Event, field: &'static str) -> Result<Vec<Uuid>, rusqlite::Error> {
    Ok(payload_uuid_array_opt(event, field)?.unwrap_or_default())
}

/// Like [`payload_uuid_array`], but distinguishes an absent `field` (`Ok(None)`)
/// from a present-but-malformed one (`Err`). Callers that fall back across a
/// chain of alternative field names (e.g. `decode_rerank_observations`'s
/// `final_scores`/`reranked`) need that distinction: collapsing
/// "missing" into `Ok(vec![])` makes every later field in the chain
/// unreachable, since `Result::or_else` only fires on `Err`.
fn payload_uuid_array_opt(
    event: &Event,
    field: &'static str,
) -> Result<Option<Vec<Uuid>>, rusqlite::Error> {
    let Some(values) = event.payload.get(field) else {
        return Ok(None);
    };
    let Some(array) = values.as_array() else {
        return Err(invalid_payload(event.kind, field, "expected array"));
    };

    array
        .iter()
        .map(|value| {
            value
                .as_str()
                .ok_or_else(|| invalid_payload(event.kind, field, "expected UUID string"))
                .and_then(|s| Uuid::parse_str(s).map_err(|e| invalid_payload(event.kind, field, e)))
        })
        .collect::<Result<Vec<_>, _>>()
        .map(Some)
}

/// Decode `RerankExecutedPayload::final_scores` (`Vec<(Id128, f32)>` per
/// `khive_types::event::RerankExecutedPayload`) into the UUID leading each
/// tuple. Deserializes through that exact typed tuple shape, so a tuple that
/// is not precisely two elements, or whose second element is not a finite
/// score, is rejected rather than silently accepted. Absent field is
/// `Ok(None)`; present-but-wrong-shape is `Err`, matching
/// [`payload_uuid_array_opt`]'s missing-vs-malformed contract.
fn payload_final_scores_uuid_array_opt(
    event: &Event,
    field: &'static str,
) -> Result<Option<Vec<Uuid>>, rusqlite::Error> {
    let Some(values) = event.payload.get(field) else {
        return Ok(None);
    };
    let tuples: Vec<(khive_types::Id128, f32)> = serde_json::from_value(values.clone())
        .map_err(|e| invalid_payload(event.kind, field, e))?;
    tuples
        .into_iter()
        .map(|(id, score)| {
            if !score.is_finite() {
                return Err(invalid_payload(event.kind, field, "score is not finite"));
            }
            Ok(Uuid::from_bytes(*id.as_bytes()))
        })
        .collect::<Result<Vec<_>, _>>()
        .map(Some)
}

/// Decode `RerankExecutedPayload::reranked` (`Vec<(Id128, Vec<(String, f32)>)>`
/// per `khive_types::event::RerankExecutedPayload`) into the UUID leading each
/// tuple. Same exact-shape decoding contract as
/// [`payload_final_scores_uuid_array_opt`], applied to `reranked`'s
/// per-reranker sub-score shape instead of a single scalar score.
fn payload_reranked_uuid_array_opt(
    event: &Event,
    field: &'static str,
) -> Result<Option<Vec<Uuid>>, rusqlite::Error> {
    let Some(values) = event.payload.get(field) else {
        return Ok(None);
    };
    let tuples: Vec<(khive_types::Id128, Vec<(String, f32)>)> =
        serde_json::from_value(values.clone())
            .map_err(|e| invalid_payload(event.kind, field, e))?;
    tuples
        .into_iter()
        .map(|(id, scores)| {
            if !scores.iter().all(|(_, s)| s.is_finite()) {
                return Err(invalid_payload(event.kind, field, "score is not finite"));
            }
            Ok(Uuid::from_bytes(*id.as_bytes()))
        })
        .collect::<Result<Vec<_>, _>>()
        .map(Some)
}

fn payload_uuid(event: &Event, field: &'static str) -> Result<Option<Uuid>, rusqlite::Error> {
    let Some(value) = event.payload.get(field) else {
        return Ok(None);
    };
    let Some(s) = value.as_str() else {
        return Err(invalid_payload(event.kind, field, "expected UUID string"));
    };
    Uuid::parse_str(s)
        .map(Some)
        .map_err(|e| invalid_payload(event.kind, field, e))
}

fn decode_candidate_observations(
    event: &Event,
    referent_kind: ReferentKind,
) -> Result<Vec<EventObservation>, rusqlite::Error> {
    let mut rows = Vec::new();

    for (position, entity_id) in payload_uuid_array(event, "candidates")?
        .into_iter()
        .enumerate()
    {
        let position_u32 = u32::try_from(position).map_err(|_| {
            invalid_payload(
                event.kind,
                "candidates[position]",
                "position out of u32 range",
            )
        })?;
        rows.push(EventObservation {
            event_id: event.id,
            entity_id,
            referent_kind,
            role: ObservationRole::Candidate,
            position: position_u32,
        });
    }

    Ok(rows)
}

fn push_selected_observations(
    event: &Event,
    selected: Vec<Uuid>,
    referent_kind: ReferentKind,
    rows: &mut Vec<EventObservation>,
) -> Result<(), rusqlite::Error> {
    for (position, entity_id) in selected.into_iter().enumerate() {
        let position_u32 = u32::try_from(position).map_err(|_| {
            invalid_payload(
                event.kind,
                "selected[position]",
                "position out of u32 range",
            )
        })?;
        rows.push(EventObservation {
            event_id: event.id,
            entity_id,
            referent_kind,
            role: ObservationRole::Selected,
            position: position_u32,
        });
    }
    Ok(())
}

/// `RecallExecuted` payloads carry flat candidate and selected note UUID lists.
fn decode_recall_observations(event: &Event) -> Result<Vec<EventObservation>, rusqlite::Error> {
    let mut rows = decode_candidate_observations(event, ReferentKind::Note)?;
    let selected = payload_uuid_array_opt(event, "selected")?.unwrap_or_default();
    push_selected_observations(event, selected, ReferentKind::Note, &mut rows)?;
    Ok(rows)
}

/// `SearchExecuted.result_kind` identifies which substrate owns every UUID in
/// the candidate and selected lists. A missing key is the historical note
/// shape under ADR-041 A3; present unknown or non-string values are invalid.
fn decode_search_observations(event: &Event) -> Result<Vec<EventObservation>, rusqlite::Error> {
    if event.payload.as_object().is_none() {
        return Err(invalid_payload(event.kind, "payload", "expected object"));
    }
    let referent_kind = match event.payload.get("result_kind") {
        Some(serde_json::Value::String(kind)) if kind == "entity" => ReferentKind::Entity,
        Some(serde_json::Value::String(kind)) if kind == "note" => ReferentKind::Note,
        Some(serde_json::Value::String(_)) => {
            return Err(invalid_payload(
                event.kind,
                "result_kind",
                "expected \"entity\" or \"note\"",
            ));
        }
        Some(_) => {
            return Err(invalid_payload(
                event.kind,
                "result_kind",
                "expected string \"entity\" or \"note\"",
            ));
        }
        None => ReferentKind::Note,
    };
    let mut rows = decode_candidate_observations(event, referent_kind)?;
    let selected = payload_uuid_array_opt(event, "selected")?.unwrap_or_default();
    push_selected_observations(event, selected, referent_kind, &mut rows)?;
    Ok(rows)
}

/// `RerankExecutedPayload` (`khive_types::event::RerankExecutedPayload`) has no
/// `selected` field at all — a stray `selected` key in the raw payload is not
/// part of its typed contract and must never be consulted. Per ADR-042 §5,
/// `final_scores` is the ordered rerank output (positions match output
/// order); `reranked` is per-reranker audit/debug data with no ordering
/// guarantee, so it is only a legacy fallback for events emitted before
/// `final_scores` existed. `reranked` is tried only when `final_scores` is
/// absent (`None`) — a present-but-malformed `final_scores` errors
/// immediately instead of masking the problem by falling through.
fn decode_rerank_observations(event: &Event) -> Result<Vec<EventObservation>, rusqlite::Error> {
    let mut rows = decode_candidate_observations(event, ReferentKind::Note)?;
    let selected = payload_final_scores_uuid_array_opt(event, "final_scores")?
        .or(payload_reranked_uuid_array_opt(event, "reranked")?)
        .unwrap_or_default();
    push_selected_observations(event, selected, ReferentKind::Note, &mut rows)?;

    Ok(rows)
}

fn decode_link_observations(event: &Event) -> Result<Vec<EventObservation>, rusqlite::Error> {
    let mut rows = Vec::new();
    let source_kind = payload_link_referent_kind(event, "source_kind")?;
    let target_kind = payload_link_referent_kind(event, "target_kind")?;
    if let Some(source) = payload_uuid(event, "source_id")? {
        if let Some(referent_kind) = source_kind {
            rows.push(EventObservation {
                event_id: event.id,
                entity_id: source,
                referent_kind,
                role: ObservationRole::Target,
                position: 0,
            });
        }
    }
    if let Some(target) = payload_uuid(event, "target_id")? {
        if let Some(referent_kind) = target_kind {
            rows.push(EventObservation {
                event_id: event.id,
                entity_id: target,
                referent_kind,
                role: ObservationRole::Target,
                position: 1,
            });
        }
    }
    if let Some(edge_id) = event.target_id.or(payload_uuid(event, "id")?) {
        rows.push(EventObservation {
            event_id: event.id,
            entity_id: edge_id,
            referent_kind: ReferentKind::Edge,
            role: ObservationRole::Target,
            position: 2,
        });
    }
    Ok(rows)
}

fn payload_link_referent_kind(
    event: &Event,
    field: &'static str,
) -> Result<Option<ReferentKind>, rusqlite::Error> {
    match event.payload.get(field) {
        None => Ok(Some(ReferentKind::Entity)),
        Some(value) => match value.as_str() {
            Some("entity") => Ok(Some(ReferentKind::Entity)),
            Some("note") => Ok(Some(ReferentKind::Note)),
            Some("edge") => Ok(Some(ReferentKind::Edge)),
            Some("event") => Ok(None),
            _ => Err(invalid_payload(
                event.kind,
                field,
                "expected \"entity\", \"note\", \"edge\", or \"event\"",
            )),
        },
    }
}

fn decode_edge_target_observation(event: &Event) -> Result<Vec<EventObservation>, rusqlite::Error> {
    let Some(edge_id) = event.target_id.or(payload_uuid(event, "id")?) else {
        return Ok(Vec::new());
    };
    Ok(vec![EventObservation {
        event_id: event.id,
        entity_id: edge_id,
        referent_kind: ReferentKind::Edge,
        role: ObservationRole::Target,
        position: 0,
    }])
}

fn decode_target_observation(event: &Event) -> Result<Vec<EventObservation>, rusqlite::Error> {
    let Some(entity_id) = event.target_id.or(payload_uuid(event, "target_id")?) else {
        return Ok(Vec::new());
    };
    Ok(vec![EventObservation {
        event_id: event.id,
        entity_id,
        referent_kind: if event.substrate == SubstrateKind::Note {
            ReferentKind::Note
        } else {
            ReferentKind::Entity
        },
        role: ObservationRole::Target,
        position: 0,
    }])
}

fn decode_signal_observation(event: &Event) -> Result<Vec<EventObservation>, rusqlite::Error> {
    let Some(entity_id) = event.target_id else {
        return Ok(Vec::new());
    };
    Ok(vec![EventObservation {
        event_id: event.id,
        entity_id,
        // ADR-041 permits both entity and note signal targets. `brain.feedback`
        // threads the resolved target's substrate onto the event (#831);
        // pre-fix events all carry the old
        // `SubstrateKind::Event` placeholder and fall back to Entity, matching
        // the decoder's prior hard-coded behavior for that historical data.
        referent_kind: if event.substrate == SubstrateKind::Note {
            ReferentKind::Note
        } else {
            ReferentKind::Entity
        },
        role: ObservationRole::Signal,
        position: 0,
    }])
}

fn invalid_payload(
    kind: EventKind,
    field: &'static str,
    reason: impl std::fmt::Display,
) -> rusqlite::Error {
    rusqlite::Error::ToSqlConversionFailure(
        format!("invalid payload for {}.{field}: {reason}", kind.name()).into(),
    )
}

// =============================================================================
// Helpers: filter SQL builder
// =============================================================================

fn build_event_filter_sql(
    conn: &rusqlite::Connection,
    default_namespace: &str,
    filter: &EventFilter,
) -> Result<(String, Vec<Box<dyn rusqlite::types::ToSql>>), rusqlite::Error> {
    reject_missing_event_filter_schema(conn, filter)?;

    let mut conditions: Vec<String> = Vec::new();
    let mut params: Vec<Box<dyn rusqlite::types::ToSql>> = Vec::new();

    params.push(Box::new(default_namespace.to_string()));
    conditions.push(format!("namespace = ?{}", params.len()));

    push_in_clause(
        &mut conditions,
        &mut params,
        "id",
        filter.ids.iter().map(Uuid::to_string),
    );
    push_in_clause(
        &mut conditions,
        &mut params,
        "kind",
        filter.kinds.iter().map(|kind| kind.name().to_string()),
    );
    push_in_clause(
        &mut conditions,
        &mut params,
        "verb",
        filter.verbs.iter().cloned(),
    );
    push_in_clause(
        &mut conditions,
        &mut params,
        "substrate",
        filter.substrates.iter().map(|s| s.name().to_string()),
    );
    push_in_clause(
        &mut conditions,
        &mut params,
        "actor",
        filter.actors.iter().cloned(),
    );

    if let Some(outcome) = filter.outcome {
        params.push(Box::new(outcome.name().to_string()));
        conditions.push(format!("outcome = ?{}", params.len()));
    }
    super::append_json_equalities(
        &mut conditions,
        &mut params,
        "payload",
        &filter.payload_equalities,
    );

    if let Some(after) = filter.after {
        params.push(Box::new(after));
        conditions.push(format!("created_at > ?{}", params.len()));
    }

    if let Some(before) = filter.before {
        params.push(Box::new(before));
        conditions.push(format!("created_at < ?{}", params.len()));
    }

    if let Some(target_id) = filter.target_id {
        params.push(Box::new(target_id.to_string()));
        conditions.push(format!("target_id = ?{}", params.len()));
    }

    if let Some(session_id) = filter.session_id {
        params.push(Box::new(session_id.to_string()));
        conditions.push(format!("session_id = ?{}", params.len()));
    }

    push_observation_exists(&mut conditions, &mut params, None, &filter.observed);
    push_observation_exists(
        &mut conditions,
        &mut params,
        Some("selected"),
        &filter.selected,
    );

    if let Some(proposal_id) = filter.payload_proposal_id {
        params.push(Box::new(proposal_id.to_string()));
        conditions.push(format!(
            "json_extract(payload, '$.proposal_id') = ?{}",
            params.len()
        ));
    }

    let clause = format!(" WHERE {}", conditions.join(" AND "));
    Ok((clause, params))
}

fn push_in_clause<I>(
    conditions: &mut Vec<String>,
    params: &mut Vec<Box<dyn rusqlite::types::ToSql>>,
    column: &'static str,
    values: I,
) where
    I: IntoIterator<Item = String>,
{
    let placeholders: Vec<String> = values
        .into_iter()
        .map(|value| {
            params.push(Box::new(value));
            format!("?{}", params.len())
        })
        .collect();
    if !placeholders.is_empty() {
        conditions.push(format!("{column} IN ({})", placeholders.join(",")));
    }
}

fn push_observation_exists(
    conditions: &mut Vec<String>,
    params: &mut Vec<Box<dyn rusqlite::types::ToSql>>,
    role: Option<&'static str>,
    entity_ids: &[Uuid],
) {
    if entity_ids.is_empty() {
        return;
    }
    let placeholders: Vec<String> = entity_ids
        .iter()
        .map(|id| {
            params.push(Box::new(id.to_string()));
            format!("?{}", params.len())
        })
        .collect();
    let role_clause = role
        .map(|role| format!(" AND o.role = '{role}'"))
        .unwrap_or_default();
    conditions.push(format!(
        "EXISTS (SELECT 1 FROM event_observations o \
         WHERE o.event_id = events.id{role_clause} AND o.entity_id IN ({}))",
        placeholders.join(",")
    ));
}

fn reject_missing_event_filter_schema(
    conn: &rusqlite::Connection,
    filter: &EventFilter,
) -> Result<(), rusqlite::Error> {
    if filter.target_id.is_some() && !has_column(conn, "events", "target_id")? {
        return Err(schema_absent("events.target_id"));
    }
    if filter.session_id.is_some() && !has_column(conn, "events", "session_id")? {
        return Err(schema_absent("events.session_id"));
    }
    if (!filter.observed.is_empty() || !filter.selected.is_empty())
        && !has_table(conn, "event_observations")?
    {
        return Err(schema_absent("event_observations"));
    }
    if filter.payload_proposal_id.is_some() && !has_column(conn, "events", "payload")? {
        return Err(schema_absent("events.payload"));
    }
    Ok(())
}

fn has_table(conn: &rusqlite::Connection, table: &'static str) -> Result<bool, rusqlite::Error> {
    conn.query_row(
        "SELECT COUNT(*) > 0 FROM sqlite_master WHERE type = 'table' AND name = ?1",
        [table],
        |row| row.get(0),
    )
}

fn has_column(
    conn: &rusqlite::Connection,
    table: &'static str,
    column: &'static str,
) -> Result<bool, rusqlite::Error> {
    conn.query_row(
        "SELECT COUNT(*) > 0 FROM pragma_table_info(?1) WHERE name = ?2",
        rusqlite::params![table, column],
        |row| row.get(0),
    )
}

fn schema_absent(name: &'static str) -> rusqlite::Error {
    rusqlite::Error::ToSqlConversionFailure(
        format!("event filter requires missing schema element {name}; run migrations").into(),
    )
}

// =============================================================================
// EventStore implementation
// =============================================================================

#[async_trait]
impl EventStore for SqlEventStore {
    async fn append_event(&self, event: Event) -> Result<(), StorageError> {
        // ADR-067 Component A: when the write queue is enabled, route
        // through the pool-wide WriterTask. DML-only closure — no BEGIN
        // IMMEDIATE/COMMIT/ROLLBACK here, since the WriterTask's run loop
        // owns the transaction.
        //
        // `event_rows` is counted here at the store seam (on success, both
        // paths) so every request-owned append — proposal lifecycle,
        // curation, mutation events — is covered without per-call-site
        // instrumentation. The enclosing per-dispatch audit row is appended
        // only after the usage snapshot is frozen, so it never counts itself.
        if let Some(writer_task) = self.current_writer_task("append_event")? {
            let result = writer_task
                .send_bounded(move |conn| {
                    insert_event_with_observations(conn, &event)
                        .map_err(|e| map_err(e, "append_event"))
                })
                .await;
            khive_storage::usage::account_event_write(result.as_ref().map(|()| 1));
            return result;
        }

        let result = self
            .with_writer("append_event", move |conn| {
                insert_event_with_observations(conn, &event)
            })
            .await;
        #[cfg(test)]
        let result = self.after_fallback_write_for_test(result);
        khive_storage::usage::account_event_write(result.as_ref().map(|()| 1));
        result
    }

    async fn append_events(&self, events: Vec<Event>) -> Result<BatchWriteSummary, StorageError> {
        let attempted = events.len() as u64;

        // ADR-067 Component A: when the write queue is enabled, route
        // through the pool-wide WriterTask. DML-only closure preserving the
        // all-or-nothing semantics (first failed insert aborts the whole
        // batch) — the WriterTask's run loop owns the enclosing transaction
        // and issues the ROLLBACK on `Err`.
        if let Some(writer_task) = self.current_writer_task("append_events")? {
            let result = writer_task
                .send_bounded(move |conn| {
                    batch_append_events_dml(conn, &events, attempted)
                        .map_err(|e| map_err(e, "append_events"))
                })
                .await;
            khive_storage::usage::account_event_write(
                result.as_ref().map(|summary| summary.affected),
            );
            return result;
        }

        let result = self
            .with_writer("append_events", move |conn| {
                batch_append_events_dml(conn, &events, attempted)
            })
            .await;
        #[cfg(test)]
        let result = self.after_fallback_write_for_test(result);
        khive_storage::usage::account_event_write(result.as_ref().map(|summary| summary.affected));
        result
    }

    async fn get_event(&self, id: Uuid) -> Result<Option<Event>, StorageError> {
        let namespace = self.namespace.clone();
        let id_str = id.to_string();

        self.with_reader("get_event", move |conn| {
            let mut stmt = conn.prepare(
                "SELECT id, namespace, verb, substrate, actor, kind, outcome, payload, \
                        payload_schema_version, profile_state_version, duration_us, target_id, \
                        session_id, aggregate_kind, aggregate_id, created_at, op_index, ref_resolution \
                 FROM events WHERE namespace = ?1 AND id = ?2",
            )?;
            let mut rows = stmt.query(rusqlite::params![namespace, id_str])?;
            match rows.next()? {
                Some(row) => Ok(Some(read_event(row)?)),
                None => Ok(None),
            }
        })
        .await
    }

    async fn query_events(
        &self,
        filter: EventFilter,
        page: PageRequest,
    ) -> Result<Page<Event>, StorageError> {
        super::validate_json_equality_paths(
            &filter.payload_equalities,
            StorageCapability::Events,
            "query_events",
        )?;
        let namespace = self.namespace.clone();
        let limit_i64 = i64::from(page.limit);
        let offset_i64 = i64::try_from(page.offset).map_err(|_| StorageError::InvalidInput {
            capability: StorageCapability::Events,
            operation: "query_events".into(),
            message: format!(
                "PageRequest: offset must be <= i64::MAX, got {}",
                page.offset
            ),
        })?;

        self.with_reader("query_events", move |conn| {
            // No `COUNT(*)` here, and `total` is therefore `None`. The count
            // carried no `LIMIT`, so it scanned the whole filtered set on every
            // paged read while the data query below fetches only
            // `offset + limit` rows — and on the merged event plane a single
            // read paid it twice, once per underlying store. `Page.total` is
            // `Option<u64>` precisely so a store may decline to compute it, and
            // the merged fold in `khive-runtime::events_split` propagates `None`
            // rather than inventing a number. `count_events` below remains for
            // callers that genuinely want a cardinality.
            let (where_clause, filter_params) = build_event_filter_sql(conn, &namespace, &filter)?;

            let mut all_params: Vec<Box<dyn rusqlite::types::ToSql>> = filter_params;
            all_params.push(Box::new(limit_i64));
            all_params.push(Box::new(offset_i64));

            let limit_idx = all_params.len() - 1;
            let offset_idx = all_params.len();

            let data_sql = format!(
                "SELECT id, namespace, verb, substrate, actor, kind, outcome, payload, \
                        payload_schema_version, profile_state_version, duration_us, target_id, \
                        session_id, aggregate_kind, aggregate_id, created_at, op_index, ref_resolution \
                 FROM events{} ORDER BY created_at DESC, id DESC LIMIT ?{} OFFSET ?{}",
                where_clause, limit_idx, offset_idx,
            );

            let mut stmt = conn.prepare(&data_sql)?;
            let param_refs: Vec<&dyn rusqlite::types::ToSql> =
                all_params.iter().map(|p| p.as_ref()).collect();
            let rows = stmt.query_map(param_refs.as_slice(), read_event)?;

            let mut items = Vec::new();
            for row in rows {
                items.push(row?);
            }

            Ok(Page { items, total: None })
        })
        .await
    }

    async fn query_event_page(
        &self,
        query: khive_storage::event::EventPageQuery,
    ) -> Result<khive_storage::event::EventPageWindow, StorageError> {
        cursor::query_event_page(self, query).await
    }

    async fn count_events(&self, filter: EventFilter) -> Result<u64, StorageError> {
        super::validate_json_equality_paths(
            &filter.payload_equalities,
            StorageCapability::Events,
            "count_events",
        )?;
        let namespace = self.namespace.clone();

        self.with_reader("count_events", move |conn| {
            let (where_clause, params) = build_event_filter_sql(conn, &namespace, &filter)?;
            let sql = format!("SELECT COUNT(*) FROM events{}", where_clause);
            let mut stmt = conn.prepare(&sql)?;
            let param_refs: Vec<&dyn rusqlite::types::ToSql> =
                params.iter().map(|p| p.as_ref()).collect();
            let count: i64 = stmt.query_row(param_refs.as_slice(), |row| row.get(0))?;
            Ok(count as u64)
        })
        .await
    }

    fn preflight_event(&self, event: &Event) -> Result<(), StorageError> {
        event_insert_statements(event)
            .map(|_| ())
            .map_err(|e| map_err(e, "preflight_event"))
    }

    async fn append_events_idempotent(
        &self,
        events: Vec<Event>,
    ) -> Result<IdempotentEventBatchResult, StorageError> {
        if let Some(writer_task) = self.current_writer_task("append_events_idempotent")? {
            return writer_task
                .send_bounded(move |conn| {
                    idempotent_batch_dml(conn, &events)
                        .map_err(|e| map_err(e, "append_events_idempotent"))
                })
                .await;
        }

        self.with_writer("append_events_idempotent", move |conn| {
            idempotent_batch_dml(conn, &events)
        })
        .await
    }

    fn supports_idempotent_audit_batch(&self) -> bool {
        true
    }
}

// =============================================================================
// DDL
// =============================================================================

const EVENTS_DDL: &str = include_str!("../../sql/events-ddl.sql");
const OPERATION_ATTRIBUTION_COLUMNS: &str =
    include_str!("../../sql/036-events-operation-attribution.sql");

pub(crate) fn ensure_events_schema(conn: &rusqlite::Connection) -> Result<(), rusqlite::Error> {
    conn.execute_batch(EVENTS_DDL)?;
    ensure_operation_attribution_columns(conn)
}

/// Add `op_index` and `ref_resolution` to an `events` table created before
/// they existed. `CREATE TABLE IF NOT EXISTS` leaves such a table unchanged,
/// and a standalone events database has no migration chain to run V36, so
/// every insert naming the two columns would fail. V36 calls this too, which
/// keeps it valid on a table the store DDL has already upgraded. Both columns
/// are added in one transaction (the caller's, if one is open).
pub(crate) fn ensure_operation_attribution_columns(
    conn: &rusqlite::Connection,
) -> Result<(), rusqlite::Error> {
    match (
        has_column(conn, "events", "op_index")?,
        has_column(conn, "events", "ref_resolution")?,
    ) {
        (true, true) => Ok(()),
        (false, false) if conn.is_autocommit() => {
            let tx = conn.unchecked_transaction()?;
            tx.execute_batch(OPERATION_ATTRIBUTION_COLUMNS)?;
            tx.commit()
        }
        (false, false) => conn.execute_batch(OPERATION_ATTRIBUTION_COLUMNS),
        (op_index, ref_resolution) => Err(rusqlite::Error::ToSqlConversionFailure(
            format!(
                "events table has only one operation attribution column \
                 (op_index={op_index}, ref_resolution={ref_resolution}); refusing to guess"
            )
            .into(),
        )),
    }
}

#[cfg(test)]
#[path = "event_tests.rs"]
mod tests;

#[cfg(test)]
#[path = "event_busy_tests.rs"]
mod direct_busy_tests;

#[cfg(test)]
#[path = "event_fallback_usage_tests.rs"]
mod fallback_usage_tests;