feat(appkit): managed eval datasets + turn semantics (stack 3/5)

apps/dev-playground/config/agents/query/evals/dataset.eval.ts

@@ -0,0 +1,53 @@
+import { defineEval, isJudgeConfigured } from "@databricks/appkit/beta";
+
+/**
+ * Dataset-driven eval: runs once per row of a Databricks managed evaluation
+ * dataset (a Unity Catalog table with `inputs`/`expectations` columns). The
+ * runner binds each row to `t.input`/`t.expected`.
+ *
+ * Run it (reading the dataset needs a workspace client + warehouse; judging the
+ * guidelines needs a judge model):
+ *   appkit agent eval dataset --root apps/dev-playground --url http://localhost:8000 \
+ *     --profile <profile> --warehouse <warehouse-id> --judge-model <endpoint>
+ *
+ * Row shape produced by the MLflow managed-dataset UI:
+ *   inputs        {"messages":[{"role":"user","content":"..."}]}
+ *   expectations  {"guidelines":{"value":["...","..."]}}   (optional)
+ */
+
+/** Pull the last user message out of an MLflow `{messages:[...]}` input. */
+function userMessage(input: Record<string, unknown>): string {
+  const messages = Array.isArray(input.messages)
+    ? (input.messages as Array<{ role?: string; content?: string }>)
+    : [];
+  const last = [...messages].reverse().find((m) => m.role === "user");
+  return last?.content ?? "";
+}
+
+/** Read `expectations.guidelines` — the UI wraps the array as `{value: [...]}`. */
+function guidelines(expected: Record<string, unknown> | undefined): string[] {
+  const g = (expected?.guidelines as { value?: unknown } | undefined)?.value;
+  return Array.isArray(g) ? g.map(String) : [];
+}
+
+export default defineEval({
+  description: "Query agent satisfies each dataset row's guidelines",
+  // Point at your own managed evaluation dataset (catalog.schema.table).
+  dataset: { table: "main.mario.appkit_eval_dataset" },
+  async test(t) {
+    // One turn per row. For a multi-turn conversation, call `t.send` again
+    // (same thread); to start an independent turn in the same test, `t.reset()`.
+    await t.send(userMessage(t.input));
+    t.succeeded();
+
+    // Each guideline is judged against the reply — gate by default, so a miss
+    // fails the eval (chain `.soft()` to only track it). Skipped cleanly when no
+    // judge model is configured, so the eval still exercises the dataset read +
+    // drive path without one.
+    if (isJudgeConfigured()) {
+      for (const guideline of guidelines(t.expected)) {
+        (await t.judge.closedQA(guideline)).atLeast(0.5);
+      }
+    }
+  },
+});

packages/appkit/src/connectors/mlflow/auth.ts

@@ -34,9 +34,8 @@ export async function resolveDatabricksAuth(
   }
 
   try {
-    const client = new WorkspaceClient(
-      options.profile ? { profile: options.profile } : {},
-    );
+    const client = resolveWorkspaceClient(options);
+    if (!client) return undefined;
     const headers = new Headers();
     // Mints the OAuth access token (or reuses a PAT from the profile) and adds
     // an `Authorization: Bearer <token>` header — the same call the connectors
@@ -53,3 +52,29 @@ export async function resolveDatabricksAuth(
     return undefined;
   }
 }
+
+/**
+ * Construct a Databricks `WorkspaceClient` for the eval runner — the object the
+ * SDK-backed connectors (e.g. `SQLWarehouseConnector`) take. An explicit
+ * host+token builds a PAT client; otherwise the profile (or ambient config) is
+ * used and the SDK resolves credentials, minting OAuth as needed. Returns
+ * `undefined` if construction throws (missing/invalid config).
+ */
+export function resolveWorkspaceClient(
+  options: ResolveDatabricksAuthOptions = {},
+): WorkspaceClient | undefined {
+  try {
+    if (options.host && options.token) {
+      return new WorkspaceClient({
+        host: options.host,
+        token: options.token,
+        authType: "pat",
+      });
+    }
+    return new WorkspaceClient(
+      options.profile ? { profile: options.profile } : {},
+    );
+  } catch {
+    return undefined;
+  }
+}

packages/appkit/src/connectors/mlflow/index.ts

@@ -2,5 +2,6 @@ export {
   type DatabricksAuth,
   type ResolveDatabricksAuthOptions,
   resolveDatabricksAuth,
+  resolveWorkspaceClient,
 } from "./auth";
 export { MlflowClient, normalizeHost, type PostResult } from "./client";

packages/appkit/src/evals/dataset.ts

@@ -0,0 +1,74 @@
+import type { WorkspaceClient } from "@databricks/sdk-experimental";
+import { SQLWarehouseConnector } from "../connectors";
+
+/**
+ * One row of a managed evaluation dataset. `inputs` are the kwargs passed to the
+ * agent for the turn; `expectations` (when present) is the row's ground truth /
+ * guidelines. Mirrors the `{inputs, expectations}` shape of `mlflow.genai`
+ * datasets and of the Unity Catalog table backing a managed eval dataset.
+ */
+export interface DatasetRow {
+  inputs: Record<string, unknown>;
+  expectations?: Record<string, unknown>;
+}
+
+export interface ReadEvalDatasetOptions {
+  /** Fully-qualified UC table: `catalog.schema.table`. */
+  table: string;
+  /** SQL warehouse id to run the read against. */
+  warehouseId: string;
+  /** Optional row cap. */
+  limit?: number;
+}
+
+/** A managed eval dataset is a UC table; only 3-level names are valid. */
+const UC_TABLE = /^[A-Za-z0-9_]+\.[A-Za-z0-9_]+\.[A-Za-z0-9_]+$/;
+
+/** Coerce a cell (already JSON-parsed by the connector for JSON columns) to a record. */
+function toRecord(value: unknown): Record<string, unknown> | undefined {
+  if (value && typeof value === "object" && !Array.isArray(value)) {
+    return value as Record<string, unknown>;
+  }
+  return undefined;
+}
+
+/**
+ * Read a Databricks managed evaluation dataset (a Unity Catalog table with
+ * `inputs`/`expectations` columns) into rows, over the public SQL Statement
+ * Execution API. Reuses {@link SQLWarehouseConnector} for submit/poll/transform
+ * — its result transform already JSON-parses string columns into objects, so
+ * `inputs`/`expectations` come back as records whether the table stores them as
+ * JSON strings or structs.
+ *
+ * The Python `mlflow.genai.datasets` API needs a Spark session (no TS
+ * equivalent), so we read the backing table directly.
+ */
+export async function readEvalDataset(
+  client: WorkspaceClient,
+  options: ReadEvalDatasetOptions,
+): Promise<DatasetRow[]> {
+  if (!UC_TABLE.test(options.table)) {
+    throw new Error(
+      `Invalid dataset table "${options.table}" — expected catalog.schema.table`,
+    );
+  }
+
+  const limit =
+    typeof options.limit === "number"
+      ? ` LIMIT ${Math.floor(options.limit)}`
+      : "";
+  const connector = new SQLWarehouseConnector({});
+  const response = await connector.executeStatement(client, {
+    warehouse_id: options.warehouseId,
+    statement: `SELECT inputs, expectations FROM ${options.table}${limit}`,
+  });
+
+  const rows =
+    (response.result as { data?: Array<Record<string, unknown>> } | undefined)
+      ?.data ?? [];
+
+  return rows.map((row) => ({
+    inputs: toRecord(row.inputs) ?? {},
+    expectations: toRecord(row.expectations),
+  }));
+}

packages/appkit/src/evals/http-driver.ts

@@ -73,6 +73,9 @@ export function createHttpDriver(options: HttpDriverOptions): EvalDriver {
   let threadId: string | undefined;
 
   return {
+    reset(): void {
+      threadId = undefined;
+    },
     async send(message: string): Promise<DriveResult> {
       let res: Response;
       try {

packages/appkit/src/evals/index.ts

@@ -5,7 +5,13 @@ export {
   type PostResult,
   type ResolveDatabricksAuthOptions,
   resolveDatabricksAuth,
+  resolveWorkspaceClient,
 } from "../connectors/mlflow";
+export {
+  type DatasetRow,
+  type ReadEvalDatasetOptions,
+  readEvalDataset,
+} from "./dataset";
 export { defineEval, defineEvalConfig } from "./define-eval";
 export { type DiscoveredEval, discoverEvalFiles } from "./discover";
 export { createHttpDriver, type HttpDriverOptions } from "./http-driver";

packages/appkit/src/evals/run-eval.ts

@@ -1,3 +1,4 @@
+import type { DatasetRow } from "./dataset";
 import { judgeClosedQA, judgeCustom, judgeFactuality } from "./judge";
 import type {
   AssertionHandle,
@@ -27,6 +28,8 @@ export interface RunEvalOptions {
   driver: EvalDriver;
   /** When true, soft assertion failures also fail the eval. */
   strict?: boolean;
+  /** Dataset row bound to `t.input`/`t.expected` for dataset-driven evals. */
+  row?: DatasetRow;
 }
 
 /**
@@ -70,22 +73,24 @@ export async function runEval(
         return handle;
       },
       atLeast(threshold: number) {
-        result.severity = "soft";
+        // Re-threshold the score; keep the current severity (gate unless the
+        // caller also chained `.soft()`).
         result.pass = (result.score ?? (result.pass ? 1 : 0)) >= threshold;
         return handle;
       },
     };
     return handle;
   };
 
-  // LLM-judge assertions are scored and soft by default; the caller chains
-  // `.atLeast(n)` to set the pass threshold or `.gate()` to promote.
+  // LLM-judge assertions are scored and gate by default (a miss fails the eval,
+  // like other assertions). The caller chains `.atLeast(n)` to change the pass
+  // threshold, or `.soft()` to demote to a tracked-only metric.
   const recordJudge = (
     label: string,
     score: number,
     rationale?: string,
   ): AssertionHandle =>
-    record(label, score >= DEFAULT_JUDGE_THRESHOLD, score, rationale).soft();
+    record(label, score >= DEFAULT_JUDGE_THRESHOLD, score, rationale);
 
   const t: TestContext = {
     async send(message) {
@@ -97,6 +102,9 @@ export async function runEval(
       lastSucceeded = r.succeeded;
       if (r.traceId) lastTraceId = r.traceId;
     },
+    reset() {
+      options.driver.reset?.();
+    },
     get reply() {
       return reply;
     },
@@ -106,6 +114,12 @@ export async function runEval(
     get sessionId() {
       return sessionId;
     },
+    get input() {
+      return options.row?.inputs ?? {};
+    },
+    get expected() {
+      return options.row?.expectations;
+    },
     succeeded() {
       return record(
         "succeeded",

packages/appkit/src/evals/run-evals.ts

@@ -1,5 +1,7 @@
 import { pathToFileURL } from "node:url";
+import type { WorkspaceClient } from "@databricks/sdk-experimental";
 import { MlflowClient } from "../connectors/mlflow";
+import { type DatasetRow, readEvalDataset } from "./dataset";
 import { discoverEvalFiles } from "./discover";
 import { createHttpDriver } from "./http-driver";
 import { configureJudge } from "./judge";
@@ -30,6 +32,13 @@ export interface RunEvalsOptions {
    * Databricks serving endpoint (`model`).
    */
   judge?: { host: string; token: string; model: string };
+  /**
+   * Workspace client used to read managed evaluation datasets (for evals that
+   * declare `dataset`). Required alongside {@link warehouseId} for those evals.
+   */
+  workspaceClient?: WorkspaceClient;
+  /** SQL warehouse id used to read managed evaluation datasets. */
+  warehouseId?: string;
   /** Wall-clock timestamp (ms) for run create/finish — pass `Date.now()`. */
   now?: number;
   /** Progress callback, invoked as evals are discovered, started, and finished. */
@@ -141,32 +150,87 @@ export async function runEvalsInDir(
   }
 
   const results: EvalResult[] = [];
+  // `total` counts eval files, not dataset rows: row counts aren't known until
+  // each file loads. Per-row detail is carried in the result id (`[row i/n]`).
   for (let index = 0; index < discovered.length; index++) {
     const d = discovered[index];
     const id = `${d.agent}/${d.id}`;
-    emit({ type: "start", id, index, total });
 
-    let result: EvalResult;
+    let def: EvalDefinition | undefined;
     try {
-      const def = await loadEval(d.file);
-      const driver = createHttpDriver({
-        baseUrl: options.baseUrl,
-        agent: def.agent ?? d.agent,
-        headers: options.headers,
-        mlflowRunId: runId,
-      });
-      result = await runEval(def, { id, driver, strict: options.strict });
+      def = await loadEval(d.file);
     } catch (err) {
-      result = {
+      emit({ type: "start", id, index, total });
+      const result: EvalResult = {
         id,
         assertions: [],
         passed: false,
         error: err instanceof Error ? err.message : String(err),
       };
+      results.push(result);
+      emit({ type: "result", result, index, total });
+      continue;
     }
 
-    results.push(result);
-    emit({ type: "result", result, index, total });
+    // Resolve dataset rows up front; a dataset eval with no rows still runs
+    // once with an empty row so a misconfiguration surfaces as a result.
+    let rows: Array<DatasetRow | undefined> = [undefined];
+    let datasetError: string | undefined;
+    if (def.dataset) {
+      if (!options.workspaceClient || !options.warehouseId) {
+        datasetError =
+          "dataset eval requires a workspace client and warehouse (pass --warehouse)";
+      } else {
+        try {
+          rows = await readEvalDataset(options.workspaceClient, {
+            table: def.dataset.table,
+            warehouseId: options.warehouseId,
+            limit: def.dataset.limit,
+          });
+          if (rows.length === 0) rows = [undefined];
+        } catch (err) {
+          datasetError = err instanceof Error ? err.message : String(err);
+        }
+      }
+    }
+
+    for (let r = 0; r < rows.length; r++) {
+      const rowId =
+        def.dataset && rows.length > 1
+          ? `${id} [row ${r + 1}/${rows.length}]`
+          : id;
+      emit({ type: "start", id: rowId, index, total });
+
+      let result: EvalResult;
+      if (datasetError) {
+        result = {
+          id: rowId,
+          assertions: [],
+          passed: false,
+          error: datasetError,
+        };
+      } else {
+        // A fresh driver per row: each row is an independent conversation, so
+        // its thread must not carry over the previous row's history. (Multiple
+        // `t.send`s within one row still share the thread — that's the driver's
+        // per-instance behavior.)
+        const driver = createHttpDriver({
+          baseUrl: options.baseUrl,
+          agent: def.agent ?? d.agent,
+          headers: options.headers,
+          mlflowRunId: runId,
+        });
+        result = await runEval(def, {
+          id: rowId,
+          driver,
+          strict: options.strict,
+          row: rows[r],
+        });
+      }
+
+      results.push(result);
+      emit({ type: "result", result, index, total });
+    }
   }
 
   if (mlflowClient && runId) {