diff --git a/build.gradle b/build.gradle
index 1f4cbbb0b..dc1c3241d 100644
--- a/build.gradle
+++ b/build.gradle
@@ -153,19 +153,23 @@ allprojects {
   }
 
   afterEvaluate {
-    for (def task in it.tasks) {
-      if (task != rootProject.tasks.CopyGitHooksTask) {
+    it.tasks
+      .findAll { task -> task != rootProject.tasks.CopyGitHooksTask }
+      .each { task ->
         task.dependsOn rootProject.tasks.CopyGitHooksTask
       }
-    }
   }
 }
 
-// Local Git Hooks cannot be shared, as .git directory is gitignore'd.
+// Resolve the repository hooks directory through Git so linked worktrees and standard checkouts use the same task.
+def gitHooksDirectory = providers.exec {
+  commandLine 'git', 'rev-parse', '--git-path', 'hooks'
+}.standardOutput.asText.map { it.trim() }
+
 tasks.register('CopyGitHooksTask', Copy) {
-  println 'Make the git hook available in .git/hooks directory.'
+  println 'Make the Git hooks available in the repository hooks directory.'
   from file('scripts/git-hooks')
-  into file('.git/hooks/')
+  into gitHooksDirectory
 }
 
 // =============================================================================
diff --git a/integrations/spark/delta-harness/build.gradle b/integrations/spark/delta-harness/build.gradle
new file mode 100644
index 000000000..48c18e12e
--- /dev/null
+++ b/integrations/spark/delta-harness/build.gradle
@@ -0,0 +1,155 @@
+plugins {
+  id 'openhouse.java-minimal-conventions'
+  id 'openhouse.maven-publish'
+  id 'scala'
+}
+
+// The delta-harness behavioral matrix is published as a portable Scala library. This repository runs
+// it against the embedded catalog, and the LinkedIn acceptance tests supply a remote environment.
+//
+// Only the portable scenario/framework sources are published. Env.scala boots the embedded OpenHouse
+// server and pulls in its test fixtures, and LocalRunner.scala carries the harness.Main launch class
+// that drives it, so both are excluded from the published library and compiled in the local source
+// set. Runner.scala remains portable so its retry contract is exercised by the module tests.
+
+ext {
+  icebergVersion = rootProject.ext.iceberg_1_5_version
+  sparkVersion = '3.5.2'
+  scalaLibVersion = '2.12.18'
+}
+
+// The embedded boot wiring and the local run loop. They are excluded from the portable library and
+// are the whole of the local source set.
+def embeddedOnlySources = ['harness/openhouse/Env.scala', 'harness/openhouse/LocalRunner.scala']
+
+sourceSets {
+  main {
+    scala {
+      srcDirs = ['src/main/scala']
+      exclude embeddedOnlySources
+    }
+  }
+  local {
+    scala {
+      srcDirs = ['src/main/scala']
+      include embeddedOnlySources
+    }
+    compileClasspath += sourceSets.main.output
+    runtimeClasspath += sourceSets.main.output
+  }
+}
+
+dependencies {
+  implementation "org.scala-lang:scala-library:${scalaLibVersion}"
+
+  // The consumer provides the Spark, Iceberg, and OpenHouse runtime. The library jar carries only the
+  // harness classes.
+  compileOnly("org.apache.spark:spark-sql_2.12:${sparkVersion}") {
+    exclude group: 'io.netty'
+  }
+  compileOnly("com.linkedin.iceberg:iceberg-spark-runtime-3.5_2.12:${icebergVersion}") {
+    exclude group: 'io.netty'
+  }
+  // Provides com.linkedin.openhouse.javaclient.* (WebClientResponseWithMessageException, etc.).
+  compileOnly(project(path: ':integrations:spark:spark-3.5:openhouse-spark-3.5-runtime_2.12', configuration: 'shadow'))
+
+  // The catalog tests read Catalog contributions and cases without starting Spark. Their JVM still
+  // loads the harness signatures, so the compile-only harness dependencies must be present on the test classpath.
+  testImplementation("org.apache.spark:spark-sql_2.12:${sparkVersion}") {
+    exclude group: 'io.netty'
+  }
+  testImplementation("com.linkedin.iceberg:iceberg-spark-runtime-3.5_2.12:${icebergVersion}") {
+    exclude group: 'io.netty'
+  }
+  testImplementation(
+    project(path: ':integrations:spark:spark-3.5:openhouse-spark-3.5-runtime_2.12',
+            configuration: 'shadow'))
+
+  localImplementation sourceSets.main.output
+  localImplementation "org.scala-lang:scala-library:${scalaLibVersion}"
+  localImplementation("org.apache.spark:spark-sql_2.12:${sparkVersion}") {
+    exclude group: 'io.netty'
+  }
+  localImplementation(
+    project(path: ':integrations:spark:spark-3.5:openhouse-spark-3.5-runtime_2.12',
+            configuration: 'shadow')) {
+    exclude group: 'org.apache.commons', module: 'commons-lang3'
+  }
+  localImplementation project(':tables-test-fixtures:tables-test-fixtures-iceberg-1.5_2.12')
+}
+
+configurations.localRuntimeClasspath {
+  exclude group: 'com.linkedin.iceberg', module: 'iceberg-core'
+  exclude group: 'com.linkedin.iceberg', module: 'iceberg-api'
+  exclude group: 'com.linkedin.iceberg', module: 'iceberg-common'
+  exclude group: 'com.linkedin.iceberg', module: 'iceberg-data'
+  resolutionStrategy.force(
+    'com.fasterxml.jackson.core:jackson-annotations:2.15.2',
+    'com.fasterxml.jackson.core:jackson-core:2.15.2',
+    'com.fasterxml.jackson.core:jackson-databind:2.15.2',
+    'com.fasterxml.jackson.dataformat:jackson-dataformat-xml:2.15.2',
+    'com.fasterxml.jackson.dataformat:jackson-dataformat-yaml:2.15.2',
+    'com.fasterxml.jackson.datatype:jackson-datatype-jdk8:2.15.2',
+    'com.fasterxml.jackson.datatype:jackson-datatype-jsr310:2.15.2',
+    'com.fasterxml.jackson.jaxrs:jackson-jaxrs-base:2.15.2',
+    'com.fasterxml.jackson.jaxrs:jackson-jaxrs-json-provider:2.15.2',
+    'com.fasterxml.jackson.module:jackson-module-jaxb-annotations:2.15.2',
+    'com.fasterxml.jackson.module:jackson-module-parameter-names:2.15.2',
+    'com.fasterxml.jackson.module:jackson-module-scala_2.12:2.15.2')
+}
+
+def configureEmbeddedHarnessRun = { JavaExec task ->
+  task.dependsOn localClasses
+  task.classpath = sourceSets.local.runtimeClasspath
+  task.mainClass = 'harness.Main'
+  if (JavaVersion.current() >= JavaVersion.VERSION_1_9) {
+    task.jvmArgs(
+      '--add-opens=java.base/java.nio=ALL-UNNAMED',
+      '--add-exports=java.base/sun.nio.ch=ALL-UNNAMED',
+      '--add-opens=java.base/sun.util.calendar=ALL-UNNAMED',
+      '--add-exports=java.base/sun.util.calendar=ALL-UNNAMED')
+  }
+}
+
+tasks.register('runOpenHouse', JavaExec) {
+  group = 'verification'
+  description = 'Runs the delta harness against an embedded OpenHouse catalog.'
+  configureEmbeddedHarnessRun(delegate)
+}
+
+def verifyOpenHouseFoundation = tasks.register('verifyOpenHouseFoundation', JavaExec) {
+  group = 'verification'
+  description = 'Runs the foundation scenarios against an embedded OpenHouse catalog.'
+  configureEmbeddedHarnessRun(delegate)
+  args '--catalog=foundation'
+}
+
+def verifyPortableJar = tasks.register('verifyPortableJar') {
+  group = 'verification'
+  description = 'Checks that the published harness jar contains only portable entry points.'
+  dependsOn tasks.named('jar')
+  doLast {
+    def jarFile = tasks.named('jar').get().archiveFile.get().asFile
+    def entries = new java.util.zip.ZipFile(jarFile).withCloseable { archive ->
+      archive.entries().collect { entry -> entry.name }.toSet()
+    }
+    ['harness/Plan.class', 'harness/Runner.class'].each { requiredEntry ->
+      if (!entries.contains(requiredEntry)) {
+        throw new GradleException("Portable harness jar is missing ${requiredEntry}")
+      }
+    }
+    ['harness/Main.class', 'harness/Main$.class', 'harness/OpenHouseEnv.class', 'harness/OpenHouseEnv$.class']
+      .each { embeddedEntry ->
+        if (entries.contains(embeddedEntry)) {
+          throw new GradleException("Portable harness jar contains embedded-only ${embeddedEntry}")
+        }
+      }
+  }
+}
+
+tasks.named('check') {
+  dependsOn verifyOpenHouseFoundation
+  dependsOn verifyPortableJar
+}
+
+jar.enabled = true
diff --git a/integrations/spark/delta-harness/run-openhouse.sh b/integrations/spark/delta-harness/run-openhouse.sh
new file mode 100755
index 000000000..6cda85599
--- /dev/null
+++ b/integrations/spark/delta-harness/run-openhouse.sh
@@ -0,0 +1,28 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)"
+REPO_ROOT="$(cd "$SCRIPT_DIR/../../.." && pwd)"
+JAVA_HOME="${JAVA17_HOME:-${JAVA_HOME:-}}"
+if [[ -z "$JAVA_HOME" ]]; then
+  echo "Set JAVA17_HOME or JAVA_HOME to a JDK 17 installation." >&2
+  exit 2
+fi
+export JAVA_HOME
+
+JAVA_VERSION="$("$JAVA_HOME/bin/java" -version 2>&1 | sed -n '1s/.*version "\([0-9][0-9]*\).*/\1/p')"
+if [[ "$JAVA_VERSION" != "17" ]]; then
+  echo "The delta harness requires JDK 17; $JAVA_HOME reports Java $JAVA_VERSION." >&2
+  exit 2
+fi
+
+cd "$REPO_ROOT"
+if (( $# == 0 )); then
+  exec ./gradlew --no-daemon \
+    :integrations:spark:openhouse-spark-delta-harness_2.12:runOpenHouse
+fi
+
+printf -v FILTERS ' %q' "$@"
+exec ./gradlew --no-daemon \
+  :integrations:spark:openhouse-spark-delta-harness_2.12:runOpenHouse \
+  --args="${FILTERS# }"
diff --git a/integrations/spark/delta-harness/src/main/scala/harness/openhouse/Env.scala b/integrations/spark/delta-harness/src/main/scala/harness/openhouse/Env.scala
new file mode 100644
index 000000000..cfd10a2cb
--- /dev/null
+++ b/integrations/spark/delta-harness/src/main/scala/harness/openhouse/Env.scala
@@ -0,0 +1,72 @@
+package harness
+
+import org.apache.spark.sql.SparkSession
+import scala.io.Source
+
+/**
+ * Embedded environment wiring: boots the in-process OpenHouse server and hands back a SparkSession pointed at its
+ * catalog. This file is compiled into the `local` source set only, because it pulls in the server test fixtures that
+ * the published portable library leaves out.
+ */
+object OpenHouseEnv {
+  import com.linkedin.openhouse.tablestest.OpenHouseLocalServer
+
+  private def authToken(): String =
+    Option(getClass.getClassLoader.getResourceAsStream("dummy.token"))
+      .map(tokenStream => Source.fromInputStream(tokenStream, "UTF-8").mkString.trim)
+      .getOrElse("default-token")
+
+  private def wireCatalog(
+      builder: SparkSession.Builder,
+      name: String,
+      uri: String,
+      token: String): SparkSession.Builder =
+    builder
+      .config(s"spark.sql.catalog.$name", "org.apache.iceberg.spark.SparkCatalog")
+      .config(
+        s"spark.sql.catalog.$name.catalog-impl",
+        "com.linkedin.openhouse.spark.OpenHouseCatalog")
+      .config(s"spark.sql.catalog.$name.uri", uri)
+      .config(s"spark.sql.catalog.$name.cluster", "local-cluster")
+      .config(s"spark.sql.catalog.$name.auth-token", token)
+
+  def start(): (OpenHouseLocalServer, SparkSession) = {
+    // The embedded server uses Hibernate to create its H2 schema. Hibernate owns initialization for this process, so
+    // classpath SQL initialization stays disabled.
+    System.setProperty("spring.sql.init.mode", "never")
+    System.setProperty("spring.jpa.hibernate.ddl-auto", "create-drop")
+
+    val server = new OpenHouseLocalServer()
+    server.start()
+    try {
+      val uri = s"http://localhost:${server.getPort}"
+      val token = authToken()
+
+      val base = SparkSession.builder()
+        .appName("delta-harness-openhouse")
+        .master("local[2]")
+        .config("spark.sql.extensions",
+          "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions," +
+            "com.linkedin.openhouse.spark.extensions.OpenhouseSparkSessionExtensions")
+        .config("spark.hadoop.fs.defaultFS", "file:///")
+        .config("spark.sql.session.timeZone", "UTC")
+        .config("spark.sql.autoBroadcastJoinThreshold", "-1")
+        .config("spark.driver.bindAddress", "127.0.0.1")
+        .config("spark.ui.enabled", "false")
+
+      val wired =
+        Seq("openhouse", "default_iceberg")
+          .foldLeft(base)(wireCatalog(_, _, uri, token))
+      (server, wired.getOrCreate())
+    } catch {
+      case startupFailure: Throwable =>
+        try {
+          server.stop()
+        } catch {
+          case cleanupFailure: Throwable =>
+            startupFailure.addSuppressed(cleanupFailure)
+        }
+        throw startupFailure
+    }
+  }
+}
diff --git a/integrations/spark/delta-harness/src/main/scala/harness/openhouse/Framework.scala b/integrations/spark/delta-harness/src/main/scala/harness/openhouse/Framework.scala
new file mode 100644
index 000000000..9d9476bbd
--- /dev/null
+++ b/integrations/spark/delta-harness/src/main/scala/harness/openhouse/Framework.scala
@@ -0,0 +1,418 @@
+package harness
+
+import org.apache.spark.sql.{Row, SparkSession}
+import java.math.{BigDecimal => JavaBigDecimal}
+import java.net.{ConnectException, SocketException, SocketTimeoutException}
+import java.sql.{Date, Timestamp}
+import java.time.LocalDate
+import java.time.LocalDateTime
+import java.time.format.DateTimeFormatter
+import java.util.UUID
+import java.util.concurrent.atomic.AtomicInteger
+import scala.annotation.tailrec
+import scala.reflect.{ClassTag, classTag}
+import scala.util.control.NonFatal
+
+// The harness defines typed, reusable table preparations and self-contained TestCase bodies. Each case gets a fresh table,
+// executes its preparation, runs its action and assertions, and drops the table during teardown.
+
+/**
+ * One catalog case: the ID that names it, the body that runs it, and its two skip policies.
+ * `knownBugReason` marks a case the catalog under test is known to fail; `embeddedSkipReason` marks a case the
+ * embedded local catalog cannot run at all. A runner reports either policy as a skip.
+ */
+final case class TestCase(
+  id:                 String,
+  run:                Ctx => Unit,
+  knownBugReason:     Option[String] = None,
+  embeddedSkipReason: Option[String] = None
+) {
+  /** The skip reason a known bug produces, phrased so a run log explains the skip. */
+  def bugReason: Option[String] = knownBugReason.map(reason => s"bug: $reason")
+}
+
+final case class Ctx(spark: SparkSession, namespace: String)
+
+sealed trait Outcome { def label: String }
+object Outcome {
+  case object Passed extends Outcome { val label = "PASS" }
+  final case class Failed(cause: Throwable) extends Outcome {
+    val label = "FAIL"
+    def reason: String = {
+      val rootCause = Exceptions.root(cause)
+      val rootReason =
+        s"${rootCause.getClass.getSimpleName}: ${Option(rootCause.getMessage).getOrElse(rootCause.toString)}"
+      if (rootCause eq cause) {
+        rootReason
+      } else {
+        s"${cause.getClass.getSimpleName}: ${Option(cause.getMessage).getOrElse(cause.toString)}; caused by $rootReason"
+      }
+    }
+  }
+  final case class Skipped(reason: String) extends Outcome { val label = "SKIP" }
+}
+
+object Exceptions {
+  def causeChain(throwable: Throwable): List[Throwable] = {
+    @tailrec
+    def collect(
+      current: Option[Throwable],
+      seen: Set[Throwable],
+      collected: List[Throwable]
+    ): List[Throwable] =
+      current match {
+        case Some(cause) if !seen.contains(cause) =>
+          collect(Option(cause.getCause), seen + cause, cause :: collected)
+        case _ =>
+          collected.reverse
+      }
+
+    collect(Some(throwable), Set.empty, Nil)
+  }
+
+  def root(throwable: Throwable): Throwable = causeChain(throwable).last
+
+  /**
+   * Retries errors positively identified as transient. Other failures remain terminal so data, permission, and
+   * assertion failures surface on their first attempt.
+   */
+  def isTransientConnectionFailure(throwable: Throwable): Boolean = causeChain(throwable).exists {
+    case _: SocketTimeoutException => true
+    case _: ConnectException       => true
+    case socketFailure: SocketException =>
+      Option(socketFailure.getMessage).exists(_.toLowerCase.contains("reset"))
+    case _ => false
+  }
+}
+
+// Tests assert with plain `assert`; a failed assertion throws AssertionError, which is NonFatal and so is caught at the
+// Runner edge and reported as a (terminal) failure.
+object Check {
+  /** Requires `operation` to throw `E` and returns the exception for message assertions. */
+  def intercept[E <: Throwable: ClassTag](operation: => Unit): E = {
+    val expected = classTag[E].runtimeClass
+    val caught: Option[Throwable] =
+      try {
+        operation
+        None
+      } catch {
+        case NonFatal(throwable) => Some(throwable)
+      }
+    caught match {
+      case Some(throwable) if expected.isInstance(throwable) =>
+        throwable.asInstanceOf[E]
+      case Some(throwable) =>
+        throw new AssertionError(
+          s"expected ${expected.getName} but got ${throwable.getClass.getName}: " +
+            throwable.getMessage,
+          throwable)
+      case None =>
+        throw new AssertionError(
+          s"expected ${expected.getName} to be thrown, but nothing was")
+    }
+  }
+}
+
+// `Column[T]` carries the Scala type the column reads back as, so typed row access (`row.get(CoreTable.long0): Long`)
+// is compiler-checked. `literalAt(rowIndex)` is a pure function of the row index, so generated data is reproducible.
+// Value generation lives on the column, which keeps RowGenerator a plain iteration with no knowledge of types.
+final case class Column[T](columnName: String, sqlType: String, literalAt: Int => String)
+
+sealed trait Schema {
+  def tableColumns: Seq[Column[_]]
+  def columnNames: Seq[String] = tableColumns.map(_.columnName)
+}
+
+/** Typed row access, keyed by the column's name: `row.get(CoreTable.long0)` returns a `Long`. */
+object Rows {
+  implicit class TypedRow(val row: Row) extends AnyVal {
+    def get[T](column: Column[T]): T = row.getAs[T](column.columnName)
+  }
+}
+
+// A representative core table with one column per common data type and a string-encoded date. Tests reference columns
+// through these handles, so a column rename propagates to every caller.
+object CoreTable extends Schema {
+  val long0:    Column[Long]    = Column("foo_col_long",    "bigint",  rowIndex => rowIndex.toString)
+  val int0:     Column[Int]     = Column("foo_col_int",     "int",     rowIndex => rowIndex.toString)
+  val string0:  Column[String]  = Column("foo_col_string",  "string",  rowIndex => s"'row-$rowIndex'")
+  val double0:  Column[Double]  = Column("foo_col_double",  "double",  rowIndex => s"$rowIndex.5")
+  val boolean0: Column[Boolean] =
+    Column("foo_col_boolean", "boolean", rowIndex => if (rowIndex % 2 == 0) "true" else "false")
+  val date0: Column[String] =
+    Column("foo_col_date", "string", rowIndex => s"'${CoreTable.dateLiteral(rowIndex)}'")
+  def tableColumns: Seq[Column[_]] = Seq(long0, int0, string0, double0, boolean0, date0)
+
+  private val DateFormat = DateTimeFormatter.ofPattern("yyyy-MM-dd-HH")
+  private val DateEpoch  = LocalDateTime.of(2024, 1, 1, 0, 0)
+
+  /** Deterministic YYYY-MM-DD-HH date value (one hour per row), formatted via java.time. */
+  def dateLiteral(rowIndex: Int): String =
+    DateEpoch.plusHours((rowIndex - 1).toLong).format(DateFormat)
+}
+
+// A schema containing complex/nested types: a struct, an array, a map, and a struct-in-struct. Struct/array read back
+// as Row/Seq; map as a Map. `id` is first so it is the ordering key.
+object NestedTable extends Schema {
+  val id: Column[Long] =
+    Column("id", "bigint", rowIndex => rowIndex.toString)
+  val s: Column[Row] =
+    Column(
+      "s",
+      "struct<x:int,y:string>",
+      rowIndex => s"named_struct('x', $rowIndex, 'y', 'row-$rowIndex')")
+  val arr: Column[Seq[Int]] =
+    Column("arr", "array<int>", rowIndex => s"array($rowIndex, ${rowIndex + 1})")
+  val m: Column[Map[String, Int]] =
+    Column("m", "map<string,int>", rowIndex => s"map('k', $rowIndex)")
+  val nested: Column[Row] =
+    Column(
+      "nested",
+      "struct<inner:struct<z:int>>",
+      rowIndex => s"named_struct('inner', named_struct('z', $rowIndex))")
+  def tableColumns: Seq[Column[_]] = Seq(id, s, arr, m, nested)
+
+  val columnDefinitions: String =
+    "id bigint, s struct<x:int,y:string>, arr array<int>, m map<string,int>, nested struct<inner:struct<z:int>>"
+}
+
+// A schema for the common scalar types, including literals for null, floating-point, boundary, unicode, and empty
+// string cases.
+object TypesTable extends Schema {
+  val id: Column[Long] =
+    Column("id", "bigint", rowIndex => rowIndex.toString)
+  val n: Column[Int] =
+    Column("n", "int", rowIndex => rowIndex.toString)
+  val x: Column[Double] =
+    Column("x", "double", rowIndex => s"$rowIndex.5")
+  val dec: Column[JavaBigDecimal] =
+    Column("dec", "decimal(10,2)", rowIndex => s"CAST($rowIndex.50 AS decimal(10,2))")
+  val str: Column[String] =
+    Column("str", "string", rowIndex => s"'row-$rowIndex'")
+  val bin: Column[Array[Byte]] =
+    Column("bin", "binary", rowIndex => s"CAST('bin-$rowIndex' AS binary)")
+  val dt: Column[Date] =
+    Column(
+      "dt",
+      "date",
+      rowIndex => s"DATE '${DateEpoch.plusDays((rowIndex - 1).toLong)}'")
+  val ts: Column[Timestamp] =
+    Column(
+      "ts",
+      "timestamp",
+      rowIndex =>
+        s"TIMESTAMP '${TimestampEpoch.plusHours((rowIndex - 1).toLong).format(TimestampFormat)}'")
+  val tsntz: Column[LocalDateTime] =
+    Column(
+      "tsntz",
+      "timestamp_ntz",
+      rowIndex =>
+        s"TIMESTAMP_NTZ '${TimestampEpoch.plusHours((rowIndex - 1).toLong).format(TimestampFormat)}'")
+  def tableColumns: Seq[Column[_]] = Seq(id, n, x, dec, str, bin, dt, ts, tsntz)
+
+  val columnDefinitions: String =
+    "id bigint, n int, x double, dec decimal(10,2), str string, bin binary, dt date, ts timestamp, " +
+      "tsntz timestamp_ntz"
+
+  private val DateEpoch = LocalDate.of(2024, 1, 1)
+  private val TimestampEpoch = LocalDateTime.of(2024, 1, 1, 0, 0)
+  private val TimestampFormat = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")
+}
+
+object RowGenerator {
+  /** VALUES clause for `numberOfRows` deterministic rows, one literal per column. */
+  def valuesClause(schema: Schema, numberOfRows: Int): String =
+    (1 to numberOfRows).map { rowIndex =>
+      schema.tableColumns.map(column => column.literalAt(rowIndex)).mkString("(", ", ", ")")
+    }.mkString("VALUES ", ", ", "")
+}
+
+/**
+ * What a step's validation thunk sees: the live table, its rows before and after the step, and the table's snapshot
+ * (commit) count before and after, so a test can assert the delta in both data and commits (e.g. "a no-match UPDATE
+ * still commits exactly one snapshot").
+ */
+final case class StepView[S <: Schema](
+  spark:           SparkSession,
+  table:           String,
+  schema:          S,
+  before:          Seq[Row],
+  after:           Seq[Row],
+  snapshotsBefore: Long,
+  snapshotsAfter:  Long
+)
+
+final case class TableState(rows: Seq[Row], snapshotCount: Long)
+
+/** A fresh table after its reusable preparation has completed. */
+final case class PreparedTable[S <: Schema](
+  spark:                 SparkSession,
+  name:                  String,
+  schema:                S,
+  preparedRows:          Seq[Row],
+  preparedSnapshotCount: Long
+) {
+  def rows: Seq[Row] = PreparedTable.currentRows(spark, name, schema)
+  def snapshotCount: Long = PreparedTable.snapshotCount(spark, name)
+  def state: TableState = TableState(rows, snapshotCount)
+}
+
+object PreparedTable {
+  private[harness] def currentRows[S <: Schema](spark: SparkSession, table: String, schema: S): Seq[Row] = {
+    val columns = schema.columnNames.mkString(", ")
+    spark.sql(s"SELECT $columns FROM $table ORDER BY ${schema.columnNames.head}").collect().toSeq
+  }
+
+  private[harness] def snapshotCount(spark: SparkSession, table: String): Long =
+    spark.sql(s"SELECT count(*) FROM $table.snapshots").collect()(0).getLong(0)
+}
+
+/** One preparation step and its validation. */
+final case class Step[S <: Schema](
+  label:    String,
+  execute:  (SparkSession, String, S) => Unit,
+  validate: StepView[S] => Unit
+)
+
+/** An immutable, typed sequence of table-preparation steps. */
+final class TableTest[S <: Schema] private (val schema: S, val steps: Vector[Step[S]]) {
+  private def add(step: Step[S]): TableTest[S] = new TableTest(schema, steps :+ step)
+
+  // The default validator asserts the seed actually appended `numberOfRows` rows. This defends the case assertions
+  // from a vacuous pass on an empty or short baseline.
+  def insert(numberOfRows: Int)(
+      validate: StepView[S] => Unit = view => assert(
+        view.after.size == view.before.size + numberOfRows,
+        s"seed insert($numberOfRows) expected ${view.before.size + numberOfRows} rows, got ${view.after.size}")
+  ): TableTest[S] =
+    add(Step(s"insert($numberOfRows)", (spark, table, schema) =>
+      spark.sql(s"INSERT INTO $table ${RowGenerator.valuesClause(schema, numberOfRows)}"), validate))
+
+  /** Run an arbitrary preparation step, then validate its result. */
+  def step(label: String)(mutate: (SparkSession, String) => Unit)
+          (validate: StepView[S] => Unit = _ => ()): TableTest[S] =
+    add(Step(label, (spark, table, _) => mutate(spark, table), validate))
+
+  /** Run one preparation SQL statement, then validate its result. */
+  def sql(label: String)(statement: String => String)
+         (validate: StepView[S] => Unit = _ => ()): TableTest[S] =
+    step(label)((spark, table) => spark.sql(statement(table)))(validate)
+
+  /**
+   * Execute these steps as a reusable preparation, then hand the prepared table to one self-contained test body. The
+   * fresh-table lifecycle covers both the preparation and the test body.
+   */
+  def prepare(ctx: Ctx)(use: PreparedTable[S] => Unit): Unit =
+    withTable(ctx) { (table, markTableCreated) =>
+    val (preparedRows, preparedSnapshotCount) =
+      steps.zipWithIndex.foldLeft((Seq.empty[Row], 0L)) {
+        case ((beforeRows, beforeSnapshots), (step, stepIndex)) =>
+        step.execute(ctx.spark, table, schema)
+        if (stepIndex == 0) {
+          markTableCreated()
+        }
+        val afterRows = PreparedTable.currentRows(ctx.spark, table, schema)
+        val afterSnapshots = PreparedTable.snapshotCount(ctx.spark, table)
+        step.validate(
+          StepView(
+            ctx.spark,
+            table,
+            schema,
+            beforeRows,
+            afterRows,
+            beforeSnapshots,
+            afterSnapshots))
+        (afterRows, afterSnapshots)
+      }
+    use(PreparedTable(ctx.spark, table, schema, preparedRows, preparedSnapshotCount))
+  }
+
+  // Gives the preparation a unique table name and drops that table after the test. Cleanup starts only after the first
+  // preparation step creates the table, so a name conflict preserves the pre-existing table. A test failure stays
+  // primary, and a cleanup failure is attached to it as a suppressed exception.
+  private def withTable(ctx: Ctx)(use: (String, () => Unit) => Unit): Unit = {
+    val table = TableTest.nextQualifiedTableName(ctx.namespace)
+    OwnedTableLifecycle.withOwnership(
+      ctx.spark.sql(s"DROP TABLE IF EXISTS $table"))(
+      markTableCreated => use(table, markTableCreated))
+  }
+
+}
+
+private[harness] object OwnedTableLifecycle {
+  /**
+   * Runs `use`, then runs `cleanUp` on every outcome. A failure from `use` is the failure the caller sees, with a
+   * cleanup failure attached to it as a suppressed exception. When `use` returns normally a cleanup failure is the
+   * failure the caller sees, so cleanup that silently fails cannot pass for a clean run.
+   */
+  def withCleanup(cleanUp: => Unit)(use: => Unit): Unit = {
+    var primaryFailure: Option[Throwable] = None
+    try {
+      use
+    } catch {
+      case failure: Throwable =>
+        primaryFailure = Some(failure)
+        throw failure
+    } finally {
+      try {
+        cleanUp
+      } catch {
+        case cleanupFailure: Throwable =>
+          primaryFailure match {
+            case Some(failure) => failure.addSuppressed(cleanupFailure)
+            case None          => throw cleanupFailure
+          }
+      }
+    }
+  }
+
+  /**
+   * Runs `use` with a mark it calls once the table exists. `dropOwnedTable` runs only when that mark was set, so a
+   * create that fails leaves whatever already answered to the name untouched.
+   */
+  def withOwnership(dropOwnedTable: => Unit)(use: (() => Unit) => Unit): Unit = {
+    var tableCreated = false
+    withCleanup(if (tableCreated) dropOwnedTable)(use(() => tableCreated = true))
+  }
+}
+
+object TableTest {
+  private val counter = new AtomicInteger(0)
+
+  def apply[S <: Schema](schema: S): TableTest[S] = new TableTest(schema, Vector.empty)
+  def seedCounter(value: Int): Unit = counter.set(value)
+
+  private[harness] def nextQualifiedTableName(namespace: String): String =
+    s"$namespace.t_${UUID.randomUUID().toString.replace("-", "")}_${counter.incrementAndGet()}"
+}
+
+/** An immutable recipe that prepares one fresh table for each test case. */
+final case class TablePreparation[S <: Schema](
+  label: String,
+  preparation: TableTest[S],
+  casePrefix: String = "",
+  afterTest: PreparedTable[S] => Unit = (_: PreparedTable[S]) => ()
+) {
+  /**
+   * Build the case that runs `body` against one freshly prepared table. The case ID combines the preparation's prefix
+   * and label with `caseName`, so one test body yields a separate case on every preparation it runs on.
+   */
+  def test(caseName: String)(body: PreparedTable[S] => Unit): TestCase =
+    TestCase(
+      s"$casePrefix$caseName @ $label",
+      context => preparation.prepare(context) { table =>
+        OwnedTableLifecycle.withCleanup(afterTest(table))(body(table))
+      })
+}
+
+final case class DmlTestCase[S <: Schema](
+  id: String,
+  run: PreparedTable[S] => Unit,
+  knownBugReason: Option[String] = None
+) {
+  /** Build the case that runs this operation against a table `preparation` produces. */
+  def runOn(preparation: TablePreparation[S]): TestCase =
+    preparation
+      .test(id)(run)
+      .copy(knownBugReason = knownBugReason)
+}
diff --git a/integrations/spark/delta-harness/src/main/scala/harness/openhouse/LocalRunner.scala b/integrations/spark/delta-harness/src/main/scala/harness/openhouse/LocalRunner.scala
new file mode 100644
index 000000000..89e0d88fe
--- /dev/null
+++ b/integrations/spark/delta-harness/src/main/scala/harness/openhouse/LocalRunner.scala
@@ -0,0 +1,127 @@
+package harness
+
+import java.util.concurrent.{Callable, Executors, TimeUnit}
+
+/**
+ * The `harness.Main` launch class used by the embedded OpenHouse Gradle tasks. This file is compiled into the `local`
+ * source set only, so the published portable library carries the catalog and retry policy without a launch loop.
+ */
+object Main {
+  private val FoundationCatalogArgument = "--catalog=foundation"
+
+  def main(args: Array[String]): Unit = {
+    val (server, spark) = OpenHouseEnv.start()
+    var runFailure: Option[Throwable] = None
+    try {
+      spark.sparkContext.setLogLevel("ERROR")
+      val ctx = Ctx(spark, "openhouse.dbMatrix")
+
+      val (catalogArguments, filters) = args.toList.partition(_.startsWith("--catalog="))
+      val selectedCatalog = catalogArguments match {
+        case Nil =>
+          Catalog.cases
+        case List(FoundationCatalogArgument) =>
+          Catalog.foundationContributions.flatMap { case (_, contribution) => contribution }
+        case unsupported =>
+          throw new HarnessConfigurationException(
+            s"supported catalog selection: $FoundationCatalogArgument; received ${unsupported.mkString(", ")}")
+      }
+      val cases = selectedCatalog.filter(testCase =>
+        filters.forall(testCase.id.contains))
+
+      val header =
+        if (catalogArguments.nonEmpty && filters.isEmpty) {
+          "foundation catalog"
+        } else if (filters.isEmpty) {
+          "full catalog"
+        } else {
+          s"filter ${filters.mkString(", ")} -> ${cases.size} cases"
+        }
+      println(s"\n=== delta-harness :: scenario cases @ OpenHouse catalog ($header) ===\n")
+
+      // Each case owns a fresh table. Worker tasks use separate Spark sessions over the shared Spark context, and
+      // results are printed in catalog order.
+      val parallelism =
+        RunnerConfiguration.parallelism(sys.env, Runtime.getRuntime.availableProcessors())
+      println(s"parallelism: $parallelism worker sessions\n")
+
+      def runOne(testCase: TestCase): (TestCase, (Outcome, Int)) =
+        testCase.embeddedSkipReason
+          .map(reason => s"embedded limitation: $reason")
+          .orElse(testCase.bugReason) match {
+          case Some(reason) =>
+            (testCase, (Outcome.Skipped(reason): Outcome, 0))
+          case None =>
+            (testCase, Runner.execute(testCase, ctx))
+        }
+
+      val results =
+        if (parallelism <= 1) {
+          cases.map(runOne)
+        } else {
+          val pool = Executors.newFixedThreadPool(parallelism)
+          try {
+            val futures = cases.map(testCase =>
+              pool.submit(
+                new Callable[(TestCase, (Outcome, Int))] {
+                  def call(): (TestCase, (Outcome, Int)) = runOne(testCase)
+                }))
+            futures.map(_.get(60, TimeUnit.MINUTES))
+          } finally {
+            pool.shutdownNow()
+          }
+        }
+
+      results.foreach { case (testCase, (outcome, attempts)) =>
+        val note = outcome match {
+          case failure: Outcome.Failed =>
+            s"  (${failure.reason})"
+          case Outcome.Skipped(reason) =>
+            s"  ($reason)"
+          case Outcome.Passed =>
+            ""
+        }
+        println(f"${outcome.label}%-4s ${testCase.id}%-52s try=$attempts$note")
+      }
+
+      val failed =
+        results.count { case (_, (outcome, _)) => outcome.isInstanceOf[Outcome.Failed] }
+      val skipped =
+        results.count { case (_, (outcome, _)) => outcome.isInstanceOf[Outcome.Skipped] }
+      val passed = results.size - failed - skipped
+      println(f"\n$passed passed, $skipped skipped, $failed failed  (${results.size} cases)")
+
+      if (failed > 0 || passed == 0) {
+        throw new AssertionError(
+          s"delta harness finished with $passed passed, $skipped skipped, and $failed failed cases")
+      }
+    } catch {
+      case failure: Throwable =>
+        runFailure = Some(failure)
+        throw failure
+    } finally {
+      val cleanupFailures =
+        List[() => Unit](
+          () => spark.stop(),
+          () => server.stop())
+          .flatMap { cleanup =>
+            try {
+              cleanup()
+              None
+            } catch {
+              case failure: Throwable => Some(failure)
+            }
+          }
+
+      runFailure match {
+        case Some(failure) =>
+          cleanupFailures.foreach(failure.addSuppressed)
+        case None =>
+          cleanupFailures.headOption.foreach { failure =>
+            cleanupFailures.drop(1).foreach(failure.addSuppressed)
+            throw failure
+          }
+      }
+    }
+  }
+}
diff --git a/integrations/spark/delta-harness/src/main/scala/harness/openhouse/Runner.scala b/integrations/spark/delta-harness/src/main/scala/harness/openhouse/Runner.scala
new file mode 100644
index 000000000..1a84f58ee
--- /dev/null
+++ b/integrations/spark/delta-harness/src/main/scala/harness/openhouse/Runner.scala
@@ -0,0 +1,55 @@
+package harness
+
+import scala.annotation.tailrec
+import scala.util.Try
+import scala.util.control.NonFatal
+
+final class HarnessConfigurationException(message: String) extends Exception(message)
+
+private[harness] object RunnerConfiguration {
+  def parallelism(environment: Map[String, String], availableProcessors: Int): Int =
+    environment.get("HARNESS_PARALLELISM") match {
+      case None => math.max(1, availableProcessors)
+      case Some(value) =>
+        Try(value.toInt).toOption.filter(_ > 0).getOrElse(
+          throw new HarnessConfigurationException(
+            s"HARNESS_PARALLELISM must be a positive integer; received '$value'"))
+    }
+}
+
+/** Executes one case, retrying only transient session creation failures before the case starts. */
+object Runner {
+  val MaxAttempts = 3
+
+  /** Runs a case. Once the case body starts, every failure is terminal because observable state may have changed. */
+  def execute(testCase: TestCase, context: Ctx): (Outcome, Int) = {
+    @tailrec def attempt(attemptIndex: Int): (Outcome, Int) = {
+      val attemptContext =
+        try {
+          Right(context.copy(spark = context.spark.newSession()))
+        } catch {
+          case NonFatal(throwable) => Left(Outcome.Failed(throwable))
+        }
+
+      attemptContext match {
+        case Left(failure)
+            if Exceptions.isTransientConnectionFailure(failure.cause) &&
+              attemptIndex + 1 < MaxAttempts =>
+          attempt(attemptIndex + 1)
+        case Left(failure) =>
+          (failure, attemptIndex + 1)
+        case Right(caseContext) =>
+          val outcome =
+            try {
+              testCase.run(caseContext)
+              Outcome.Passed
+            } catch {
+              case NonFatal(throwable) => Outcome.Failed(throwable)
+            }
+          (outcome, attemptIndex + 1)
+      }
+    }
+
+    attempt(0)
+  }
+}
diff --git a/integrations/spark/delta-harness/src/main/scala/harness/openhouse/scenarios/Catalog.scala b/integrations/spark/delta-harness/src/main/scala/harness/openhouse/scenarios/Catalog.scala
new file mode 100644
index 000000000..fa2becd09
--- /dev/null
+++ b/integrations/spark/delta-harness/src/main/scala/harness/openhouse/scenarios/Catalog.scala
@@ -0,0 +1,59 @@
+package harness
+
+/**
+ * Mixes every registered scenario with one shared fixture instance. A scenario becomes runnable when this object
+ * mixes in its trait and `Catalog` registers its case list.
+ */
+object Scenarios
+    extends ScenarioCoreDml
+    with ScenarioDataType
+    with ScenarioDmlRejection
+
+/**
+ * The ordered case catalog. Each named contribution owns its scenario body, preparation, assertions, and case IDs.
+ * Extensions add one scenario mixin to `Scenarios` and one named case list to `extensionContributions`.
+ */
+object Catalog {
+
+  /** The three scenario contributions that exercise the framework's core composition paths. */
+  def foundationContributions: List[(String, List[TestCase])] =
+    List(
+      "dataTypeCases"     -> Scenarios.dataTypeCases,
+      "dmlCoreCases"      -> Scenarios.dmlCoreCases,
+      "dmlRejectionCases" -> Scenarios.dmlRejectionCases)
+
+  /** Additional named scenario contributions supplied by a composed catalog. */
+  def extensionContributions: List[(String, List[TestCase])] =
+    List.empty
+
+  /** Every capability contribution, named once, in the order the catalog integrates them. */
+  def contributions: List[(String, List[TestCase])] =
+    (foundationContributions ++ extensionContributions).sortBy { case (name, _) => name }
+
+  /** The deterministic ordered case catalog. */
+  def cases: List[TestCase] = contributions.flatMap { case (_, contribution) => contribution }
+
+  /** The case IDs in catalog order. Reading them is a Spark-free catalog operation. */
+  def caseIds: List[String] = cases.map(_.id)
+
+}
+
+/** Published facade for adapters that construct cases or enumerate the catalog through `Plan`. */
+object Plan {
+
+  /** The catalog's case type. */
+  type Case = TestCase
+
+  /** The case constructor and extractor exposed through the facade. */
+  val Case: TestCase.type = TestCase
+
+  /** The deterministic ordered case catalog. */
+  def cases: List[TestCase] = Catalog.cases
+
+  /** The case IDs in catalog order. */
+  def caseIds: List[String] = Catalog.caseIds
+
+  /** The skip reason a known bug produces, phrased so a run log explains the skip. */
+  def bugReason(testCase: TestCase): Option[String] = testCase.bugReason
+
+}
diff --git a/integrations/spark/delta-harness/src/main/scala/harness/openhouse/scenarios/ScenarioCoreDml.scala b/integrations/spark/delta-harness/src/main/scala/harness/openhouse/scenarios/ScenarioCoreDml.scala
new file mode 100644
index 000000000..7a1e62808
--- /dev/null
+++ b/integrations/spark/delta-harness/src/main/scala/harness/openhouse/scenarios/ScenarioCoreDml.scala
@@ -0,0 +1,184 @@
+package harness
+
+import org.apache.spark.sql.Row
+
+/**
+ * The core behavior slice: one exact read, append, overwrite, delete, update, and merge contract on each supported
+ * columnar format.
+ *
+ * Each case starts from the same unpartitioned three-row table. The preparation proves table creation and seeding,
+ * the body applies one operation, and the assertions prove the exact rows and snapshot delta it produced. The
+ * operation definitions can run on additional preparations without changing their assertions.
+ *
+ * Case families: six operations over Parquet and ORC, contributing 12 cases.
+ */
+trait ScenarioCoreDml extends TableTestFixtures {
+  import Rows._
+
+  /** One representative from every principal read and write family, on each standard format. */
+  lazy val dmlCoreCases: List[TestCase] =
+    preparedCoreFormats.flatMap(preparation =>
+      coreDmlOperations.map(_.runOn(preparation)))
+
+  /** The six operations that prove the complete table lifecycle and each principal DML family. */
+  lazy val coreDmlOperations: List[DmlTestCase[CoreTable.type]] =
+    List(
+      readProjection,
+      insertInto,
+      insertOverwrite,
+      deleteByPredicate,
+      updateByPredicate,
+      mergeUpsert)
+
+  /**
+   * SELECT of foo_col_string alone agrees with the same column read through the full table state and leaves that state
+   * unchanged.
+   */
+  private val readProjection: DmlTestCase[CoreTable.type] =
+    DmlTestCase(
+      "read.projection",
+      table => {
+        val before = table.state
+        val projected = table.spark
+          .sql(
+            s"SELECT ${Core.string0.columnName} FROM ${table.name} " +
+              s"ORDER BY ${Core.long0.columnName}")
+          .collect()
+          .toSeq
+          .map(_.get(Core.string0))
+        val after = table.state
+
+        assert(
+          projected == before.rows.sortBy(_.get(Core.long0)).map(_.get(Core.string0)),
+          s"projection returned $projected")
+        assert(after == before, "a read leaves the rows and the snapshot count unchanged")
+      })
+
+  /**
+   * INSERT INTO appends two literal rows, leaves every prepared row unchanged, and commits one snapshot.
+   */
+  private val insertInto: DmlTestCase[CoreTable.type] =
+    DmlTestCase(
+      "insert.into",
+      table => {
+        val before = table.state
+
+        table.spark.sql(
+          s"""INSERT INTO ${table.name} VALUES
+                (CAST(4 AS BIGINT), 4, 'row-4', 4.5, true,  '2024-01-01-03'),
+                (CAST(5 AS BIGINT), 5, 'row-5', 5.5, false, '2024-01-01-04')""")
+        val after = table.state
+
+        assert(
+          after.rows == inKeyOrder(before.rows ++ Seq(
+            Row(4L, 4, "row-4", 4.5, true, "2024-01-01-03"),
+            Row(5L, 5, "row-5", 5.5, false, "2024-01-01-04"))),
+          s"rows after the INSERT: ${after.rows}")
+        assert(
+          after.snapshotCount == before.snapshotCount + 1,
+          "INSERT INTO commits one snapshot")
+      })
+
+  /**
+   * INSERT OVERWRITE replaces the table contents with two literal rows and commits one snapshot.
+   */
+  private val insertOverwrite: DmlTestCase[CoreTable.type] =
+    DmlTestCase(
+      "insert.overwrite",
+      table => {
+        val before = table.state
+
+        table.spark.sql(
+          s"""INSERT OVERWRITE ${table.name} VALUES
+                (CAST(1 AS BIGINT), 1, 'p', 1.5, false, '2024-01-01-00'),
+                (CAST(2 AS BIGINT), 2, 'q', 2.5, true,  '2024-01-02-01')""")
+        val after = table.state
+
+        assert(
+          after.rows == Seq(
+            Row(1L, 1, "p", 1.5, false, "2024-01-01-00"),
+            Row(2L, 2, "q", 2.5, true, "2024-01-02-01")),
+          s"rows after the overwrite: ${after.rows}")
+        assert(
+          after.snapshotCount == before.snapshotCount + 1,
+          "INSERT OVERWRITE commits one snapshot")
+      })
+
+  /**
+   * DELETE WHERE foo_col_long < 2 removes the rows below key 2, leaves every other row unchanged, and commits one
+   * snapshot.
+   */
+  private val deleteByPredicate: DmlTestCase[CoreTable.type] =
+    DmlTestCase(
+      "delete.byPredicate",
+      table => {
+        val before = table.state
+
+        table.spark.sql(
+          s"DELETE FROM ${table.name} WHERE ${Core.long0.columnName} < 2")
+        val after = table.state
+
+        assert(
+          after.rows == before.rows.filterNot(_.get(Core.long0) < 2),
+          s"rows after the DELETE: ${after.rows}")
+        assert(
+          after.snapshotCount == before.snapshotCount + 1,
+          "DELETE by a predicate commits one snapshot")
+      })
+
+  /**
+   * UPDATE SET foo_col_string = 'X' WHERE foo_col_long = 2 rewrites that column for key 2 only, leaves every other row
+   * unchanged, and commits one snapshot.
+   */
+  private val updateByPredicate: DmlTestCase[CoreTable.type] =
+    DmlTestCase(
+      "update.byPredicate",
+      table => {
+        val before = table.state
+
+        table.spark.sql(
+          s"UPDATE ${table.name} SET ${Core.string0.columnName} = 'X' " +
+            s"WHERE ${Core.long0.columnName} = 2")
+        val after = table.state
+
+        assert(
+          after.rows == before.rows.map(row =>
+            if (row.get(Core.long0) == 2L) withColumnValue(row, Core.string0, "X") else row),
+          s"rows after the UPDATE: ${after.rows}")
+        assert(
+          after.snapshotCount == before.snapshotCount + 1,
+          "UPDATE by a predicate commits one snapshot")
+      })
+
+  /**
+   * MERGE with an UPDATE clause and an INSERT clause rewrites key 2, appends key 7, and commits one snapshot.
+   */
+  private val mergeUpsert: DmlTestCase[CoreTable.type] =
+    DmlTestCase(
+      "merge.upsert",
+      table => {
+        val before = table.state
+
+        table.spark.sql(
+          s"""MERGE INTO ${table.name} t USING (
+                SELECT * FROM VALUES
+                  (CAST(2 AS BIGINT), 2, 'U', 2.5, true,  '2024-01-02-01'),
+                  (CAST(7 AS BIGINT), 7, 'g', 7.5, false, '2024-01-01-06')
+                AS s($columnNameList)
+              ) s ON t.${Core.long0.columnName} = s.${Core.long0.columnName}
+              WHEN MATCHED THEN UPDATE
+              SET t.${Core.string0.columnName} = s.${Core.string0.columnName}
+              WHEN NOT MATCHED THEN INSERT *""")
+        val after = table.state
+
+        assert(
+          after.rows == inKeyOrder(
+            before.rows.map(row =>
+              if (row.get(Core.long0) == 2L) withColumnValue(row, Core.string0, "U") else row) :+
+              Row(7L, 7, "g", 7.5, false, "2024-01-01-06")),
+          s"rows after the MERGE: ${after.rows}")
+        assert(
+          after.snapshotCount == before.snapshotCount + 1,
+          "an upsert MERGE commits one snapshot")
+      })
+}
diff --git a/integrations/spark/delta-harness/src/main/scala/harness/openhouse/scenarios/ScenarioDataType.scala b/integrations/spark/delta-harness/src/main/scala/harness/openhouse/scenarios/ScenarioDataType.scala
new file mode 100644
index 000000000..6d21d900d
--- /dev/null
+++ b/integrations/spark/delta-harness/src/main/scala/harness/openhouse/scenarios/ScenarioDataType.scala
@@ -0,0 +1,174 @@
+package harness
+
+import java.math.BigDecimal
+import java.nio.charset.StandardCharsets
+import java.sql.Date
+import java.time.{Instant, LocalDateTime}
+
+/**
+ * Scalar data types: representative round-trip, null, numeric-boundary, special-floating-value, and string behavior
+ * for the typed scalar table.
+ *
+ * Operations: a round trip of the seeded long, int, double, decimal and string values; an INSERT of an all-null row;
+ * an INSERT of the special double values NaN and Infinity; an INSERT at the long, int and decimal boundaries; and an
+ * INSERT of a unicode string and an empty string.
+ *
+ * Preparation axes: one unpartitioned TypesTable layout per file format, each seeded with three rows covering every
+ * scalar column.
+ *
+ * Case families: five families over two layouts, contributing 10 cases.
+ */
+trait ScenarioDataType extends TableTestFixtures {
+
+  /** Every scalar-type case, one layout at a time. */
+  lazy val dataTypeCases: List[TestCase] =
+    preparedTypesTables.flatMap(preparation =>
+      List(
+        roundtripCase(preparation),
+        nullsCase(preparation),
+        specialFloatsCase(preparation),
+        boundariesCase(preparation),
+        unicodeAndEmptyCase(preparation)))
+
+  /** One unpartitioned scalar-type table per file format. */
+  lazy val typesLayouts: List[Layout] =
+    fileFormats.map(format =>
+      Layout(
+        s"types-unpartitioned/$format",
+        table =>
+          s"CREATE TABLE $table (${TypesTable.columnDefinitions}) USING $dataSource " +
+            s"TBLPROPERTIES ('write.format.default'='$format')"))
+
+  /** One preparation per scalar-type layout: the table is created, then seeded with three fully valued rows. */
+  lazy val preparedTypesTables: List[TablePreparation[TypesTable.type]] =
+    typesLayouts.map(layout =>
+      TablePreparation(
+        layout.label,
+        TableTest(TypesTable).sql("create")(layout.create)().insert(standardSeedRowCount)()))
+
+  // A fully valued TypesTable row with the given ID; each case supplies the columns it is about.
+  private def typesRow(id: Long, n: String, x: String, dec: String, str: String): String =
+    s"(CAST($id AS BIGINT), $n, $x, $dec, $str, CAST('b' AS binary), DATE '2024-01-01', " +
+      s"TIMESTAMP '2024-01-01 00:00:00', TIMESTAMP_NTZ '2024-01-01 00:00:00')"
+
+  /**
+   * Selecting the first seeded row reads back every scalar value exactly.
+   */
+  private def roundtripCase(preparation: TablePreparation[TypesTable.type]): TestCase =
+    preparation.test("types.roundtrip") { table =>
+      val row = table.spark
+        .sql(
+          s"SELECT id, n, x, dec, str, bin, dt, ts, tsntz FROM ${table.name} WHERE id = 1")
+        .collect()(0)
+
+      assert(
+        row.getLong(0) == 1L &&
+          row.getInt(1) == 1 &&
+          row.getDouble(2) == 1.5,
+        s"unexpected numeric values: ${row.toSeq}")
+      assert(
+        row.getDecimal(3).compareTo(new BigDecimal("1.50")) == 0,
+        s"unexpected decimal value: ${row.getDecimal(3)}")
+      assert(row.getString(4) == "row-1", s"unexpected string value: ${row.getString(4)}")
+      assert(
+        java.util.Arrays.equals(
+          row.getAs[Array[Byte]](5),
+          "bin-1".getBytes(StandardCharsets.UTF_8)),
+        s"unexpected binary value: ${row.getAs[Array[Byte]](5).mkString("[", ",", "]")}")
+      assert(
+        row.getDate(6) == Date.valueOf("2024-01-01"),
+        s"unexpected date value: ${row.getDate(6)}")
+      assert(
+        row.getTimestamp(7).toInstant == Instant.parse("2024-01-01T00:00:00Z"),
+        s"unexpected timestamp value: ${row.getTimestamp(7)}")
+      assert(
+        row.getAs[LocalDateTime](8) == LocalDateTime.of(2024, 1, 1, 0, 0),
+        s"unexpected timestamp_ntz value: ${row.getAs[LocalDateTime](8)}")
+    }
+
+  /**
+   * Inserting a row with every non-key column NULL reads every non-key column back as null.
+   */
+  private def nullsCase(preparation: TablePreparation[TypesTable.type]): TestCase =
+    preparation.test("types.nulls") { table =>
+      table.spark.sql(
+        s"INSERT INTO ${table.name} VALUES (" +
+          "CAST(10 AS BIGINT), NULL, NULL, NULL, NULL, " +
+          "NULL, NULL, NULL, NULL)")
+
+      val row = table.spark
+        .sql(
+          s"SELECT n, x, dec, str, bin, dt, ts, tsntz FROM ${table.name} WHERE id = 10")
+        .collect()(0)
+
+      assert((0 until 8).forall(row.isNullAt))
+    }
+
+  /** Inserting rows with double('NaN') and double('Infinity') reads back as NaN and positive infinity respectively. */
+  private def specialFloatsCase(preparation: TablePreparation[TypesTable.type]): TestCase =
+    preparation.test("types.specialFloats") { table =>
+      table.spark.sql(
+        s"INSERT INTO ${table.name} VALUES " +
+          s"${typesRow(11, "0", "double('NaN')", "CAST(0 AS decimal(10,2))", "'x'")}, " +
+          s"${typesRow(12, "0", "double('Infinity')", "CAST(0 AS decimal(10,2))", "'y'")}")
+
+      assert(
+        table.spark
+          .sql(s"SELECT x FROM ${table.name} WHERE id = 11")
+          .collect()(0)
+          .getDouble(0)
+          .isNaN)
+      assert(
+        table.spark
+          .sql(s"SELECT x FROM ${table.name} WHERE id = 12")
+          .collect()(0)
+          .getDouble(0) == Double.PositiveInfinity)
+    }
+
+  /**
+   * Inserting a row at Long.MaxValue, Int.MaxValue and a max-precision decimal reads those boundary values back
+   * unchanged.
+   */
+  private def boundariesCase(preparation: TablePreparation[TypesTable.type]): TestCase =
+    preparation.test("types.boundaries") { table =>
+      table.spark.sql(
+        s"INSERT INTO ${table.name} VALUES " +
+          typesRow(
+            Long.MaxValue,
+            Int.MaxValue.toString,
+            "0.0",
+            "CAST(99999999.99 AS decimal(10,2))",
+            "'max'"))
+
+      val row = table.spark
+        .sql(
+          s"SELECT id, n, dec FROM ${table.name} WHERE str = 'max'")
+        .collect()(0)
+
+      assert(
+        row.getLong(0) == Long.MaxValue &&
+          row.getInt(1) == Int.MaxValue)
+      assert(
+        row.getDecimal(2).compareTo(new BigDecimal("99999999.99")) == 0)
+    }
+
+  /** Inserting rows with a unicode string and an empty string reads each back unchanged. */
+  private def unicodeAndEmptyCase(preparation: TablePreparation[TypesTable.type]): TestCase =
+    preparation.test("types.unicodeAndEmpty") { table =>
+      table.spark.sql(
+        s"INSERT INTO ${table.name} VALUES " +
+          s"${typesRow(13, "0", "0.0", "CAST(0 AS decimal(10,2))", "'\u65e5\u672c\u8a9e \uD83C\uDF89'")}, " +
+          s"${typesRow(14, "0", "0.0", "CAST(0 AS decimal(10,2))", "''")}")
+
+      assert(
+        table.spark
+          .sql(s"SELECT str FROM ${table.name} WHERE id = 13")
+          .collect()(0)
+          .getString(0) == "\u65e5\u672c\u8a9e \uD83C\uDF89")
+      assert(
+        table.spark
+          .sql(s"SELECT str FROM ${table.name} WHERE id = 14")
+          .collect()(0)
+          .getString(0) == "")
+    }
+}
diff --git a/integrations/spark/delta-harness/src/main/scala/harness/openhouse/scenarios/ScenarioDmlRejection.scala b/integrations/spark/delta-harness/src/main/scala/harness/openhouse/scenarios/ScenarioDmlRejection.scala
new file mode 100644
index 000000000..3a4d3a5b7
--- /dev/null
+++ b/integrations/spark/delta-harness/src/main/scala/harness/openhouse/scenarios/ScenarioDmlRejection.scala
@@ -0,0 +1,131 @@
+package harness
+
+import org.apache.spark.sql.AnalysisException
+
+/**
+ * DML rejection: the DML statements the analyzer and the row-level rewrite refuse, and the message each rejection
+ * carries.
+ *
+ * Operations: DELETE on a column the table does not declare, DELETE and UPDATE with a nondeterministic predicate,
+ * INSERT INTO with fewer values than the table has columns, a MERGE whose UPDATE SET assigns one target column twice,
+ * and a MERGE whose source matches one target row twice.
+ *
+ * Preparation axes: the standard seeded core table in each columnar format.
+ *
+ * Case families: six families contributing 12 cases.
+ */
+trait ScenarioDmlRejection extends TableTestFixtures {
+
+  /** Every rejected-DML case, one file format at a time. */
+  lazy val dmlRejectionCases: List[TestCase] =
+    preparedCoreFormats.flatMap { preparation =>
+      List(
+        nonExistentColumnCase(preparation),
+        nonDeterministicDeleteCase(preparation),
+        nonDeterministicUpdateCase(preparation),
+        insertArityCase(preparation),
+        mergeConflictingUpdatesCase(preparation),
+        mergeCardinalityViolationCase(preparation))
+    }
+
+  /** DELETE with a WHERE clause on a nonexistent column is rejected with an AnalysisException naming that column. */
+  private def nonExistentColumnCase(preparation: TablePreparation[CoreTable.type]): TestCase =
+    preparation.test("dmlValidation.nonExistentColumn") { table =>
+      val exception = Check.intercept[AnalysisException](
+        table.spark.sql(
+          s"DELETE FROM ${table.name} WHERE no_such_column = 1"))
+
+      assert(exception.getMessage.contains("no_such_column"))
+    }
+
+  /**
+   * DELETE with a nondeterministic WHERE clause (rand() < 0.5) is rejected with an AnalysisException about
+   * determinism.
+   */
+  private def nonDeterministicDeleteCase(preparation: TablePreparation[CoreTable.type]): TestCase =
+    preparation.test("dmlValidation.nonDeterministicDelete") { table =>
+      val exception = Check.intercept[AnalysisException](
+        table.spark.sql(
+          s"DELETE FROM ${table.name} WHERE rand() < 0.5"))
+
+      assert(exception.getMessage.toLowerCase.contains("deterministic"))
+    }
+
+  /**
+   * UPDATE with a nondeterministic WHERE clause (rand() < 0.5) is rejected with an AnalysisException about
+   * determinism.
+   */
+  private def nonDeterministicUpdateCase(preparation: TablePreparation[CoreTable.type]): TestCase =
+    preparation.test("dmlValidation.nonDeterministicUpdate") { table =>
+      val exception = Check.intercept[AnalysisException](
+        table.spark.sql(
+          s"UPDATE ${table.name} SET ${Core.string0.columnName} = 'x' WHERE rand() < 0.5"))
+
+      assert(exception.getMessage.toLowerCase.contains("deterministic"))
+    }
+
+  /**
+   * INSERT INTO with too few values for the table's columns is rejected with an AnalysisException about the missing
+   * data columns.
+   */
+  private def insertArityCase(preparation: TablePreparation[CoreTable.type]): TestCase =
+    preparation.test("dmlValidation.insertArity") { table =>
+      val exception = Check.intercept[AnalysisException](
+        table.spark.sql(
+          s"INSERT INTO ${table.name} VALUES (CAST(1 AS BIGINT), 1)"))
+
+      assert(exception.getMessage.toLowerCase.contains("not enough data columns"))
+    }
+
+  /**
+   * A MERGE whose UPDATE SET assigns the same target column twice is rejected with an AnalysisException about multiple
+   * assignments.
+   */
+  private def mergeConflictingUpdatesCase(
+      preparation: TablePreparation[CoreTable.type]): TestCase =
+    preparation.test("dmlValidation.mergeConflictingUpdates") { table =>
+      val keyColumn = Core.long0.columnName
+      val stringColumn = Core.string0.columnName
+      val exception = Check.intercept[AnalysisException](
+        table.spark.sql(
+          s"""MERGE INTO ${table.name} target USING (
+                    SELECT * FROM VALUES (CAST(2 AS BIGINT)) AS source($keyColumn)
+                  ) source
+                  ON target.$keyColumn = source.$keyColumn
+                  WHEN MATCHED THEN UPDATE
+                  SET target.$stringColumn = 'a', target.$stringColumn = 'b'"""))
+
+      assert(exception.getMessage.contains("Multiple assignments"))
+    }
+
+  /**
+   * A MERGE whose source has two rows matching the same target row fails with a cardinality-violation error naming the
+   * multi-row match.
+   */
+  private def mergeCardinalityViolationCase(
+      preparation: TablePreparation[CoreTable.type]): TestCase =
+    preparation.test("dmlValidation.mergeCardinalityViolation") { table =>
+      val keyColumn = Core.long0.columnName
+      val stringColumn = Core.string0.columnName
+      val before = table.state
+      val exception = Check.intercept[Exception](
+        table.spark.sql(
+          s"""MERGE INTO ${table.name} target USING (
+                    SELECT * FROM VALUES
+                      (CAST(2 AS BIGINT), 'a'),
+                      (CAST(2 AS BIGINT), 'b')
+                    AS source($keyColumn, $stringColumn)
+                  ) source
+                  ON target.$keyColumn = source.$keyColumn
+                  WHEN MATCHED THEN UPDATE SET target.$stringColumn = source.$stringColumn"""))
+
+      assert(
+        Exceptions.causeChain(exception).exists { cause =>
+          Option(cause.getMessage).exists(
+            _.contains("matched a single row from the target table"))
+        },
+        s"expected a MERGE cardinality-violation message, got: ${exception.getMessage}")
+      val after = table.state
+      assert(after == before, s"rejected MERGE changed table state: before=$before, after=$after")
+    }
+}
diff --git a/integrations/spark/delta-harness/src/main/scala/harness/openhouse/scenarios/TableTestFixtures.scala b/integrations/spark/delta-harness/src/main/scala/harness/openhouse/scenarios/TableTestFixtures.scala
new file mode 100644
index 000000000..7d6e1b48e
--- /dev/null
+++ b/integrations/spark/delta-harness/src/main/scala/harness/openhouse/scenarios/TableTestFixtures.scala
@@ -0,0 +1,81 @@
+package harness
+
+import org.apache.spark.sql.Row
+
+/**
+ * Table fixtures used by the foundation scenarios: the core table shape, row helpers, file formats, standard seed,
+ * standard unpartitioned preparations, and late-bound Spark data source.
+ *
+ * Capability layers build their specialized preparations from these table primitives in the layer that first uses
+ * them.
+ */
+trait TableTestFixtures {
+
+  protected val Core = CoreTable            // brevity in the typed column references below
+  protected val columnNameList = Core.columnNames.mkString(", ") // source column list, so renames propagate
+
+  // The rows a case reads back are ordered by the long column and carry exactly the core columns in their declared
+  // order, so an expected row set is written as the rows the case started from, filtered, mapped through
+  // `withColumnValue`, extended with literal rows, and re-sorted. Both helpers address columns by position so they also
+  // work on the literal rows a case writes out.
+  private def columnPosition(column: Column[_]): Int = Core.columnNames.indexOf(column.columnName)
+
+  protected def withColumnValue[T](row: Row, column: Column[T], value: T): Row =
+    Row.fromSeq(row.toSeq.updated(columnPosition(column), value))
+
+  protected def inKeyOrder(rows: Seq[Row]): Seq[Row] =
+    rows.sortBy(_.getLong(columnPosition(Core.long0)))
+
+  // A layout belongs to its preparation, so one test body can run on each table shape the scenario declares.
+  protected val columnDefinitions =
+    "foo_col_long bigint, foo_col_int int, foo_col_string string, foo_col_double double, " +
+      "foo_col_boolean boolean, foo_col_date string"
+
+  /** One starting table shape: the label that names it in a case ID and the CREATE statement that builds it. */
+  final case class Layout(label: String, create: String => String)
+
+  /**
+   * Every file format the harness runs on. This is the single source for a format list, so every format-crossed family
+   * covers both columnar formats.
+   */
+  val fileFormats: List[String] = List("parquet", "orc")
+
+  /** One core table in `format`, shaped by `partitionClause` and labelled for its case IDs. */
+  protected def coreLayout(label: String, format: String, partitionClause: String): Layout =
+    Layout(
+      label,
+      table =>
+        s"CREATE TABLE $table ($columnDefinitions) USING $dataSource $partitionClause " +
+          s"TBLPROPERTIES ('write.format.default'='$format')")
+
+  /** The foundation's unpartitioned core layout in each file format. */
+  val coreLayouts: List[Layout] =
+    fileFormats.map(format => coreLayout(format, format, ""))
+
+  /**
+   * The standard seed writes three deterministic rows with keys 1, 2 and 3. Row `n` holds key `n` in the long column,
+   * `n` in the int column, `row-n` in the string column, `n.5` in the double column, `n % 2 == 0` in the boolean
+   * column, and the date value `n - 1` hours after `2024-01-01-00`. `RowGenerator` builds those literals from
+   * `CoreTable`, so the seed follows a column rename. Every preparation that seeds a core table writes exactly these
+   * rows, so a case that starts from a seeded table knows its three starting keys.
+   */
+  val standardSeedRowCount: Int = 3
+
+  /** Creates a core table under `layout` and leaves it empty. */
+  protected def createCoreTable(layout: Layout): TableTest[CoreTable.type] =
+    TableTest(Core).sql("create")(layout.create)()
+
+  /** An unpartitioned core table in `format`, created and seeded with the standard rows. */
+  protected def preparedStandardTable(format: String): TablePreparation[CoreTable.type] =
+    TablePreparation(
+      format,
+      createCoreTable(coreLayout(format, format, "")).insert(standardSeedRowCount)())
+
+  /** The standard seeded table in each file format. */
+  val preparedCoreFormats: List[TablePreparation[CoreTable.type]] =
+    fileFormats.map(preparedStandardTable)
+
+  // The Spark data source used by CREATE TABLE statements. A remote adapter sets this before reading Catalog.cases.
+  var dataSource: String = "iceberg"
+
+}
diff --git a/integrations/spark/delta-harness/src/test/scala/harness/framework/RowGeneratorTest.scala b/integrations/spark/delta-harness/src/test/scala/harness/framework/RowGeneratorTest.scala
new file mode 100644
index 000000000..08009550b
--- /dev/null
+++ b/integrations/spark/delta-harness/src/test/scala/harness/framework/RowGeneratorTest.scala
@@ -0,0 +1,17 @@
+package harness
+
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Test
+
+final class RowGeneratorTest {
+  @Test
+  def coreSeedLiteralsAreStable(): Unit = {
+    assertEquals(
+      "VALUES " +
+        "(1, 1, 'row-1', 1.5, false, '2024-01-01-00'), " +
+        "(2, 2, 'row-2', 2.5, true, '2024-01-01-01'), " +
+        "(3, 3, 'row-3', 3.5, false, '2024-01-01-02')",
+      RowGenerator.valuesClause(CoreTable, 3))
+    assertEquals("2024-01-02-00", CoreTable.dateLiteral(25))
+  }
+}
diff --git a/integrations/spark/delta-harness/src/test/scala/harness/framework/RunnerTest.scala b/integrations/spark/delta-harness/src/test/scala/harness/framework/RunnerTest.scala
new file mode 100644
index 000000000..17591263c
--- /dev/null
+++ b/integrations/spark/delta-harness/src/test/scala/harness/framework/RunnerTest.scala
@@ -0,0 +1,133 @@
+package harness
+
+import java.net.{ConnectException, SocketException, SocketTimeoutException}
+import java.util.concurrent.atomic.AtomicInteger
+import org.apache.spark.sql.SparkSession
+import org.junit.jupiter.api.Assertions.{assertEquals, assertFalse, assertSame, assertThrows, assertTrue}
+import org.junit.jupiter.api.Test
+import org.mockito.Mockito.{mock, when}
+
+final class RunnerTest {
+  @Test
+  def configuredParallelismRequiresAPositiveInteger(): Unit = {
+    assertEquals(8, RunnerConfiguration.parallelism(Map.empty, 8))
+    assertEquals(1, RunnerConfiguration.parallelism(Map.empty, 0))
+    assertEquals(
+      4,
+      RunnerConfiguration.parallelism(Map("HARNESS_PARALLELISM" -> "4"), 8))
+
+    List("0", "-1", "many").foreach { value =>
+      val failure = assertThrows(
+        classOf[HarnessConfigurationException],
+        () =>
+          RunnerConfiguration.parallelism(
+            Map("HARNESS_PARALLELISM" -> value),
+            availableProcessors = 8))
+      assertEquals(
+        s"HARNESS_PARALLELISM must be a positive integer; received '$value'",
+        failure.getMessage)
+    }
+  }
+
+  @Test
+  def transientClassificationMatchesTheRetryContract(): Unit = {
+    assertTrue(Exceptions.isTransientConnectionFailure(new SocketTimeoutException("timeout")))
+    assertTrue(Exceptions.isTransientConnectionFailure(new ConnectException("refused")))
+    assertTrue(Exceptions.isTransientConnectionFailure(new SocketException("Connection reset by peer")))
+    assertTrue(
+      Exceptions.isTransientConnectionFailure(
+        new Exception("outer", new Exception("middle", new SocketTimeoutException("timeout")))))
+
+    assertFalse(Exceptions.isTransientConnectionFailure(new SocketException("broken pipe")))
+    assertFalse(Exceptions.isTransientConnectionFailure(new java.io.IOException("input failed")))
+    assertFalse(Exceptions.isTransientConnectionFailure(new AssertionError("wrong rows")))
+  }
+
+  @Test
+  def causeTraversalStopsAtCycles(): Unit = {
+    val cyclicFailure = new Exception("cycle") {
+      override def getCause: Throwable = this
+    }
+
+    assertEquals(List(cyclicFailure), Exceptions.causeChain(cyclicFailure))
+  }
+
+  @Test
+  def runnerRetriesOnlyTransientSessionCreationFailures(): Unit = {
+    val rootSpark = mock(classOf[SparkSession])
+    val freshSpark = mock(classOf[SparkSession])
+    val retryFailure =
+      new RuntimeException("session creation failed", new SocketTimeoutException("retry"))
+    when(rootSpark.newSession())
+      .thenThrow(retryFailure)
+      .thenReturn(freshSpark)
+    var receivedSpark = Option.empty[SparkSession]
+    val retryThenPass = TestCase("retry-then-pass", ctx => receivedSpark = Some(ctx.spark))
+
+    assertEquals(
+      (Outcome.Passed, 2),
+      Runner.execute(retryThenPass, Ctx(rootSpark, "openhouse.test")))
+    assertSame(freshSpark, receivedSpark.get)
+
+    val exhaustedSpark = mock(classOf[SparkSession])
+    val exhaustedFailure =
+      new RuntimeException("session creation failed", new SocketTimeoutException("exhausted"))
+    when(exhaustedSpark.newSession()).thenThrow(exhaustedFailure)
+    val caseRuns = new AtomicInteger()
+    val (exhaustedOutcome, exhaustedCount) = Runner.execute(
+      TestCase("exhaust-retries", _ => caseRuns.incrementAndGet()),
+      Ctx(exhaustedSpark, "openhouse.test"))
+
+    assertEquals(Runner.MaxAttempts, exhaustedCount)
+    assertEquals(0, caseRuns.get())
+    assertSame(exhaustedFailure, exhaustedOutcome.asInstanceOf[Outcome.Failed].cause)
+
+    val terminalSpark = mock(classOf[SparkSession])
+    val terminalFailure = new AssertionError("wrong rows")
+    when(terminalSpark.newSession()).thenThrow(terminalFailure)
+    val (terminalOutcome, terminalCount) = Runner.execute(
+      TestCase("terminal", _ => ()),
+      Ctx(terminalSpark, "openhouse.test"))
+
+    assertEquals(1, terminalCount)
+    assertSame(terminalFailure, terminalOutcome.asInstanceOf[Outcome.Failed].cause)
+  }
+
+  @Test
+  def runnerNeverRetriesAfterTheCaseStarts(): Unit = {
+    val rootSpark = mock(classOf[SparkSession])
+    val freshSpark = mock(classOf[SparkSession])
+    when(rootSpark.newSession()).thenReturn(freshSpark)
+
+    val assertionAttempts = new AtomicInteger()
+    val assertionFailure =
+      new AssertionError("unexpected exception", new SocketTimeoutException("nested timeout"))
+    val (assertionOutcome, assertionCount) = Runner.execute(
+      TestCase(
+        "assertion",
+        _ => {
+          assertionAttempts.incrementAndGet()
+          throw assertionFailure
+        }),
+      Ctx(rootSpark, "openhouse.test"))
+
+    assertEquals(1, assertionCount)
+    assertEquals(1, assertionAttempts.get())
+    assertSame(assertionFailure, assertionOutcome.asInstanceOf[Outcome.Failed].cause)
+
+    val cleanupAttempts = new AtomicInteger()
+    val cleanupFailure = new SocketTimeoutException("cleanup timeout")
+    val (cleanupOutcome, cleanupCount) = Runner.execute(
+      TestCase(
+        "cleanup",
+        _ => {
+          cleanupAttempts.incrementAndGet()
+          OwnedTableLifecycle.withCleanup(throw cleanupFailure)(())
+        }),
+      Ctx(rootSpark, "openhouse.test"))
+
+    assertEquals(1, cleanupCount)
+    assertEquals(1, cleanupAttempts.get())
+    assertSame(cleanupFailure, cleanupOutcome.asInstanceOf[Outcome.Failed].cause)
+  }
+}
diff --git a/integrations/spark/delta-harness/src/test/scala/harness/framework/TableLifecycleTest.scala b/integrations/spark/delta-harness/src/test/scala/harness/framework/TableLifecycleTest.scala
new file mode 100644
index 000000000..fc0d5a7c7
--- /dev/null
+++ b/integrations/spark/delta-harness/src/test/scala/harness/framework/TableLifecycleTest.scala
@@ -0,0 +1,77 @@
+package harness
+
+import java.util.concurrent.{Callable, Executors, TimeUnit}
+import org.junit.jupiter.api.Assertions.{assertEquals, assertSame, assertThrows}
+import org.junit.jupiter.api.Test
+
+final class TableLifecycleTest {
+  @Test
+  def cleanupRunsOnlyForAnOwnedTable(): Unit = {
+    var cleanupCount = 0
+    val createFailure = new Exception("table already exists")
+
+    val thrown = assertThrows(
+      classOf[Exception],
+      () =>
+        OwnedTableLifecycle.withOwnership(cleanupCount += 1)(_ =>
+          throw createFailure))
+
+    assertSame(createFailure, thrown)
+    assertEquals(0, cleanupCount)
+
+    OwnedTableLifecycle.withOwnership(cleanupCount += 1)(markTableCreated =>
+      markTableCreated())
+
+    assertEquals(1, cleanupCount)
+  }
+
+  @Test
+  def cleanupFailureIsSuppressedBehindTheBodyFailure(): Unit = {
+    val bodyFailure = new Exception("test failed")
+    val cleanupFailure = new Exception("cleanup failed")
+
+    val thrown = assertThrows(
+      classOf[Exception],
+      () =>
+        OwnedTableLifecycle.withOwnership(throw cleanupFailure) { markTableCreated =>
+          markTableCreated()
+          throw bodyFailure
+        })
+
+    assertSame(bodyFailure, thrown)
+    assertEquals(List(cleanupFailure), thrown.getSuppressed.toList)
+  }
+
+  @Test
+  def cleanupFailureAfterSuccessfulBodyIsPrimary(): Unit = {
+    val cleanupFailure = new Exception("cleanup failed")
+
+    val thrown = assertThrows(
+      classOf[Exception],
+      () => OwnedTableLifecycle.withCleanup(throw cleanupFailure)(()))
+
+    assertSame(cleanupFailure, thrown)
+  }
+
+  @Test
+  def generatedTableNamesStayUniqueAcrossCounterResets(): Unit = {
+    val pool = Executors.newFixedThreadPool(8)
+    try {
+      def generateNames(): List[String] = {
+        TableTest.seedCounter(0)
+        (1 to 500)
+          .map(_ =>
+            pool.submit(new Callable[String] {
+              override def call(): String = TableTest.nextQualifiedTableName("openhouse.test")
+            }))
+          .map(_.get(30, TimeUnit.SECONDS))
+          .toList
+      }
+
+      val names = generateNames() ++ generateNames()
+      assertEquals(names.size, names.distinct.size)
+    } finally {
+      pool.shutdownNow()
+    }
+  }
+}
diff --git a/integrations/spark/delta-harness/src/test/scala/harness/scenarios/CaseCatalogTest.scala b/integrations/spark/delta-harness/src/test/scala/harness/scenarios/CaseCatalogTest.scala
new file mode 100644
index 000000000..6df190d94
--- /dev/null
+++ b/integrations/spark/delta-harness/src/test/scala/harness/scenarios/CaseCatalogTest.scala
@@ -0,0 +1,82 @@
+package harness
+
+import org.junit.jupiter.api.Assertions.{assertEquals, assertTrue}
+import org.junit.jupiter.api.Test
+
+final class CaseCatalogTest {
+  @Test
+  def catalogContainsEachCoreScenarioInStableOrder(): Unit = {
+    assertEquals(
+      List("dataTypeCases", "dmlCoreCases", "dmlRejectionCases"),
+      Catalog.foundationContributions.map(_._1))
+    assertEquals(
+      List(
+        "types.roundtrip @ types-unpartitioned/parquet",
+        "types.nulls @ types-unpartitioned/parquet",
+        "types.specialFloats @ types-unpartitioned/parquet",
+        "types.boundaries @ types-unpartitioned/parquet",
+        "types.unicodeAndEmpty @ types-unpartitioned/parquet",
+        "types.roundtrip @ types-unpartitioned/orc",
+        "types.nulls @ types-unpartitioned/orc",
+        "types.specialFloats @ types-unpartitioned/orc",
+        "types.boundaries @ types-unpartitioned/orc",
+        "types.unicodeAndEmpty @ types-unpartitioned/orc",
+        "read.projection @ parquet",
+        "insert.into @ parquet",
+        "insert.overwrite @ parquet",
+        "delete.byPredicate @ parquet",
+        "update.byPredicate @ parquet",
+        "merge.upsert @ parquet",
+        "read.projection @ orc",
+        "insert.into @ orc",
+        "insert.overwrite @ orc",
+        "delete.byPredicate @ orc",
+        "update.byPredicate @ orc",
+        "merge.upsert @ orc",
+        "dmlValidation.nonExistentColumn @ parquet",
+        "dmlValidation.nonDeterministicDelete @ parquet",
+        "dmlValidation.nonDeterministicUpdate @ parquet",
+        "dmlValidation.insertArity @ parquet",
+        "dmlValidation.mergeConflictingUpdates @ parquet",
+        "dmlValidation.mergeCardinalityViolation @ parquet",
+        "dmlValidation.nonExistentColumn @ orc",
+        "dmlValidation.nonDeterministicDelete @ orc",
+        "dmlValidation.nonDeterministicUpdate @ orc",
+        "dmlValidation.insertArity @ orc",
+        "dmlValidation.mergeConflictingUpdates @ orc",
+        "dmlValidation.mergeCardinalityViolation @ orc"),
+      Catalog.foundationContributions.flatMap { case (_, contribution) => contribution.map(_.id) })
+
+    val duplicateCaseIds = Catalog.caseIds
+      .groupBy(identity)
+      .collect {
+        case (caseId, occurrences) if occurrences.size > 1 => caseId
+      }
+      .toList
+      .sorted
+
+    assertTrue(
+      duplicateCaseIds.isEmpty,
+      s"case IDs must be unique; duplicates=${duplicateCaseIds.mkString(", ")}")
+  }
+
+  @Test
+  def layoutsReadTheConfiguredDataSourceAtExecutionTime(): Unit = {
+    val originalDataSource = Scenarios.dataSource
+    try {
+      val coreLayouts = Scenarios.coreLayouts
+      val typesLayouts = Scenarios.typesLayouts
+      Catalog.cases
+
+      Scenarios.dataSource = "openhouse"
+      assertTrue(coreLayouts.forall(_.create("db.t").contains(" USING openhouse ")))
+      assertTrue(typesLayouts.forall(_.create("db.t").contains(" USING openhouse ")))
+
+      Scenarios.dataSource = "alternate"
+      assertTrue(coreLayouts.forall(_.create("db.t").contains(" USING alternate ")))
+      assertTrue(typesLayouts.forall(_.create("db.t").contains(" USING alternate ")))
+    } finally {
+      Scenarios.dataSource = originalDataSource
+    }
+  }
+}
diff --git a/integrations/spark/delta-harness/src/test/scala/harness/scenarios/TableTestFixturesTest.scala b/integrations/spark/delta-harness/src/test/scala/harness/scenarios/TableTestFixturesTest.scala
new file mode 100644
index 000000000..b49b2fd32
--- /dev/null
+++ b/integrations/spark/delta-harness/src/test/scala/harness/scenarios/TableTestFixturesTest.scala
@@ -0,0 +1,15 @@
+package harness
+
+import org.junit.jupiter.api.Assertions.{assertEquals, assertFalse}
+import org.junit.jupiter.api.Test
+
+final class TableTestFixturesTest {
+  private object Fixtures extends TableTestFixtures
+
+  @Test
+  def foundationUsesOnlyUnpartitionedCoreLayouts(): Unit = {
+    assertEquals(List("parquet", "orc"), Fixtures.coreLayouts.map(_.label))
+    assertFalse(
+      Fixtures.coreLayouts.exists(_.create("db.table").contains("PARTITIONED BY")))
+  }
+}
diff --git a/settings.gradle b/settings.gradle
index 9acc07d67..4f9b29e12 100644
--- a/settings.gradle
+++ b/settings.gradle
@@ -40,6 +40,7 @@ include ':integrations:spark:spark-3.1:openhouse-spark-runtime'
 include ':integrations:spark:spark-3.1:openhouse-spark-itest'
 include ':integrations:spark:spark-3.5:openhouse-spark-runtime'
 include ':integrations:spark:spark-3.5:openhouse-spark-itest'
+include ':integrations:spark:delta-harness'
 
 include ':iceberg:openhouse:htscatalog'
 include ':iceberg:openhouse:internalcatalog'
@@ -77,4 +78,5 @@ project(':integrations:java:iceberg-1.5:openhouse-java-runtime').name = 'openhou
 project(':integrations:java:iceberg-1.5:openhouse-java-itest').name = 'openhouse-java-iceberg-1.5-itest'
 project(':integrations:spark:spark-3.5:openhouse-spark-runtime').name = 'openhouse-spark-3.5-runtime_2.12'
 project(':integrations:spark:spark-3.5:openhouse-spark-itest').name = 'openhouse-spark-3.5-itest'
+project(':integrations:spark:delta-harness').name = 'openhouse-spark-delta-harness_2.12'
 project(':tables-test-fixtures:tables-test-fixtures-iceberg-1.5').name = 'tables-test-fixtures-iceberg-1.5_2.12'