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

apps/dev-playground/server/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

@@ -1,4 +1,7 @@
-import { createWorkspaceClient } from "../../workspace-client";
+import {
+  createWorkspaceClient,
+  type WorkspaceClient,
+} from "../../workspace-client";
 
 /** Resolved Databricks host + bearer token for the eval runner's REST calls. */
 export interface DatabricksAuth {
@@ -40,9 +43,8 @@ async function resolveViaSdk(
   options: ResolveDatabricksAuthOptions,
 ): Promise<DatabricksAuth | undefined> {
   try {
-    const client = createWorkspaceClient(
-      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
@@ -68,3 +70,29 @@ export async function resolveDatabricksAuth(
   }
   return resolveViaSdk(options);
 }
+
+/**
+ * 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 createWorkspaceClient({
+        host: options.host,
+        token: options.token,
+        authType: "pat",
+      });
+    }
+    return createWorkspaceClient(
+      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 { SQLWarehouseConnector } from "../connectors";
+import type { WorkspaceClient } from "../workspace-client";
+
+/**
+ * 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" && options.limit > 0
+      ? ` 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

@@ -126,6 +126,9 @@ export function createHttpDriver(options: HttpDriverOptions): EvalDriver {
   let threadId: string | undefined;
 
   return {
+    reset(): void {
+      threadId = undefined;
+    },
     async send(message: string): Promise<DriveResult> {
       // Bounds connect + the entire read below. Passed to both the fetch and
       // the SSE reader: on expiry the reader is cancelled and the turn fails.

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 } 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",