khive-pack-brain 0.5.0

//! ADR-081 §4 cross-session serve ledger — brain-owned record of recall serves
//! and their scorer-assigned grades.
//!
//! Row creation (`record_serve`) implements the normative write side of the
//! ledger; `memory.recall` wires into it via `brain.record_serve` (ADR-081 §5,
//! #394) to stamp `served_by_profile_id` and append the ledger row off the
//! response path. The read side is consumed by `brain.feedback` /
//! `brain.auto_feedback` when a caller supplies `serve_ledger_id` (ADR-081 §6):
//! dedup by `(scorer_run_id, id)`, the `accounting_profile_id` zero-weight
//! fail-safe, and grade backfill.
use khive_runtime::RuntimeError;
use khive_storage::types::{SqlStatement, SqlValue};
use khive_storage::SqlAccess;

pub use khive_brain_core::compute_query_class;

fn sql_err(context: &str, e: impl std::fmt::Display) -> RuntimeError {
    RuntimeError::Internal(format!("serve ledger {context}: {e}"))
}

/// One row of `brain_serve_ledger` (ADR-081 §4 normative schema).
pub struct ServeLedgerRow {
    pub id: String,
    #[allow(dead_code)]
    pub namespace: String,
    #[allow(dead_code)]
    pub consumer_kind: String,
    pub served_by_profile_id: Option<String>,
    pub resolved_profile_id: Option<String>,
    #[allow(dead_code)]
    pub resolved_at: Option<i64>,
    /// `COALESCE(served_by_profile_id, resolved_profile_id)` — generated by SQLite,
    /// read back here rather than recomputed, so a future change to the generated
    /// expression cannot silently diverge from what the fold gate reads.
    pub accounting_profile_id: Option<String>,
    #[allow(dead_code)]
    pub target_id: String,
    #[allow(dead_code)]
    pub query_class: String,
    #[allow(dead_code)]
    pub query_raw: String,
    #[allow(dead_code)]
    pub served_at: i64,
    #[allow(dead_code)]
    pub grade: Option<String>,
    #[allow(dead_code)]
    pub graded_at: Option<i64>,
    /// The scorer run that last graded this row, if any — the dedup token.
    pub scorer_run_id: Option<String>,
}

fn row_text(row: &khive_storage::types::SqlRow, col: &str) -> Result<String, RuntimeError> {
    match row.get(col) {
        Some(SqlValue::Text(s)) => Ok(s.clone()),
        _ => Err(sql_err("read row", format!("missing {col} column"))),
    }
}

fn row_opt_text(row: &khive_storage::types::SqlRow, col: &str) -> Option<String> {
    match row.get(col) {
        Some(SqlValue::Text(s)) => Some(s.clone()),
        _ => None,
    }
}

fn row_opt_int(row: &khive_storage::types::SqlRow, col: &str) -> Option<i64> {
    match row.get(col) {
        Some(SqlValue::Integer(n)) => Some(*n),
        _ => None,
    }
}

fn row_int(row: &khive_storage::types::SqlRow, col: &str) -> Result<i64, RuntimeError> {
    match row.get(col) {
        Some(SqlValue::Integer(n)) => Ok(*n),
        _ => Err(sql_err("read row", format!("missing {col} column"))),
    }
}

/// Record a new serve row (the ADR-081 §4 write side). Returns `Ok(true)` when
/// the row was written, `Ok(false)` when an exact-key duplicate was tolerated
/// (the natural key `(namespace, target_id, query_class, served_at)` already
/// has a row — a duplicate drain of the same batch, not an error).
#[allow(clippy::too_many_arguments)]
pub async fn record_serve(
    sql: &dyn SqlAccess,
    id: &str,
    namespace: &str,
    consumer_kind: &str,
    served_by_profile_id: Option<&str>,
    resolved_profile_id: Option<&str>,
    resolved_at: Option<i64>,
    target_id: &str,
    query_class: &str,
    query_raw: &str,
    served_at: i64,
) -> Result<bool, RuntimeError> {
    let mut writer = sql.writer().await.map_err(|e| sql_err("writer", e))?;
    let result = writer
        .execute(SqlStatement {
            sql: "INSERT INTO brain_serve_ledger \
                  (id, namespace, consumer_kind, served_by_profile_id, resolved_profile_id, \
                   resolved_at, target_id, query_class, query_raw, served_at) \
                  VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10)"
                .into(),
            params: vec![
                SqlValue::Text(id.to_string()),
                SqlValue::Text(namespace.to_string()),
                SqlValue::Text(consumer_kind.to_string()),
                served_by_profile_id
                    .map(|s| SqlValue::Text(s.to_string()))
                    .unwrap_or(SqlValue::Null),
                resolved_profile_id
                    .map(|s| SqlValue::Text(s.to_string()))
                    .unwrap_or(SqlValue::Null),
                resolved_at.map(SqlValue::Integer).unwrap_or(SqlValue::Null),
                SqlValue::Text(target_id.to_string()),
                SqlValue::Text(query_class.to_string()),
                SqlValue::Text(query_raw.to_string()),
                SqlValue::Integer(served_at),
            ],
            label: Some("brain_serve_ledger_insert".into()),
        })
        .await;
    match result {
        Ok(_) => Ok(true),
        Err(e) if e.is_unique_constraint_violation() => {
            // PR #583: `is_unique_constraint_violation` fires
            // for ANY unique-index collision on this table — the intended
            // natural key `(namespace, target_id, query_class, served_at)`,
            // but also the `id TEXT PRIMARY KEY` column and any future unique
            // index. Confirm the natural-key row actually exists before
            // reporting a tolerated duplicate; otherwise this would mask a
            // genuine `id` collision (or other future constraint) as a no-op.
            if natural_key_row_exists(sql, namespace, target_id, query_class, served_at).await? {
                Ok(false)
            } else {
                Err(sql_err("insert serve row", e))
            }
        }
        Err(e) => Err(sql_err("insert serve row", e)),
    }
}

async fn natural_key_row_exists(
    sql: &dyn SqlAccess,
    namespace: &str,
    target_id: &str,
    query_class: &str,
    served_at: i64,
) -> Result<bool, RuntimeError> {
    let mut reader = sql.reader().await.map_err(|e| sql_err("reader", e))?;
    let row = reader
        .query_row(SqlStatement {
            sql: "SELECT id FROM brain_serve_ledger \
                  WHERE namespace = ?1 AND target_id = ?2 AND query_class = ?3 AND served_at = ?4"
                .into(),
            params: vec![
                SqlValue::Text(namespace.to_string()),
                SqlValue::Text(target_id.to_string()),
                SqlValue::Text(query_class.to_string()),
                SqlValue::Integer(served_at),
            ],
            label: Some("brain_serve_ledger_natural_key_check".into()),
        })
        .await
        .map_err(|e| sql_err("check natural key", e))?;
    Ok(row.is_some())
}

/// Fetch a serve ledger row by id.
pub async fn get_serve_row(
    sql: &dyn SqlAccess,
    id: &str,
) -> Result<Option<ServeLedgerRow>, RuntimeError> {
    let mut reader = sql.reader().await.map_err(|e| sql_err("reader", e))?;
    let row = reader
        .query_row(SqlStatement {
            sql: "SELECT id, namespace, consumer_kind, served_by_profile_id, \
                  resolved_profile_id, resolved_at, accounting_profile_id, target_id, \
                  query_class, query_raw, served_at, grade, graded_at, scorer_run_id \
                  FROM brain_serve_ledger WHERE id = ?1"
                .into(),
            params: vec![SqlValue::Text(id.to_string())],
            label: Some("brain_serve_ledger_get".into()),
        })
        .await
        .map_err(|e| sql_err("read row", e))?;

    let Some(row) = row else {
        return Ok(None);
    };

    Ok(Some(ServeLedgerRow {
        id: row_text(&row, "id")?,
        namespace: row_text(&row, "namespace")?,
        consumer_kind: row_text(&row, "consumer_kind")?,
        served_by_profile_id: row_opt_text(&row, "served_by_profile_id"),
        resolved_profile_id: row_opt_text(&row, "resolved_profile_id"),
        resolved_at: row_opt_int(&row, "resolved_at"),
        accounting_profile_id: row_opt_text(&row, "accounting_profile_id"),
        target_id: row_text(&row, "target_id")?,
        query_class: row_text(&row, "query_class")?,
        query_raw: row_text(&row, "query_raw")?,
        served_at: row_int(&row, "served_at")?,
        grade: row_opt_text(&row, "grade"),
        graded_at: row_opt_int(&row, "graded_at"),
        scorer_run_id: row_opt_text(&row, "scorer_run_id"),
    }))
}

/// Backfill a serve row's grade after a scorer emits feedback for it
/// (ADR-081 §6: "the fold... backfills the ledger row's grade").
///
/// Also stamps `scorer_run_id`, which is what `dedup` reads on a later call —
/// idempotent per-row by `(scorer_run_id, id)` (ADR-081 §4).
pub async fn backfill_grade(
    sql: &dyn SqlAccess,
    id: &str,
    grade: &str,
    graded_at_us: i64,
    scorer_run_id: &str,
) -> Result<(), RuntimeError> {
    let mut writer = sql.writer().await.map_err(|e| sql_err("writer", e))?;
    writer
        .execute(SqlStatement {
            sql: "UPDATE brain_serve_ledger SET grade = ?1, graded_at = ?2, scorer_run_id = ?3 \
                  WHERE id = ?4"
                .into(),
            params: vec![
                SqlValue::Text(grade.to_string()),
                SqlValue::Integer(graded_at_us),
                SqlValue::Text(scorer_run_id.to_string()),
                SqlValue::Text(id.to_string()),
            ],
            label: Some("brain_serve_ledger_backfill_grade".into()),
        })
        .await
        .map_err(|e| sql_err("backfill grade", e))?;
    Ok(())
}

/// Outcome of resolving a `(scorer_run_id, serve_ledger_id)` pair against the
/// ledger, before the fold gate runs.
pub enum ServeLedgerResolution {
    /// This exact `(scorer_run_id, serve_ledger_id)` pair was already graded —
    /// the caller must treat this emission as a no-op (ADR-081 §2 dedup).
    AlreadyGraded,
    /// Row found and not yet graded by this `scorer_run_id`. Carries the
    /// accounting profile id to fold under, or `None` if unresolved (the
    /// caller must force a zero-weight fold — ADR-081 §4 fail-safe).
    Proceed {
        accounting_profile_id: Option<String>,
    },
    /// No row exists for `serve_ledger_id`.
    NotFound,
}

/// Resolve a `(scorer_run_id, serve_ledger_id)` pair against the ledger.
pub async fn resolve(
    sql: &dyn SqlAccess,
    serve_ledger_id: &str,
    scorer_run_id: &str,
) -> Result<ServeLedgerResolution, RuntimeError> {
    let Some(row) = get_serve_row(sql, serve_ledger_id).await? else {
        return Ok(ServeLedgerResolution::NotFound);
    };
    if row.scorer_run_id.as_deref() == Some(scorer_run_id) {
        return Ok(ServeLedgerResolution::AlreadyGraded);
    }
    Ok(ServeLedgerResolution::Proceed {
        accounting_profile_id: row.accounting_profile_id,
    })
}

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

    /// Fork C slice 2: proves `record_serve`'s single-row INSERT — issued via
    /// `sql.writer().execute(..)` — is actually enqueued on the pool's shared
    /// `WriterTaskHandle` channel when the write queue is enabled. This is
    /// the `SqliteWriter::execute` fix, which routes through the
    /// writer task like `execute_batch` already did): `record_serve` itself
    /// needed zero code changes, but this test proves the fix actually
    /// reaches this call site rather than asserting it by inspection alone.
    ///
    /// Deliberately NOT a wall-clock/occupier-timing test — see the sibling
    /// `fold_gate::tests::fold_gate_apply_routes_through_writer_task_when_flag_enabled`
    /// doc comment for why real SQLite file-level locking makes elapsed time
    /// alone vacuous here. Instead this reads `WriterTaskHandle::queue_depth`
    /// directly while an occupier deterministically holds the writer task's
    /// one drain slot open (parked on a oneshot via `blocking_recv`, not a
    /// sleep/timing race).
    ///
    /// Deliberately NOT built via a full `KhiveRuntime` + the
    /// `KHIVE_WRITE_QUEUE` env var: that env var is process-global and this
    /// crate's other tests are not `#[serial]` against it, so a window where
    /// it is set here could leak into a concurrently-scheduled test's own
    /// `KhiveRuntime::new` and unexpectedly flip its pool's write-queue flag
    /// (see `fold_gate::tests::fold_gate_apply_routes_through_writer_task_when_flag_enabled`'s
    /// doc comment for the concrete failure this caused before both tests
    /// were rewritten to avoid the env var). Constructing the pool directly
    /// with `write_queue_enabled: true` in the config literal, and driving
    /// `record_serve` over a bare `SqlBridge` instead of a full
    /// `KhiveRuntime`, sidesteps global mutable state entirely.
    #[tokio::test]
    async fn record_serve_routes_through_writer_task_when_flag_enabled() {
        let dir = tempfile::tempdir().expect("tempdir");
        let db_path = dir.path().join("serve-ledger-write-queue-routing.db");
        let pool_cfg = khive_db::PoolConfig {
            path: Some(db_path),
            write_queue_enabled: true,
            ..khive_db::PoolConfig::default()
        };
        let pool = std::sync::Arc::new(khive_db::ConnectionPool::new(pool_cfg).expect("pool"));
        {
            let mut writer = pool.writer().expect("writer");
            khive_db::run_migrations(writer.conn_mut()).expect("migrations");
        }
        let sql: std::sync::Arc<dyn SqlAccess> =
            std::sync::Arc::new(khive_db::SqlBridge::new(std::sync::Arc::clone(&pool), true));

        let writer_task = pool
            .writer_task_handle()
            .unwrap()
            .expect("writer task must be spawned with the flag on for a file-backed pool");

        let (started_tx, started_rx) = tokio::sync::oneshot::channel::<()>();
        let (release_tx, release_rx) = tokio::sync::oneshot::channel::<()>();
        let occupier = {
            let writer_task = writer_task.clone();
            tokio::spawn(async move {
                writer_task
                    .send(move |_conn| {
                        let _ = started_tx.send(());
                        let _ = release_rx.blocking_recv();
                        Ok::<(), khive_storage::StorageError>(())
                    })
                    .await
            })
        };

        started_rx
            .await
            .expect("occupier must signal it has started running inside the writer task");
        assert_eq!(
            writer_task.queue_depth(),
            0,
            "channel must start empty once the occupier has been dequeued and is running"
        );

        let record_task = tokio::spawn(async move {
            record_serve(
                sql.as_ref(),
                "write-queue-routing-row",
                "local",
                "recall",
                Some("profile-a"),
                None,
                None,
                "write-queue-routing-target",
                "write-queue-routing-class",
                "write queue routing test query",
                1_700_000_000_000_000,
            )
            .await
        });

        let mut saw_enqueued = false;
        for _ in 0..100 {
            if writer_task.queue_depth() >= 1 {
                saw_enqueued = true;
                break;
            }
            tokio::time::sleep(std::time::Duration::from_millis(5)).await;
        }
        assert!(
            saw_enqueued,
            "record_serve's write request never appeared in the writer task's \
             channel while the occupier held the single drain slot — \
             SqliteWriter::execute is not routing this single-row write \
             through the shared writer task"
        );

        release_tx
            .send(())
            .expect("occupier must still be waiting on the release signal");
        occupier
            .await
            .expect("occupier task must not panic")
            .expect("occupier write must succeed");

        let inserted = record_task
            .await
            .expect("record_task must not panic")
            .expect("record_serve must succeed once unblocked");
        assert!(inserted, "record_serve must report a fresh row insert");
    }
}