Bounded CSV preview and execution receipts

// Bounded previews, streaming summaries and receipts over eight fake local rows.
// Requires the updated native Aner development runtime; no release binary is implied.
// Run from the directory containing examples/fixtures/data_execution/observations.csv.
import aner.data
import aner.units
import aner.math

fn close(actual: Float64, expected: Float64, tolerance: Float64 = 0.000000001) -> Bool {
    return math.abs(actual - expected) <= tolerance
}
fn verify(label: String, condition: Bool) -> Unit {
    print(label)
    print(condition)
}

var schema = data.schema()
schema = data.column(schema, "row_id", "String", false)
schema = schema.unit_column("temperature_f", "degF", true)
schema = data.column(schema, "signal", "Float64", true)
schema = data.column(schema, "cohort", "String", false)
let source = data.scan_csv("examples/fixtures/data_execution/observations.csv", schema, "NA")
    .convert_unit("temperature_f", "degC")
let execution: DataExecution = data.execution(timeout_ms: 30000)

// No rare row occurs within the four-record validation budget.
let rare = data.where_eq(source, "cohort", "rare")
let limited = data.bounded(data.with_execution(rare, execution),
    max_scanned_rows: 4, max_read_bytes: 0, timeout_ms: 30000)
let preview = data.preview(limited, max_rows: 2, max_bytes: 1048576, batch_rows: 2)
verify("Preview returns zero matches without scanning the tail", data.rows(preview) == 0 && data.scanned_rows(preview) == 4)
verify("Preview is explicitly incomplete at the source row bound", !data.complete(preview) && data.stop_reason(preview) == "scan_rows")
verify("Preview counters distinguish validated, selected and returned rows", data.selected_rows(preview) == 0 && data.returned_rows(preview) == 0)
verify("Receipt records prefix validation and read accounting", data.receipt(preview).contains("header_validated") && data.receipt(preview).contains("read_bytes") && data.read_bytes(preview) > 0)

// Sequential reuse starts a fresh run. A generous source bound permits proven EOF.
let full_plan = data.bounded(data.with_execution(source, execution),
    max_scanned_rows: 100, max_read_bytes: 0, timeout_ms: 30000)
let full = data.summarize(full_plan, max_bytes: 1048576, batch_rows: 2)
verify("Full summary validates all eight source rows", data.complete(full) && data.stop_reason(full) == "eof" && data.scanned_rows(full) == 8 && data.rows(full) == 8)
verify("Streaming summary retains aggregate rows without returning table rows", data.selected_rows(full) == 8 && data.returned_rows(full) == 0)
verify("Temperature and signal each retain one missing observation", data.valid(full, "temperature_f") == 7 && data.missing(full, "temperature_f") == 1 && data.valid(full, "signal") == 7 && data.missing(full, "signal") == 1)
verify("Primitive extrema and population spread are observed-value statistics", data.min(full, "signal") == 1.0 && data.max(full, "signal") == 8.0 && close(data.std(full, "signal"), math.sqrt(276.0) / 7.0))
verify("Temperature extrema remain Points and spread is an interval", type(data.unit_min(full, "temperature_f", "degC")) == "Point<degC>" && close(data.unit_min(full, "temperature_f", "degC").value(), 20.0) && close(data.unit_max(full, "temperature_f", "degC").value(), 32.0) && close(data.unit_std(full, "temperature_f", "delta_degC").value(), 4.0))
verify("Controller receives the latest run while preview retains its own receipt", data.scanned_rows(execution) == 8 && data.complete(execution) && data.scanned_rows(preview) == 4 && !data.complete(preview))
verify("Receipt memory is managed workspace and elapsed time is observational", data.peak_bytes(full) > 0 && data.peak_bytes(full) <= 1048576 && data.elapsed_ms(full) >= 0)

// Pre-cancellation can be demonstrated without executing a throwing terminal call.
let cancelled = data.execution(timeout_ms: 0)
data.cancel(cancelled)
verify("Cancellation is sticky before execution", data.is_cancelled(cancelled))
verify("An unused cancelled controller has no validated or returned rows", data.scanned_rows(cancelled) == 0 && data.returned_rows(cancelled) == 0 && data.receipt(cancelled).contains("stop_reason"))