agent-block-core 0.37.1

//! `std.ts.*` — SQLite-backed time-series primitive.
//!
//! This is the only storage bridge whose implementation is in-tree
//! (mlua_batteries provides no TSDB module). Append / query / last all live in
//! this file; the DDL below is run by the host, on the connection thread,
//! before the VM is handed the isle.
//!
//! Backend: single `ts` table in `ts.sqlite` (or `:memory:`).
//!
//! Schema:
//! ```sql
//! CREATE TABLE IF NOT EXISTS ts (
//!     series TEXT NOT NULL,
//!     ts     INTEGER NOT NULL,
//!     tags   TEXT,
//!     value  TEXT NOT NULL
//! );
//! CREATE INDEX IF NOT EXISTS idx_ts_series_ts ON ts(series, ts);
//! ```
//!
//! Column notes:
//! - `series`: logical stream name (e.g. `"cpu_load"`, `"agent_events"`)
//! - `ts`: Unix timestamp in milliseconds (i64)
//! - `tags`: JSON object (`{"task": "X", "phase": "Y"}`) or NULL; filtered
//!   via `json_extract` in queries — never compared as a serialised string
//! - `value`: JSON-encoded payload; accepts both JSON numbers and JSON objects
//!   so that callers can append plain numeric metrics or structured MCP
//!   envelope payloads without loss (dual-type contract, Crux §3.8 C1)
//!
//! # The connection lives on a thread of its own, and the VM never waits on it
//!
//! The bridge holds no connection and takes no lock. What it holds is a
//! [`rusqlite_isle::AsyncIsle`] — a handle to the thread that owns the
//! connection, the same shape the kernel's event store sits on. A `std.ts`
//! call is therefore a closure sent to that thread and a result *awaited*:
//! every one of them is an `mlua` async function, so the Lua VM yields while
//! SQLite works and goes on driving its other coroutines.
//!
//! `std.sql` and `std.kv` reach the same place by the other route: their
//! connection is host-owned and shared (`Arc<Mutex<_>>`), and each statement
//! runs in `tokio::task::spawn_blocking` with the lock taken inside that
//! closure. Both satisfy the rule this file's tests are written against — the
//! VM thread hands the work off and yields — and the difference is only where
//! the waiting happens: a connection thread here, the blocking pool there.
//!
//! Argument validation (tag keys, `bucket_ms` / `limit` / `offset` bounds,
//! JSON encoding of values) is CPU work over Lua values and stays on the VM
//! thread, where the Lua state is; only the statement crosses to the isle.
//!
//! The isle is opened in `host.rs` — path, busy timeout, `journal_mode` and
//! [`SCHEMA_DDL`] are the host's, applied on the connection thread before the
//! isle takes its first job — and its driver is drained at host shutdown
//! beside the `sql` / `kv` ones. See `bridge/config.rs` for the ENV → path
//! mapping (`AGENT_BLOCK_TS_PATH`).

use std::time::{SystemTime, UNIX_EPOCH};

use mlua::prelude::*;
use rusqlite_isle::AsyncIsle;

use crate::bridge::{json_to_lua, lua_to_json};

/// The `ts` table and its `(series, ts)` index.
///
/// `IF NOT EXISTS` throughout, so opening a fresh database and reopening one an
/// earlier run wrote take the same path. Run by the host in the isle's init
/// closure: it is one statement batch that waits on SQLite, and waiting is the
/// connection thread's business, not the VM's.
pub const SCHEMA_DDL: &str = "CREATE TABLE IF NOT EXISTS ts \
     (series TEXT NOT NULL, ts INTEGER NOT NULL, \
      tags TEXT, value TEXT NOT NULL); \
     CREATE INDEX IF NOT EXISTS idx_ts_series_ts ON ts(series, ts);";

// ── helpers ──────────────────────────────────────────────────────────────────

/// Validate that every tag key contains only ASCII alphanumeric characters or
/// underscores, guarding against SQL injection via format-string tag paths.
///
/// # Arguments
///
/// - `key`: the tag key to validate
///
/// # Errors
///
/// Returns a `LuaError` if `key` contains any character outside `[a-zA-Z0-9_]`.
fn validate_tag_key(key: &str) -> LuaResult<()> {
    if key.chars().all(|c| c.is_ascii_alphanumeric() || c == '_') {
        Ok(())
    } else {
        Err(LuaError::external(
            "ts tag key must be [a-zA-Z0-9_]+".to_string(),
        ))
    }
}

/// Collect a Lua tag table into `(key, comparison-string)` pairs.
///
/// Runs on the VM thread: it reads Lua values, so it cannot travel with the
/// job. Keys are validated here for the same reason — the check is the reason
/// the key may be interpolated into a `json_extract` path at all.
///
/// # Errors
///
/// Returns `LuaError` if a key fails [`validate_tag_key`] or a value cannot be
/// converted to JSON.
fn collect_tag_filter(lua: &Lua, tbl: LuaTable) -> LuaResult<Vec<(String, String)>> {
    let mut pairs = Vec::new();
    for p in tbl.pairs::<String, LuaValue>() {
        let (k, v) = p?;
        validate_tag_key(&k)?;
        // Encode tag value as a JSON string for comparison.
        let v_json = lua_to_json(lua, v).map_err(LuaError::external)?;
        let v_str = match &v_json {
            serde_json::Value::String(s) => s.clone(),
            other => serde_json::to_string(other).map_err(LuaError::external)?,
        };
        pairs.push((k, v_str));
    }
    Ok(pairs)
}

/// Build the SQL query string for `std.ts.query`.
///
/// Constructs a parameterised SQL string for one of three query shapes:
/// - **raw** (`agg = None`): `SELECT ts, value, tags … ORDER BY ts, rowid LIMIT ? OFFSET ?`
/// - **single-aggregate** (`agg = Some(_)`, `bucket_ms = None`):
///   `SELECT <AGG_EXPR> FROM ts WHERE …` (single row, no LIMIT/OFFSET)
/// - **time-bucketed** (`agg = Some(_)`, `bucket_ms = Some(_)`):
///   `SELECT (ts/?)*? AS bucket_ts, <AGG_EXPR> … GROUP BY bucket_ts … LIMIT ? OFFSET ?`
///
/// The returned string uses positional `?` placeholders. The binding order is:
/// `series, from_ts, to_ts, [tag_values…], [bucket_ms, bucket_ms], [limit, offset]`.
///
/// # Arguments
///
/// - `agg`: optional aggregation function name (`"count"`, `"sum"`, `"avg"`, `"last"`)
/// - `bucket_ms`: optional bucket width in milliseconds (> 0)
/// - `tag_keys`: ordered list of tag keys for the AND-filter; paths become
///   `json_extract(tags, '$.<key>')` placeholders
/// - `limit`: optional maximum row count (`>= 0`)
/// - `offset`: optional row skip count (`>= 0`)
///
/// # Errors
///
/// Returns `Err(String)` for an unrecognised aggregation function name.
fn build_query_sql(
    agg: Option<&str>,
    bucket_ms: Option<i64>,
    tag_keys: &[String],
    limit: Option<i64>,
    offset: Option<i64>,
) -> Result<String, String> {
    // Build the shared WHERE clause fragment (after series / ts range filter).
    // Each tag key adds one `AND json_extract(tags, '$.<key>') = ?` clause.
    // This implements the Crux C2 conjunction contract: every k/v pair in
    // opts.tags is evaluated independently via json_extract, never as a
    // single serialised-string equality match.
    let tag_clauses: String = tag_keys
        .iter()
        .map(|k| format!(" AND json_extract(tags, '$.{k}') = ?"))
        .collect();

    let where_clause = format!("WHERE series = ? AND ts >= ? AND ts <= ?{tag_clauses}");

    // Helper to append LIMIT / OFFSET fragments (not used in single-agg mode).
    let limit_clause = match (limit, offset) {
        (Some(l), Some(o)) => format!(" LIMIT {l} OFFSET {o}"),
        (Some(l), None) => format!(" LIMIT {l}"),
        (None, Some(o)) => format!(" LIMIT -1 OFFSET {o}"),
        (None, None) => String::new(),
    };

    match agg {
        // ── path 1: raw rows ──────────────────────────────────────────────
        None => {
            let sql = format!(
                "SELECT ts, value, tags FROM ts {where_clause} ORDER BY ts, rowid{limit_clause}"
            );
            Ok(sql)
        }

        // ── path 2 / 3: aggregate ─────────────────────────────────────────
        Some(agg_name) => {
            // Build the aggregate expression.  Note that agg="last" is
            // special: it is not a SQL aggregate function but an ORDER+LIMIT
            // operation.  For the time-bucketed case (path 3) we use MAX(ts)
            // per bucket to identify the latest row, which requires the
            // caller to do a second fetch (or we use a subquery).  For
            // simplicity the bucketed "last" uses MAX(ts) as a proxy for the
            // last timestamp — callers needing the actual value should use
            // a separate query.  For single-agg "last" we use a full
            // ORDER BY ts DESC LIMIT 1 subquery form.
            let agg_expr: &str = match agg_name {
                "count" => "COUNT(*)",
                "sum" => "SUM(CAST(value AS REAL))",
                "avg" => "AVG(CAST(value AS REAL))",
                "last" => {
                    // handled specially per path below
                    "last"
                }
                other => return Err(format!("unknown agg: {other}")),
            };

            match bucket_ms {
                // ── path 2: single aggregate (no bucket) ─────────────────
                None => {
                    if agg_name == "last" {
                        // agg="last" + no bucket → ORDER BY ts DESC LIMIT 1
                        let sql = format!(
                            "SELECT value, tags, ts FROM ts {where_clause} ORDER BY ts DESC, rowid DESC LIMIT 1"
                        );
                        Ok(sql)
                    } else {
                        let sql = format!("SELECT {agg_expr} FROM ts {where_clause}");
                        Ok(sql)
                    }
                }

                // ── path 3: time-bucketed aggregate ───────────────────────
                Some(bms) => {
                    // bucket_ts = (ts / bucket_ms) * bucket_ms  (integer division)
                    // The bucket_ms literal is embedded directly into the SQL
                    // string (safe: validated positive i64, no user-supplied
                    // string content).  Embedding avoids the parameter-ordering
                    // issue that arises when SELECT-clause `?` placeholders
                    // appear before WHERE-clause `?` placeholders — SQLite
                    // positional binding fills them left-to-right, so any `?`
                    // in the SELECT would consume series/from_ts/to_ts params.
                    let bucketed_agg_expr = if agg_name == "last" {
                        // For bucketed "last" we return the maximum ts in
                        // each bucket as a proxy for the last value.
                        // CAST(value AS REAL) would not be meaningful here;
                        // instead we expose MAX(ts) for the bucket boundary.
                        "MAX(ts)".to_string()
                    } else {
                        agg_expr.to_string()
                    };

                    let sql = format!(
                        "SELECT (ts / {bms}) * {bms} AS bucket_ts, {bucketed_agg_expr} AS agg_value \
                         FROM ts {where_clause} \
                         GROUP BY bucket_ts ORDER BY bucket_ts{limit_clause}"
                    );
                    Ok(sql)
                }
            }
        }
    }
}

// ── registration ─────────────────────────────────────────────────────────────

/// Register the `std.ts` bridge into `lua`.
///
/// Registration itself waits on nothing: the table it installs holds three
/// async functions closing over `isle`, and the DDL has already run on the
/// connection thread by the time the host has an isle to pass here.
///
/// 1. Installs `std.ts.append`, `std.ts.query`, and `std.ts.last` as async
///    Lua functions.
/// 2. Loads `ts_tools.lua` to provide `std.ts.register_tools`.
///
/// # Arguments
///
/// - `lua`: the Lua state to register into (main Isle or handler Isle)
/// - `isle`: handle to the thread that owns the `ts` connection
///
/// # Errors
///
/// Returns a `LuaError` if the `std` global is not a table, any `std.ts`
/// assignment fails, or `ts_tools.lua` fails to load.
pub fn register(lua: &Lua, isle: AsyncIsle) -> LuaResult<()> {
    // ── Build std.ts table ────────────────────────────────────────────────
    let ts_tbl = lua.create_table()?;

    // ── std.ts.append ─────────────────────────────────────────────────────
    ts_tbl.set("append", make_append(lua, isle.clone())?)?;

    // ── std.ts.query ──────────────────────────────────────────────────────
    ts_tbl.set("query", make_query(lua, isle.clone())?)?;

    // ── std.ts.last ───────────────────────────────────────────────────────
    ts_tbl.set("last", make_last(lua, isle)?)?;

    // ── Install into std global ───────────────────────────────────────────
    let std_table: LuaTable = lua.globals().get("std")?;
    std_table.set("ts", ts_tbl)?;

    // ── Load ts_tools.lua (std.ts.register_tools) ─────────────────────────
    lua.load(include_str!("ts_tools.lua"))
        .set_name("std.ts.register_tools")
        .exec()?;

    Ok(())
}

// ── append ────────────────────────────────────────────────────────────────────

/// Create the `std.ts.append(series, value, tags?, at?)` async function.
///
/// The row is written inside the isle's `IMMEDIATE` transaction
/// ([`AsyncIsle::call_write`]), so a contended write waits out the busy
/// timeout on the connection thread while the VM goes on running.
///
/// # Arguments
///
/// - `lua`: the Lua state
/// - `isle`: handle to the connection thread
///
/// # Errors
///
/// Returns `LuaError` on tag key validation failure, JSON encode error, a
/// failed insert, or a closed / cancelled isle.
fn make_append(lua: &Lua, isle: AsyncIsle) -> LuaResult<LuaFunction> {
    lua.create_async_function(
        move |lua, (series, value, tags, at): (String, LuaValue, Option<LuaTable>, Option<i64>)| {
            let isle = isle.clone();
            async move {
                tracing::trace!(series = %series, "ts.append");

                // ── resolve timestamp ─────────────────────────────────────
                // unwrap_or_default: UNIX_EPOCH-before fallback → Duration::ZERO.
                // This is a safe fallback path, not an unguarded unwrap.
                let ts_ms = at.unwrap_or_else(|| {
                    SystemTime::now()
                        .duration_since(UNIX_EPOCH)
                        .unwrap_or_default()
                        .as_millis() as i64
                });

                // ── encode value before handing the job to the isle ───────
                // lua_to_json requires &Lua (Lua VM access) and the job runs
                // on the connection thread.  Serialise to String here, on the
                // VM thread, then move the String into the closure.
                let value_json = lua_to_json(&lua, value).map_err(LuaError::external)?;
                let value_str = serde_json::to_string(&value_json).map_err(LuaError::external)?;

                // ── encode tags ───────────────────────────────────────────
                let tags_str: Option<String> = match tags {
                    None => None,
                    Some(tbl) => {
                        // Validate all tag keys before encoding.
                        for pair in tbl.clone().pairs::<String, LuaValue>() {
                            let (k, _) = pair?;
                            validate_tag_key(&k)?;
                        }
                        let tags_json =
                            lua_to_json(&lua, LuaValue::Table(tbl)).map_err(LuaError::external)?;
                        Some(serde_json::to_string(&tags_json).map_err(LuaError::external)?)
                    }
                };

                // ── the insert, on the connection thread ──────────────────
                // The inner `Result<_, String>` carries the statement's own
                // message; the outer one is the isle's (closed, cancelled,
                // panicked), which is why the two are unwrapped separately.
                let result: Result<(), String> = isle
                    .call_write(move |tx| {
                        Ok(tx
                            .execute(
                                "INSERT INTO ts (series, ts, tags, value) VALUES (?1, ?2, ?3, ?4)",
                                rusqlite::params![series, ts_ms, tags_str, value_str],
                            )
                            .map(|_| ())
                            .map_err(|e| format!("ts append: {e}")))
                    })
                    .await
                    .map_err(|e| LuaError::external(format!("ts task: {e}")))?;

                result.map_err(|e| {
                    tracing::warn!(error = %e, "ts append failed");
                    LuaError::external(e)
                })?;

                Ok(LuaValue::Nil)
            }
        },
    )
}

// ── query ─────────────────────────────────────────────────────────────────────

/// Create the `std.ts.query(series, opts)` async function.
///
/// `opts` fields (all optional):
/// - `from` (integer): start timestamp ms, default `i64::MIN`
/// - `to` (integer): end timestamp ms, default `i64::MAX`
/// - `tags` (table): AND-filter; each k/v pair becomes a `json_extract` clause
/// - `agg` (string): aggregation — `"count"` | `"sum"` | `"avg"` | `"last"`
/// - `bucket_ms` (integer > 0): bucket width; requires `agg`
/// - `limit` (integer >= 0): maximum result rows
/// - `offset` (integer >= 0): result rows to skip
///
/// Returns a Lua array of row tables:
/// - raw mode: `{ ts, value, tags }`
/// - single-agg (agg, no bucket): `{ value }` (scalar result)
/// - bucketed-agg: `{ bucket_ts, value }`
///
/// Note: `sum`/`avg` treat `value` as a JSON number via `CAST(value AS REAL)`.
/// Rows whose `value` is a JSON object produce `0.0` in SQLite's CAST — prefer
/// number-only series when using `sum`/`avg`.
///
/// # Errors
///
/// Returns `LuaError` on validation failure, a failed statement, or a closed /
/// cancelled isle.
fn make_query(lua: &Lua, isle: AsyncIsle) -> LuaResult<LuaFunction> {
    lua.create_async_function(move |lua, (series, opts): (String, Option<LuaTable>)| {
        let isle = isle.clone();
        async move {
            tracing::trace!(series = %series, "ts.query");

            // ── parse opts ────────────────────────────────────────────
            let from_ts: i64 = opts
                .as_ref()
                .and_then(|t| t.get::<Option<i64>>("from").ok().flatten())
                .unwrap_or(i64::MIN);
            let to_ts: i64 = opts
                .as_ref()
                .and_then(|t| t.get::<Option<i64>>("to").ok().flatten())
                .unwrap_or(i64::MAX);

            let agg: Option<String> = opts
                .as_ref()
                .and_then(|t| t.get::<Option<String>>("agg").ok().flatten());

            let bucket_ms: Option<i64> = opts
                .as_ref()
                .and_then(|t| t.get::<Option<i64>>("bucket_ms").ok().flatten());

            let limit: Option<i64> = opts
                .as_ref()
                .and_then(|t| t.get::<Option<i64>>("limit").ok().flatten());

            let offset: Option<i64> = opts
                .as_ref()
                .and_then(|t| t.get::<Option<i64>>("offset").ok().flatten());

            // ── validate opts ─────────────────────────────────────────
            if let Some(bms) = bucket_ms {
                if bms <= 0 {
                    return Err(LuaError::external(
                        "ts bucket_ms must be positive".to_string(),
                    ));
                }
                if agg.is_none() {
                    return Err(LuaError::external("ts bucket_ms requires agg".to_string()));
                }
            }
            if let Some(l) = limit {
                if l < 0 {
                    return Err(LuaError::external("ts opts.limit must be >= 0".to_string()));
                }
            }
            if let Some(o) = offset {
                if o < 0 {
                    return Err(LuaError::external(
                        "ts opts.offset must be >= 0".to_string(),
                    ));
                }
            }

            // ── extract and validate tags filter ──────────────────────
            // Tags k/v pairs are collected into (key, json_string) pairs on
            // the VM thread, where the Lua values are.
            let tags_filter: Vec<(String, String)> = match opts
                .as_ref()
                .and_then(|t| t.get::<Option<LuaTable>>("tags").ok().flatten())
            {
                None => vec![],
                Some(tbl) => collect_tag_filter(&lua, tbl)?,
            };

            let tag_keys: Vec<String> = tags_filter.iter().map(|(k, _)| k.clone()).collect();
            let tag_values: Vec<String> = tags_filter.into_iter().map(|(_, v)| v).collect();

            // ── build SQL ─────────────────────────────────────────────
            let agg_ref = agg.as_deref();
            let sql = build_query_sql(agg_ref, bucket_ms, &tag_keys, limit, offset)
                .map_err(LuaError::external)?;

            // ── execute on the connection thread ──────────────────────
            let is_single_agg = agg.is_some() && bucket_ms.is_none();
            let is_last_single = agg.as_deref() == Some("last") && bucket_ms.is_none();
            let is_bucketed = agg.is_some() && bucket_ms.is_some();

            let rows_raw: Result<Vec<Vec<Option<String>>>, String> = isle
                .call(move |conn| {
                    Ok((|| -> Result<Vec<Vec<Option<String>>>, String> {
                        let mut stmt = conn
                            .prepare(&sql)
                            .map_err(|e| format!("ts query prepare: {e}"))?;

                        // Build the parameter list dynamically.
                        // Order: series, from_ts, to_ts, [tag_values…]
                        let mut params: Vec<Box<dyn rusqlite::ToSql>> =
                            vec![Box::new(series), Box::new(from_ts), Box::new(to_ts)];
                        for v in tag_values {
                            params.push(Box::new(v));
                        }
                        // Note: bucket_ms is embedded as a literal in the SQL
                        // (see build_query_sql path 3), so no additional params
                        // are needed for the bucketed-aggregate case.

                        let param_refs: Vec<&dyn rusqlite::ToSql> =
                            params.iter().map(|p| p.as_ref()).collect();

                        let col_count = stmt.column_count();
                        let rows: Vec<Vec<Option<String>>> = stmt
                            .query(param_refs.as_slice())
                            .map_err(|e| format!("ts query exec: {e}"))?
                            .mapped(|row| {
                                let mut cols = Vec::with_capacity(col_count);
                                for i in 0..col_count {
                                    // Use Value (not String) to handle INTEGER and REAL columns.
                                    // rusqlite's FromSql for String only accepts Text/Blob and
                                    // returns FromSqlError::InvalidType for INTEGER or REAL values
                                    // (e.g. the `ts` column, COUNT(*), SUM, AVG, bucket_ts).
                                    let v: rusqlite::types::Value =
                                        row.get::<_, rusqlite::types::Value>(i)?;
                                    let s = match v {
                                        rusqlite::types::Value::Null => None,
                                        rusqlite::types::Value::Integer(n) => Some(n.to_string()),
                                        rusqlite::types::Value::Real(f) => Some(f.to_string()),
                                        rusqlite::types::Value::Text(s) => Some(s),
                                        rusqlite::types::Value::Blob(_) => None,
                                    };
                                    cols.push(s);
                                }
                                Ok(cols)
                            })
                            .collect::<Result<_, _>>()
                            .map_err(|e| format!("ts query row: {e}"))?;

                        Ok(rows)
                    })())
                })
                .await
                .map_err(|e| LuaError::external(format!("ts task: {e}")))?;

            let rows_raw = rows_raw.map_err(|e| {
                tracing::warn!(error = %e, "ts query failed");
                LuaError::external(e)
            })?;

            // ── decode rows into Lua ───────────────────────────────────
            // Column layout depends on the query path:
            //   raw:         [ts(i64), value(text), tags(text|null)]
            //   single-agg:  [agg_value(text|null)]  (or [value,tags,ts] for last)
            //   bucketed:    [bucket_ts(i64), agg_value(text|null)]
            let result_table = lua.create_table()?;

            if is_last_single {
                // path 2 agg=last: columns are [value, tags, ts]
                // Returns at most 1 row; wrap as single-element array for
                // consistency with other agg modes.
                for (idx, row) in rows_raw.iter().enumerate() {
                    let row_tbl = lua.create_table()?;
                    // col 0: value TEXT
                    let value_lv = decode_value_col(&lua, row.first().and_then(|s| s.as_deref()))?;
                    row_tbl.set("value", value_lv)?;
                    // col 1: tags TEXT|NULL
                    let tags_lv = decode_tags_col(&lua, row.get(1).and_then(|s| s.as_deref()))?;
                    row_tbl.set("tags", tags_lv)?;
                    // col 2: ts INTEGER
                    let ts_lv = if let Some(Some(s)) = row.get(2) {
                        let n: i64 = s.parse().map_err(LuaError::external)?;
                        LuaValue::Integer(n)
                    } else {
                        LuaValue::Nil
                    };
                    row_tbl.set("ts", ts_lv)?;
                    result_table.set(idx + 1, row_tbl)?;
                }
            } else if is_single_agg {
                // path 2 (non-last): single column [agg_value]
                for (idx, row) in rows_raw.iter().enumerate() {
                    let row_tbl = lua.create_table()?;
                    let agg_lv = decode_value_col(&lua, row.first().and_then(|s| s.as_deref()))?;
                    row_tbl.set("value", agg_lv)?;
                    result_table.set(idx + 1, row_tbl)?;
                }
            } else if is_bucketed {
                // path 3: columns [bucket_ts, agg_value]
                for (idx, row) in rows_raw.iter().enumerate() {
                    let row_tbl = lua.create_table()?;
                    // col 0: bucket_ts INTEGER (from (ts/?)*? expression)
                    let bts_lv = if let Some(Some(s)) = row.first() {
                        let n: i64 = s.parse().map_err(LuaError::external)?;
                        LuaValue::Integer(n)
                    } else {
                        LuaValue::Nil
                    };
                    row_tbl.set("bucket_ts", bts_lv)?;
                    // col 1: agg_value
                    let agg_lv = decode_value_col(&lua, row.get(1).and_then(|s| s.as_deref()))?;
                    row_tbl.set("value", agg_lv)?;
                    result_table.set(idx + 1, row_tbl)?;
                }
            } else {
                // path 1: raw rows [ts, value, tags]
                for (idx, row) in rows_raw.iter().enumerate() {
                    let row_tbl = lua.create_table()?;
                    // col 0: ts INTEGER
                    let ts_lv = if let Some(Some(s)) = row.first() {
                        let n: i64 = s.parse().map_err(LuaError::external)?;
                        LuaValue::Integer(n)
                    } else {
                        LuaValue::Nil
                    };
                    row_tbl.set("ts", ts_lv)?;
                    // col 1: value TEXT
                    let value_lv = decode_value_col(&lua, row.get(1).and_then(|s| s.as_deref()))?;
                    row_tbl.set("value", value_lv)?;
                    // col 2: tags TEXT|NULL
                    let tags_lv = decode_tags_col(&lua, row.get(2).and_then(|s| s.as_deref()))?;
                    row_tbl.set("tags", tags_lv)?;
                    result_table.set(idx + 1, row_tbl)?;
                }
            }

            Ok(LuaValue::Table(result_table))
        }
    })
}

// ── last ──────────────────────────────────────────────────────────────────────

/// Create the `std.ts.last(series, tags?)` async function.
///
/// Returns the most-recent data point for `series` (optionally filtered by
/// `tags` using the same AND-conjunction as `std.ts.query`).
///
/// Return value:
/// - `nil` — no matching row found
/// - `{ ts = <i64>, value = <decoded>, tags = <table or nil> }` — latest row
///
/// # Arguments
///
/// - `lua`: the Lua state
/// - `isle`: handle to the connection thread
///
/// # Errors
///
/// Returns `LuaError` on tag key validation failure, a failed statement, or a
/// closed / cancelled isle.
fn make_last(lua: &Lua, isle: AsyncIsle) -> LuaResult<LuaFunction> {
    lua.create_async_function(move |lua, (series, tags): (String, Option<LuaTable>)| {
        let isle = isle.clone();
        async move {
            tracing::trace!(series = %series, "ts.last");

            // ── extract and validate tags filter ──────────────────────
            let tags_filter: Vec<(String, String)> = match tags {
                None => vec![],
                Some(tbl) => collect_tag_filter(&lua, tbl)?,
            };

            let tag_keys: Vec<String> = tags_filter.iter().map(|(k, _)| k.clone()).collect();
            let tag_values: Vec<String> = tags_filter.into_iter().map(|(_, v)| v).collect();

            // Build tag_clauses for WHERE.
            let tag_clauses: String = tag_keys
                .iter()
                .map(|k| format!(" AND json_extract(tags, '$.{k}') = ?"))
                .collect();

            let sql = format!(
                "SELECT ts, value, tags FROM ts \
                 WHERE series = ? AND ts >= ? AND ts <= ?{tag_clauses} \
                 ORDER BY ts DESC, rowid DESC LIMIT 1"
            );

            type LastRow = Option<(i64, String, Option<String>)>;
            let row_raw: Result<LastRow, String> = isle
                .call(move |conn| {
                    Ok((|| -> Result<LastRow, String> {
                        let mut stmt = conn
                            .prepare(&sql)
                            .map_err(|e| format!("ts last prepare: {e}"))?;

                        let mut params: Vec<Box<dyn rusqlite::ToSql>> =
                            vec![Box::new(series), Box::new(i64::MIN), Box::new(i64::MAX)];
                        for v in tag_values {
                            params.push(Box::new(v));
                        }
                        let param_refs: Vec<&dyn rusqlite::ToSql> =
                            params.iter().map(|p| p.as_ref()).collect();

                        let mut rows = stmt
                            .query(param_refs.as_slice())
                            .map_err(|e| format!("ts last query: {e}"))?;

                        if let Some(row) = rows.next().map_err(|e| format!("ts last row: {e}"))? {
                            let ts_val: i64 =
                                row.get(0).map_err(|e| format!("ts last ts col: {e}"))?;
                            let value_str: String =
                                row.get(1).map_err(|e| format!("ts last value col: {e}"))?;
                            let tags_str: Option<String> =
                                row.get(2).map_err(|e| format!("ts last tags col: {e}"))?;
                            Ok(Some((ts_val, value_str, tags_str)))
                        } else {
                            Ok(None)
                        }
                    })())
                })
                .await
                .map_err(|e| LuaError::external(format!("ts task: {e}")))?;

            let row_opt = row_raw.map_err(|e| {
                tracing::warn!(error = %e, "ts last failed");
                LuaError::external(e)
            })?;

            match row_opt {
                None => Ok(LuaValue::Nil),
                Some((ts_val, value_str, tags_str)) => {
                    let row_tbl = lua.create_table()?;
                    row_tbl.set("ts", LuaValue::Integer(ts_val))?;

                    // Two-Phase decode: raw JSON string → serde_json::Value → LuaValue
                    let v_json: serde_json::Value =
                        serde_json::from_str(&value_str).map_err(LuaError::external)?;
                    let v_lv = json_to_lua(&lua, v_json)?;
                    row_tbl.set("value", v_lv)?;

                    let tags_lv = decode_tags_col(&lua, tags_str.as_deref())?;
                    row_tbl.set("tags", tags_lv)?;

                    Ok(LuaValue::Table(row_tbl))
                }
            }
        }
    })
}

// ── decode helpers ────────────────────────────────────────────────────────────

/// Decode a SQLite `value` column (TEXT storing JSON) into a `LuaValue`.
///
/// Implements the Two-Phase deserialization pattern (Outline K-NEW):
/// `String → serde_json::Value → LuaValue`.  A NULL column (`None`) or a
/// NULL aggregate result returns `LuaValue::Nil`.
///
/// # Arguments
///
/// - `lua`: the Lua state
/// - `raw`: the raw column string, or `None` for SQL NULL
///
/// # Errors
///
/// Returns `LuaError` if the string is not valid JSON.
fn decode_value_col(lua: &Lua, raw: Option<&str>) -> LuaResult<LuaValue> {
    match raw {
        None => Ok(LuaValue::Nil),
        Some(s) => {
            let v: serde_json::Value = serde_json::from_str(s).map_err(LuaError::external)?;
            json_to_lua(lua, v)
        }
    }
}

/// Decode a SQLite `tags` column (TEXT storing a JSON object, or NULL) into a
/// `LuaValue`.
///
/// NULL tags columns return `LuaValue::Nil` (row has no tags).
///
/// # Arguments
///
/// - `lua`: the Lua state
/// - `raw`: the raw column string, or `None` for SQL NULL
///
/// # Errors
///
/// Returns `LuaError` if the string is not valid JSON.
fn decode_tags_col(lua: &Lua, raw: Option<&str>) -> LuaResult<LuaValue> {
    decode_value_col(lua, raw)
}

#[cfg(test)]
mod tests {
    use super::*;
    use rusqlite::{params, Connection};

    /// Same-ms INSERT order is preserved by `ORDER BY ts, rowid` (raw path).
    ///
    /// Inserts three rows sharing the same `ts` value into an in-memory SQLite
    /// using the production DDL, runs the SQL produced by `build_query_sql`
    /// (raw mode), and verifies the returned values match INSERT order exactly.
    /// Also asserts that the generated SQL string contains the rowid tie-breaker.
    ///
    /// # Test categories
    ///
    /// - (T1) Happy path: normal INSERT + query flow with production DDL.
    /// - (T2) Edge case: all rows share identical `ts` millisecond value.
    /// - (T3) Regression guard: SQL string must contain `ORDER BY ts, rowid`.
    #[test]
    fn raw_path_same_ms_preserves_insert_order() {
        // 1) Assert generated SQL contains the tie-breaker.
        let sql = build_query_sql(None, None, &[], None, None).expect("build_query_sql");
        assert!(
            sql.contains("ORDER BY ts, rowid"),
            "raw path SQL missing rowid tie-breaker: {sql}"
        );

        // 2) Execute against in-memory SQLite using production DDL.
        // Safety: open_in_memory() only fails on internal SQLite allocation
        // errors which cannot occur in a controlled test environment.
        let conn = Connection::open_in_memory().expect("open in-memory sqlite");
        conn.execute_batch(SCHEMA_DDL).expect("ddl");

        // Insert three rows with identical ts=1000 ms.
        conn.execute(
            "INSERT INTO ts (series, ts, tags, value) VALUES (?, ?, NULL, ?)",
            params!["s", 1000_i64, "\"first\""],
        )
        .expect("insert 1");
        conn.execute(
            "INSERT INTO ts (series, ts, tags, value) VALUES (?, ?, NULL, ?)",
            params!["s", 1000_i64, "\"second\""],
        )
        .expect("insert 2");
        conn.execute(
            "INSERT INTO ts (series, ts, tags, value) VALUES (?, ?, NULL, ?)",
            params!["s", 1000_i64, "\"third\""],
        )
        .expect("insert 3");

        // Safety: prepare() only fails if SQL is malformed; sql is generated by
        // build_query_sql which is already tested above.
        let mut stmt = conn.prepare(&sql).expect("prepare");
        let rows: Vec<String> = stmt
            .query_map(params!["s", i64::MIN, i64::MAX], |r| r.get::<_, String>(1))
            .expect("query")
            .collect::<Result<Vec<_>, _>>()
            .expect("collect");

        assert_eq!(
            rows,
            vec![
                "\"first\"".to_string(),
                "\"second\"".to_string(),
                "\"third\"".to_string()
            ],
            "raw path returned rows in non-INSERT order: {rows:?}"
        );
    }

    /// Same-ms `last` returns the most recently INSERTed row.
    ///
    /// Verifies both the SQL string produced by `build_query_sql(Some("last"), ...)`
    /// contains `ORDER BY ts DESC, rowid DESC LIMIT 1`, and that executing an
    /// equivalent query against in-memory SQLite returns the last INSERT value
    /// even when all rows share the same `ts` ms.
    ///
    /// # Test categories
    ///
    /// - (T1) Happy path: last value retrieval with production DDL.
    /// - (T2) Edge case: all rows share identical `ts` millisecond value.
    /// - (T3) Regression guard: SQL string must contain `ORDER BY ts DESC, rowid DESC LIMIT 1`.
    #[test]
    fn last_path_same_ms_returns_last_insert() {
        // 1) Assert build_query_sql last form contains tie-breaker.
        let sql_last =
            build_query_sql(Some("last"), None, &[], None, None).expect("build_query_sql last");
        assert!(
            sql_last.contains("ORDER BY ts DESC, rowid DESC LIMIT 1"),
            "last path SQL missing rowid DESC tie-breaker: {sql_last}"
        );

        // 2) Execute the make_last-equivalent SQL against in-memory SQLite.
        // Safety: open_in_memory() only fails on internal SQLite allocation
        // errors which cannot occur in a controlled test environment.
        let conn = Connection::open_in_memory().expect("open in-memory sqlite");
        conn.execute_batch(SCHEMA_DDL).expect("ddl");

        for v in ["\"first\"", "\"second\"", "\"third\""] {
            conn.execute(
                "INSERT INTO ts (series, ts, tags, value) VALUES (?, ?, NULL, ?)",
                params!["s", 1000_i64, v],
            )
            .expect("insert");
        }

        // make_last inline SQL form (mirrors src/bridge/ts.rs post-fix).
        // This string is the expected form after the rowid tie-breaker fix;
        // if make_last's format! drifts, the behaviour test below will catch it.
        let make_last_sql = "SELECT ts, value, tags FROM ts \
             WHERE series = ? AND ts >= ? AND ts <= ? \
             ORDER BY ts DESC, rowid DESC LIMIT 1";

        // Safety: make_last_sql is a literal string, prepare() cannot fail here.
        let mut stmt = conn.prepare(make_last_sql).expect("prepare");
        let value: String = stmt
            .query_row(params!["s", i64::MIN, i64::MAX], |r| r.get::<_, String>(1))
            .expect("query_row");

        assert_eq!(
            value, "\"third\"",
            "last path returned non-last INSERT value: {value}"
        );
    }

    // -- the rule this round exists for -------------------------------------

    /// **A slow `std.ts.append` does not stop the VM.**
    ///
    /// This is the property the round is about, so it is asserted directly
    /// rather than inferred from the shape of the code: a second coroutine on
    /// the same Lua state goes on running — advancing a counter through an
    /// async function of its own — for the *whole* time an append is waiting
    /// on a write lock another connection is holding.
    ///
    /// Before this round the bridge held a `Mutex<Connection>` and locked it
    /// *on the VM thread* before handing the statement to `spawn_blocking`,
    /// and ran its DDL there too — so a contended write parked the VM and the
    /// ticker counted nothing until the lock was released. The shared
    /// connection is not what was wrong with that: `std.sql` and `std.kv` run
    /// on one, and pass this same property, because the lock is taken inside
    /// the blocking closure and the schema is the host's. What was wrong was
    /// waiting for either on the thread the VM runs on.
    ///
    /// # Test categories
    ///
    /// - (T1) Happy path: the append lands once the lock is released.
    /// - (T2) Concurrency: ticks keep landing while the write waits.
    #[test]
    fn a_slow_write_does_not_block_another_coroutine_on_the_same_vm() {
        use std::sync::atomic::{AtomicUsize, Ordering};
        use std::sync::Arc;
        use std::time::Duration;

        /// How long the blocker holds the write lock.
        const HELD: Duration = Duration::from_millis(300);
        /// How long each tick takes, so ~60 fit inside `HELD`.
        const TICK: Duration = Duration::from_millis(5);
        /// The floor the assertion uses.  Far below what should actually
        /// happen (~60), because the point is "the VM kept running", not a
        /// measurement of how fast it ran.
        const AT_LEAST: usize = 5;
        /// Long enough that the contended write waits the lock out rather
        /// than failing — the host's default is the same order.
        const BUSY: Duration = Duration::from_secs(5);

        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .expect("a runtime for the VM to yield into");

        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join("ts.sqlite");

        // The isle the host would open: the DDL runs on its thread, before
        // the VM is handed the handle.
        let (isle, driver) = rt.block_on(async {
            AsyncIsle::builder()
                .thread_name("ts-test")
                .busy_timeout(BUSY)
                .spawn(&path, |conn| conn.execute_batch(SCHEMA_DDL))
                .await
                .expect("open the ts isle")
        });

        let lua = Lua::new();
        lua.globals()
            .set("std", lua.create_table().expect("std table"))
            .expect("set std");
        register(&lua, isle).expect("register std.ts");

        // `tick()` waits like any async bridge function does; `ticks()` reads
        // the counter without waiting for anything.
        let ticks = Arc::new(AtomicUsize::new(0));
        let counter = Arc::clone(&ticks);
        let tick = lua
            .create_async_function(move |_, ()| {
                let counter = Arc::clone(&counter);
                async move {
                    tokio::time::sleep(TICK).await;
                    counter.fetch_add(1, Ordering::Relaxed);
                    Ok(())
                }
            })
            .expect("create tick");
        lua.globals().set("tick", tick).expect("set tick");
        let counter = Arc::clone(&ticks);
        let read_ticks = lua
            .create_function(move |_, ()| Ok(counter.load(Ordering::Relaxed)))
            .expect("create ticks");
        lua.globals().set("ticks", read_ticks).expect("set ticks");

        // A second connection holds the write lock for `HELD`, on a thread of
        // its own so the test can go on driving the VM.
        let (locked_tx, locked_rx) = std::sync::mpsc::channel();
        let blocker_path = path.clone();
        let blocker = std::thread::spawn(move || {
            let conn = Connection::open(&blocker_path).expect("open the blocker");
            conn.busy_timeout(HELD).expect("busy timeout");
            conn.execute_batch("BEGIN EXCLUSIVE")
                .expect("take the write lock");
            locked_tx.send(()).expect("announce the lock");
            std::thread::sleep(HELD);
            conn.execute_batch("ROLLBACK").expect("release the lock");
        });
        locked_rx.recv().expect("the lock was taken");

        // Two coroutines, driven together on the VM's runtime: one blocked on
        // the write, one counting.  `during` is the number of ticks that
        // landed while the append was waiting.
        let during: usize = rt.block_on(async {
            let writer = lua
                .load(
                    r#"
                    local before = ticks()
                    std.ts.append("slow", 1)
                    return ticks() - before
                "#,
                )
                .eval_async::<usize>();
            let ticker = lua.load(r#"for _ = 1, 200 do tick() end"#).exec_async();
            // Both futures poll the same Lua state on this one thread, which
            // is exactly what the VM's own LocalSet does with its coroutines.
            let (written, _ticked) = tokio::join!(writer, ticker);
            written.expect("the append eventually lands")
        });

        blocker.join().expect("the blocker thread");

        assert!(
            during >= AT_LEAST,
            "the VM stopped while the write was waiting: only {during} tick(s) ran"
        );

        // And the write itself landed once the lock was released.
        let value: i64 = rt.block_on(async {
            lua.load(r#"return std.ts.last("slow").value"#)
                .eval_async::<i64>()
                .await
                .expect("the slow append landed")
        });
        assert_eq!(value, 1, "the row the append wrote");

        drop(lua);
        rt.block_on(driver.shutdown())
            .expect("the connection thread shut down cleanly");
    }
}