Add bt datasets command for managing remote datasets

README.md

@@ -118,6 +118,7 @@ Remove-Item -Recurse -Force (Join-Path $env:APPDATA "bt") -ErrorAction SilentlyC
 | `bt sql`         | Run SQL queries against Braintrust                                 |
 | `bt view`        | View logs, traces, and spans                                       |
 | `bt projects`    | Manage projects (list, create, view, delete)                       |
+| `bt datasets`    | Manage remote datasets (list, create, update, view, delete)        |
 | `bt prompts`     | Manage prompts (list, view, delete)                                |
 | `bt sync`        | Synchronize project logs between Braintrust and local NDJSON files |
 | `bt self update` | Update bt in-place                                                 |
@@ -151,6 +152,22 @@ Use `--` to forward extra arguments to the eval file via `process.argv`:
 bt eval foo.eval.ts -- --description "Prod" --shard=1/4
 ```
 
+## `bt datasets`
+
+- `bt datasets` works directly against remote Braintrust datasets — no local `bt sync` artifact flow is required.
+- `bt datasets create my-dataset` — create an empty remote dataset in the current project.
+- `bt datasets create my-dataset --description "Dataset for smoke tests"` — create a dataset with a description.
+- `bt datasets create my-dataset --file records.jsonl` — create the remote dataset and seed it from a JSON/JSONL file.
+- `cat records.jsonl | bt datasets create my-dataset` — create the dataset and seed it from stdin.
+- `bt datasets create my-dataset --rows '[{"id":"case-1","input":{"text":"hi"},"expected":"hello"}]'` — create the dataset from inline JSON rows.
+- `bt datasets create my-dataset --rows '[{"input":{"text":"hi"},"expected":"hello"}]'` — create a dataset when rows do not include `id`; bt auto-generates deterministic record IDs.
+- `bt datasets update my-dataset --file records.jsonl` — deterministically upsert rows by stable record id.
+- `bt datasets add my-dataset --rows '[{"id":"case-2","input":{"text":"bye"},"expected":"goodbye"}]'` — alias for `update`.
+- `bt datasets refresh my-dataset --file records.jsonl --id-field metadata.case_id` — alias for `update` with explicit id path (fails if the dataset does not exist, and does not delete remote rows missing from the input).
+- `bt datasets view my-dataset` — show dataset metadata and row payloads; defaults to loading up to 200 rows. Use `--limit <N>` to adjust or `--all-rows` to load everything.
+- `update`/`add`/`refresh` require explicit stable IDs via `id` or `--id-field`.
+- Accepted top-level record fields are `id`, `input`, `expected`, `output`, `metadata`, and `tags` (plus the root field referenced by `--id-field`, if different).
+
 ## `bt sql`
 
 - Runs interactively on TTY by default.

src/datasets/api.rs

@@ -0,0 +1,253 @@
+use std::collections::HashSet;
+
+use anyhow::{bail, Result};
+use serde::{Deserialize, Serialize};
+use serde_json::{Map, Value};
+use urlencoding::encode;
+
+use crate::http::ApiClient;
+
+const MAX_DATASET_ROWS_PAGE_LIMIT: usize = 1000;
+const MAX_DATASET_ROWS_PAGES: usize = 10_000;
+const DATASET_ROWS_SINCE: &str = "1970-01-01T00:00:00Z";
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct Dataset {
+    pub id: String,
+    pub name: String,
+    #[serde(default)]
+    pub project_id: Option<String>,
+    #[serde(default)]
+    pub description: Option<String>,
+    #[serde(default)]
+    pub created: Option<String>,
+    #[serde(default)]
+    pub created_at: Option<String>,
+    #[serde(default)]
+    pub metadata: Option<serde_json::Value>,
+}
+
+pub type DatasetRow = Map<String, Value>;
+
+impl Dataset {
+    pub fn description_text(&self) -> Option<&str> {
+        self.description
+            .as_deref()
+            .filter(|description| !description.is_empty())
+            .or_else(|| {
+                self.metadata
+                    .as_ref()
+                    .and_then(|metadata| metadata.get("description"))
+                    .and_then(|description| description.as_str())
+                    .filter(|description| !description.is_empty())
+            })
+    }
+
+    pub fn created_text(&self) -> Option<&str> {
+        self.created
+            .as_deref()
+            .filter(|created| !created.is_empty())
+            .or_else(|| {
+                self.created_at
+                    .as_deref()
+                    .filter(|created| !created.is_empty())
+            })
+    }
+}
+
+#[derive(Debug, Deserialize)]
+struct ListResponse {
+    objects: Vec<Dataset>,
+}
+
+#[derive(Debug, Deserialize)]
+struct DatasetRowsResponse {
+    #[serde(default)]
+    data: Vec<DatasetRow>,
+    #[serde(default)]
+    cursor: Option<String>,
+}
+
+pub async fn list_datasets(client: &ApiClient, project_id: &str) -> Result<Vec<Dataset>> {
+    let path = format!(
+        "/v1/dataset?org_name={}&project_id={}",
+        encode(client.org_name()),
+        encode(project_id)
+    );
+    let list: ListResponse = client.get(&path).await?;
+    Ok(list.objects)
+}
+
+pub async fn get_dataset_by_name(
+    client: &ApiClient,
+    project_id: &str,
+    name: &str,
+) -> Result<Option<Dataset>> {
+    let datasets = list_datasets(client, project_id).await?;
+    Ok(datasets.into_iter().find(|dataset| dataset.name == name))
+}
+
+pub async fn list_dataset_rows(client: &ApiClient, dataset_id: &str) -> Result<Vec<DatasetRow>> {
+    let (rows, _truncated) = list_dataset_rows_limited(client, dataset_id, None).await?;
+    Ok(rows)
+}
+
+pub async fn list_dataset_rows_limited(
+    client: &ApiClient,
+    dataset_id: &str,
+    max_rows: Option<usize>,
+) -> Result<(Vec<DatasetRow>, bool)> {
+    if matches!(max_rows, Some(0)) {
+        return Ok((Vec::new(), false));
+    }
+
+    let mut rows = Vec::new();
+    let mut cursor: Option<String> = None;
+    let mut seen_cursors = HashSet::new();
+    let mut page_count = 0usize;
+    let mut truncated = false;
+
+    loop {
+        page_count += 1;
+        if page_count > MAX_DATASET_ROWS_PAGES {
+            bail!(
+                "dataset rows pagination exceeded {} pages for dataset '{}'",
+                MAX_DATASET_ROWS_PAGES,
+                dataset_id
+            );
+        }
+        if let Some(current_cursor) = cursor.as_ref() {
+            if !seen_cursors.insert(current_cursor.clone()) {
+                bail!(
+                    "dataset rows pagination loop detected for dataset '{}'",
+                    dataset_id
+                );
+            }
+        }
+
+        let query =
+            build_dataset_rows_query(dataset_id, MAX_DATASET_ROWS_PAGE_LIMIT, cursor.as_deref());
+        let body = serde_json::json!({
+            "query": query,
+            "fmt": "json",
+        });
+        let org_name = client.org_name();
+        let headers = if !org_name.is_empty() {
+            vec![("x-bt-org-name", org_name)]
+        } else {
+            Vec::new()
+        };
+        let response: DatasetRowsResponse =
+            client.post_with_headers("/btql", &body, &headers).await?;
+
+        let next_cursor = response.cursor.filter(|cursor| !cursor.is_empty());
+
+        if response.data.is_empty() {
+            if next_cursor.is_some() {
+                bail!(
+                    "dataset rows response for '{}' returned an empty page with a cursor",
+                    dataset_id
+                );
+            }
+            break;
+        }
+
+        if let Some(max_rows) = max_rows {
+            let remaining = max_rows.saturating_sub(rows.len());
+            if remaining == 0 {
+                truncated = true;
+                break;
+            }
+            if response.data.len() > remaining {
+                rows.extend(response.data.into_iter().take(remaining));
+                truncated = true;
+                break;
+            }
+        }
+
+        rows.extend(response.data);
+
+        match next_cursor {
+            Some(next_cursor) => {
+                if max_rows.is_some_and(|max_rows| rows.len() >= max_rows) {
+                    truncated = true;
+                    break;
+                }
+                cursor = Some(next_cursor);
+            }
+            None => break,
+        }
+    }
+
+    Ok((rows, truncated))
+}
+
+pub async fn create_dataset(
+    client: &ApiClient,
+    project_id: &str,
+    name: &str,
+    description: Option<&str>,
+) -> Result<Dataset> {
+    let mut body = serde_json::json!({
+        "name": name,
+        "project_id": project_id,
+        "org_name": client.org_name(),
+    });
+    if let Some(description) = description.filter(|description| !description.is_empty()) {
+        body["description"] = serde_json::Value::String(description.to_string());
+    }
+    client.post("/v1/dataset", &body).await
+}
+
+pub async fn delete_dataset(client: &ApiClient, dataset_id: &str) -> Result<()> {
+    let path = format!("/v1/dataset/{}", encode(dataset_id));
+    client.delete(&path).await
+}
+
+fn build_dataset_rows_query(dataset_id: &str, limit: usize, cursor: Option<&str>) -> String {
+    let cursor_clause = cursor
+        .map(|cursor| format!(" | cursor: {}", btql_quote(cursor)))
+        .unwrap_or_default();
+    format!(
+        "select: * | from: dataset({}) | filter: created >= {} | limit: {}{}",
+        sql_quote(dataset_id),
+        btql_quote(DATASET_ROWS_SINCE),
+        limit,
+        cursor_clause
+    )
+}
+
+fn btql_quote(value: &str) -> String {
+    serde_json::to_string(value)
+        .unwrap_or_else(|_| format!("\"{}\"", value.replace('\\', "\\\\").replace('"', "\\\"")))
+}
+
+fn sql_quote(value: &str) -> String {
+    format!("'{}'", value.replace('\'', "''"))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn dataset_rows_query_includes_required_filter() {
+        let query = build_dataset_rows_query("dataset-id", 1000, None);
+        assert_eq!(
+            query,
+            "select: * | from: dataset('dataset-id') | filter: created >= \"1970-01-01T00:00:00Z\" | limit: 1000"
+        );
+    }
+
+    #[test]
+    fn dataset_rows_query_quotes_cursor() {
+        let query = build_dataset_rows_query("dataset-id", 200, Some("cursor-123"));
+        assert!(query.contains("limit: 200 | cursor: \"cursor-123\""));
+    }
+
+    #[test]
+    fn dataset_rows_query_escapes_dataset_id() {
+        let query = build_dataset_rows_query("dataset'with-quote", 10, None);
+        assert!(query.contains("from: dataset('dataset''with-quote')"));
+    }
+}

src/datasets/create.rs

@@ -0,0 +1,88 @@
+use std::{path::Path, time::Duration};
+
+use anyhow::{bail, Result};
+use serde_json::json;
+
+use crate::ui::{print_command_status, with_spinner, with_spinner_visible, CommandStatus};
+
+use super::{api, records::load_optional_upload_records, upload, ResolvedContext};
+
+pub async fn run(
+    ctx: &ResolvedContext,
+    name: Option<&str>,
+    description: Option<&str>,
+    input_path: Option<&Path>,
+    inline_rows: Option<&str>,
+    id_field: &str,
+    json_output: bool,
+) -> Result<()> {
+    let name = upload::resolve_dataset_name(name, "create")?;
+
+    let exists = with_spinner(
+        "Checking dataset...",
+        api::get_dataset_by_name(&ctx.client, &ctx.project.id, &name),
+    )
+    .await?;
+    if exists.is_some() {
+        bail!(
+            "dataset '{name}' already exists in project '{}'; use `bt datasets update {name}` to add rows",
+            ctx.project.name
+        );
+    }
+
+    let records = load_optional_upload_records(input_path, inline_rows, id_field)?;
+    let uploaded = records.as_ref().map_or(0, |records| records.len());
+
+    let dataset = match with_spinner_visible(
+        "Creating dataset...",
+        api::create_dataset(&ctx.client, &ctx.project.id, &name, description),
+        Duration::from_millis(300),
+    )
+    .await
+    {
+        Ok(dataset) => dataset,
+        Err(error) => {
+            print_command_status(CommandStatus::Error, &format!("Failed to create '{name}'"));
+            return Err(error);
+        }
+    };
+
+    if let Some(records) = records.as_ref() {
+        if let Err(error) = upload::submit_prepared_records(
+            ctx,
+            &dataset.id,
+            records,
+            "Uploading dataset rows...",
+            "dataset upload failed",
+        )
+        .await
+        {
+            print_command_status(
+                CommandStatus::Error,
+                &format!("Created '{name}' but failed to upload initial rows"),
+            );
+            return Err(error);
+        }
+    }
+
+    if json_output {
+        println!(
+            "{}",
+            serde_json::to_string(&json!({
+                "dataset": dataset,
+                "created_dataset": true,
+                "uploaded": uploaded,
+                "mode": "create",
+            }))?
+        );
+        return Ok(());
+    }
+
+    let detail = if uploaded == 0 {
+        format!("Successfully created '{name}'")
+    } else {
+        format!("Created '{}' and uploaded {} records.", name, uploaded)
+    };
+    print_command_status(CommandStatus::Success, &detail);
+    Ok(())
+}