diff --git a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/ConfigCoverageSpec.scala b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/ConfigCoverageSpec.scala
new file mode 100644
index 00000000000..9165b44d599
--- /dev/null
+++ b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/ConfigCoverageSpec.scala
@@ -0,0 +1,68 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.texera.amber.translator.verify
+
+import org.apache.texera.amber.operator.LogicalOp
+import org.apache.texera.amber.operator.source.SourceOperatorDescriptor
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+
+/**
+  * Holds the floor under what the harness can run.
+  *
+  * Every operator carrying standalone code is classified runnable or flagged, and
+  * a flag is how an operator stops being checked at all. For most that is a state
+  * someone is working through. For the ones below it would be a regression: they
+  * are the plainest operators the harness has, they take the shared table as it
+  * is, and nothing about them is hard to run. An operator arriving here means
+  * something upstream of the disposition broke rather than that this operator
+  * became difficult.
+  */
+class ConfigCoverageSpec extends AnyFlatSpec with Matchers {
+
+  private val mustRun = Set(
+    "IntersectOpDesc",
+    "DifferenceOpDesc",
+    "SymmetricDifferenceOpDesc",
+    "HashJoinOpDesc",
+    "SpecializedFilterOpDesc",
+    "SortOpDesc",
+    "LimitOpDesc"
+  )
+
+  "the harness" should "keep runnable every operator it must be able to run" in {
+    val flagged = OperatorBehaviorSpec
+      .discoverStandaloneOperators()
+      .filter(opClass => mustRun.contains(opClass.getSimpleName))
+      .filterNot(runnable)
+      .map(_.getSimpleName)
+    withClue(s"must-run operators no longer runnable: $flagged") {
+      flagged shouldBe empty
+    }
+  }
+
+  private def runnable(opClass: Class[_ <: LogicalOp]): Boolean =
+    if (classOf[SourceOperatorDescriptor].isAssignableFrom(opClass))
+      SourceCategoryRunner.canRun(opClass)
+    else
+      TransformVerificationRunner
+        .disposition(opClass)
+        .isInstanceOf[TransformVerificationRunner.Runnable]
+}
diff --git a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/OperatorBehaviorSpec.scala b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/OperatorBehaviorSpec.scala
new file mode 100644
index 00000000000..72abc6be0f3
--- /dev/null
+++ b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/OperatorBehaviorSpec.scala
@@ -0,0 +1,158 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.texera.amber.translator.verify
+
+import com.fasterxml.jackson.annotation.JsonSubTypes
+import org.apache.texera.amber.operator.{LogicalOp, StandaloneCodeGenerator}
+import org.apache.texera.amber.operator.source.SourceOperatorDescriptor
+import org.apache.texera.amber.translator.verify.tags.IntegrationTest
+import org.scalatest.ParallelTestExecution
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+
+/**
+  * Auto-discovered behavioral-parity tests: for every operator registered
+  * with [[LogicalOp]]'s `@JsonSubTypes` that implements
+  * [[StandaloneCodeGenerator]], emit a test that runs both Path A (Texera
+  * exec) and Path B (translator-generated Python via [[StandaloneRunner]])
+  * and asserts their outputs are equivalent.
+  *
+  * [[TransformVerificationRunner]] decides how each non-source transform is
+  * configured and whether it can run at all; sources go to
+  * [[SourceCategoryRunner]]. An operator it cannot run is registered as an
+  * ignored test carrying the reason, so the report lists every operator rather
+  * than reading as though the unrunnable ones do not exist.
+  *
+  * No edits to this spec are needed when a new operator is added — reflection
+  * discovers it automatically via `@JsonSubTypes`. The tier label appears in
+  * the test name so the report shows which path exercised each operator.
+  *
+  * Requires Python 3 with pandas on the [[Comparator]] / [[StandaloneRunner]]
+  * resolution chain (`UDF_PYTHON_PATH` env var, then `python3.12`).
+  */
+// Tagged @IntegrationTest: this is the only verify spec that forks a real
+// Python process end-to-end, so CI routes it to the Python-provisioned
+// integration job (see workflow-compiling-service/build.sbt WCS_TEST_FILTER).
+@IntegrationTest
+class OperatorBehaviorSpec extends AnyFlatSpec with Matchers with ParallelTestExecution {
+
+  // Build the test list at class construction. Each branch below registers
+  // one test (`in` for runnable, `ignore` for skipped) so the test report
+  // shows every translator-eligible operator and why it did or didn't run.
+  OperatorBehaviorSpec.discoverStandaloneOperators().foreach { opClass =>
+    val name = opClass.getSimpleName
+
+    if (!OperatorBehaviorSpec.isSelected(name)) {
+      // Narrowed out by VERIFY_ONLY / VERIFY_SKIP, which only a local run sets.
+      // Still registered, as an `ignore`, so the report lists every operator
+      // rather than reading as though the narrowed-out ones do not exist.
+      name should "NARROWED OUT — outside this run's VERIFY_ONLY / VERIFY_SKIP" ignore {}
+    } else if (classOf[SourceOperatorDescriptor].isAssignableFrom(opClass)) {
+      // Sources keep their handler-per-source design: each needs a real file
+      // in its specific format, which a generic fixture can't supply.
+      if (SourceCategoryRunner.canRun(opClass)) {
+        name should "produce equivalent output in Texera and standalone Python (source)" in {
+          SourceCategoryRunner.run(opClass)
+        }
+      } else {
+        name should s"FLAGGED — ${SourceCategoryRunner.flagReason(opClass)}" ignore {}
+      }
+    } else {
+      TransformVerificationRunner.disposition(opClass) match {
+        case TransformVerificationRunner.Runnable(tier) =>
+          name should s"produce equivalent output in Texera and standalone Python ($tier)" in {
+            TransformVerificationRunner.run(opClass)
+          }
+        case TransformVerificationRunner.Flagged(reason) =>
+          name should s"FLAGGED — $reason" ignore {
+            // Reason is in the test name so the report carries it; the
+            // coverage table in ConfigCoverageSpec aggregates these.
+          }
+      }
+    }
+  }
+
+  // Not one test per operator like the rest of this spec: it is one assertion
+  // over all of them, and it deliberately ignores the selection knobs above so a
+  // VERIFY_ONLY run still cannot hide a broken splice site.
+  "Generated standalone code" should "stay parseable when the column names are hostile" in {
+    StandaloneEscapingCheck.run() shouldBe empty
+  }
+
+  // Also one assertion over all of them: a workflow that draws two charts from
+  // one upstream hands both the same variable, and only a plan with a branch
+  // ever notices an operator writing to it.
+  it should "leave the frame it was handed alone" in {
+    StandaloneInputCheck.run() shouldBe empty
+  }
+}
+
+object OperatorBehaviorSpec {
+
+  // Narrowing knobs for a local run, both unset by default, so the default run
+  // is every operator: VERIFY_ONLY names the only ones to run, VERIFY_SKIP the
+  // ones to leave out. Case-sensitive substrings against the operator's simple
+  // name, comma-separated. Neither is set in CI, which therefore runs the lot.
+  //
+  // There is deliberately no third list withholding operators by default. What
+  // stays withheld is narrower than an operator and lives where it can say why:
+  // a single variant in [[TransformVerificationRunner.variantsNotRun]], or an
+  // operator that cannot be run at all in its `knownIssues`, each against an
+  // issue or a reason. A name here would withdraw an operator's every variant
+  // and record nothing about what is wrong with it.
+  private def patterns(envVar: String): Seq[String] =
+    sys.env.getOrElse(envVar, "").split(",").iterator.map(_.trim).filter(_.nonEmpty).toSeq
+
+  private lazy val onlyPatterns: Seq[String] = patterns("VERIFY_ONLY")
+  private lazy val skipPatterns: Seq[String] = patterns("VERIFY_SKIP")
+
+  /** True if `name` should run: in VERIFY_ONLY when that is set, and not in
+    * VERIFY_SKIP. True for everything when neither is set.
+    */
+  def isSelected(name: String): Boolean = {
+    val included = onlyPatterns.isEmpty || onlyPatterns.exists(name.contains)
+    val excluded = skipPatterns.exists(name.contains)
+    included && !excluded
+  }
+
+  /**
+    * Enumerates every concrete subclass of [[LogicalOp]] declared in its
+    * `@JsonSubTypes` annotation, filters to those implementing
+    * [[StandaloneCodeGenerator]], and returns them sorted by simple name
+    * (stable test report order).
+    *
+    * Uses the same registry Jackson uses to deserialize operators — no
+    * separate discovery mechanism needed. Adding an operator to
+    * `LogicalOp.@JsonSubTypes` makes it visible here automatically.
+    */
+  def discoverStandaloneOperators(): Seq[Class[_ <: LogicalOp]] = {
+    val annotation = classOf[LogicalOp].getAnnotation(classOf[JsonSubTypes])
+    if (annotation == null) Seq.empty
+    else
+      annotation
+        .value()
+        .toSeq
+        .map(_.value())
+        .filter(classOf[StandaloneCodeGenerator].isAssignableFrom)
+        .map(_.asInstanceOf[Class[_ <: LogicalOp]])
+        .distinct
+        .sortBy(_.getSimpleName)
+  }
+}
diff --git a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/SourceCategoryRunner.scala b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/SourceCategoryRunner.scala
new file mode 100644
index 00000000000..ab0561c0164
--- /dev/null
+++ b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/SourceCategoryRunner.scala
@@ -0,0 +1,466 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.texera.amber.translator.verify
+
+import org.apache.texera.amber.core.tuple.{Schema, Tuple}
+import org.apache.texera.amber.core.workflow.PortIdentity
+import org.apache.texera.amber.operator.LogicalOp
+import org.apache.texera.amber.operator.source.fetcher.URLFetcherOpDesc
+import org.apache.texera.amber.operator.source.scan.ScanSourceOpDesc
+import org.apache.texera.amber.operator.source.scan.file.{FileScanOpDesc, FileScanSourceOpDesc}
+import org.apache.texera.amber.operator.source.scan.text.TextInputSourceOpDesc
+import com.fasterxml.jackson.databind.node.ObjectNode
+import org.apache.texera.amber.util.JSONUtils.objectMapper
+import org.apache.arrow.memory.RootAllocator
+import org.apache.arrow.vector.ipc.ArrowFileWriter
+import org.apache.arrow.vector.VectorSchemaRoot
+import org.apache.texera.amber.util.ArrowUtils
+
+import java.nio.channels.FileChannel
+import java.nio.charset.{Charset, StandardCharsets}
+import java.nio.file.{Files, Path, StandardOpenOption}
+import scala.collection.mutable
+import scala.util.{Try, Using}
+
+/**
+  * Per-category test runner for source operators (operators with no input
+  * ports — they read from an external resource and emit tuples).
+  *
+  * Dispatch is auto-first. A scan source declares the format it reads via
+  * [[ScanSourceOpDesc.fileTypeName]], and where [[encoderByFileType]] knows that
+  * format the shared [[CanonicalSourceFixture]] is encoded into it with no
+  * per-operator code at all: a newly registered file-scan source in a known
+  * format is verified the moment it appears in [[LogicalOp]]'s `@JsonSubTypes`.
+  * A source that cannot take the shared table — the text family emits a single
+  * `line` column, and some carry their data inline — keeps a hand-written
+  * [[SourceHandler]] instead. Anything else is flagged, never silently skipped.
+  *
+  * The runner itself is operator-agnostic: it builds an OpDesc, drives
+  * [[OpExecHarness]] (Path A) and [[StandaloneRunner]] (Path B), compares via
+  * [[Comparator]]. Sources have no input ports so `inputs = Map.empty` for both.
+  */
+object SourceCategoryRunner {
+
+  /**
+    * The curated tier: sources that keep a hand-written handler because they
+    * can't go through the shared-fixture + encoder (auto) path — their output
+    * isn't the shared 3-column table (text-family, single `line` column) or
+    * their data is inline config rather than a file. Mirrors the transform
+    * side's [[CuratedHandlers]] (hand-written vs auto-generated fixture).
+    */
+  private val curatedHandlersByClass: Map[Class[_ <: LogicalOp], SourceHandler] =
+    Seq[SourceHandler](TextInputHandler, FileScanSourceHandler)
+      .map(h => h.opDescClass -> h)
+      .toMap
+
+  /**
+    * The auto tier. A scan source declares the file format it reads via
+    * [[ScanSourceOpDesc.fileTypeName]] ("CSV", "JSONL", "Arrow", …). Map that
+    * tag to the [[CanonicalSourceFixture]] encoder that writes a file in that
+    * format. Any source whose `fileTypeName` is a key here runs with zero
+    * per-operator code, so a newly added file-scan source in a known format is
+    * verified the moment it is registered in `@JsonSubTypes` — no handler, no
+    * edit here. (ParallelCSV also declares "CSV" and would be covered for free,
+    * but it is currently commented out of `@JsonSubTypes`, so the suite doesn't
+    * enumerate it.)
+    */
+  private val encoderByFileType: Map[String, (Path, Charset) => Path] = Map(
+    "CSV" -> CanonicalSourceFixture.writeCsv,
+    "CSVOld" -> CanonicalSourceFixture.writeCsv,
+    "JSONL" -> CanonicalSourceFixture.writeJsonl,
+    // Arrow is binary and its descriptor declares fileEncoding ignored, so the
+    // charset a variant asks for has nothing to apply to.
+    "Arrow" -> ((dir, _) => CanonicalSourceFixture.writeArrow(dir))
+  )
+
+  /**
+    * Sources this runner cannot verify, with the honest reason. Mirrors
+    * `TransformVerificationRunner.knownIssues`: the reason surfaces in the
+    * ignored test's name and the coverage table.
+    */
+  private val knownIssues: Map[Class[_ <: LogicalOp], String] = Map(
+    classOf[FileScanOpDesc] ->
+      ("input-driven source: filenames arrive on an input port at runtime, but this runner " +
+        "feeds sources no inputs — Path B's generated code references an undefined in1df"),
+    classOf[URLFetcherOpDesc] ->
+      ("live-network source: the operator fetches a real URL over the network, so its " +
+        "output is non-deterministic and depends on external connectivity — it cannot be " +
+        "verified against a fixed fixture in isolation")
+  )
+
+  /** The format tag a source declares, or `None` if it isn't an instantiable
+    * ScanSourceOpDesc (non-scan sources, or ones that fail to construct).
+    */
+  private def declaredFileType(opDescClass: Class[_ <: LogicalOp]): Option[String] =
+    Try(opDescClass.getDeclaredConstructor().newInstance()).toOption.collect {
+      case scan: ScanSourceOpDesc => scan.fileTypeName
+    }.flatten
+
+  def canRun(opDescClass: Class[_ <: LogicalOp]): Boolean =
+    curatedHandlersByClass.contains(opDescClass) ||
+      declaredFileType(opDescClass).exists(encoderByFileType.contains)
+
+  /**
+    * Tier label for a runnable source, mirroring the transform side's
+    * auto/curated distinction: `"curated source"` when a hand-written
+    * [[SourceHandler]] serves it, else `"auto source"` (a declared-format scan
+    * source fixtured by an [[encoderByFileType]] encoder with zero per-op code).
+    */
+  def tier(opDescClass: Class[_ <: LogicalOp]): String =
+    if (curatedHandlersByClass.contains(opDescClass)) "curated source" else "auto source"
+
+  /** Why a non-runnable source is flagged: a specific known issue, an
+    * unsupported declared format, or no handler/format match at all.
+    */
+  def flagReason(opDescClass: Class[_ <: LogicalOp]): String =
+    knownIssues.getOrElse(
+      opDescClass,
+      declaredFileType(opDescClass) match {
+        case Some(fileType) =>
+          s"unsupported source format '$fileType' — no encoder registered in SourceCategoryRunner"
+        case None => "no source handler registered yet"
+      }
+    )
+
+  private def newScanSource(opDescClass: Class[_ <: LogicalOp]): ScanSourceOpDesc =
+    opDescClass.getDeclaredConstructor().newInstance() match {
+      case s: ScanSourceOpDesc => s
+      case other =>
+        throw new IllegalArgumentException(
+          s"${opDescClass.getSimpleName} has no curated handler and is not a " +
+            s"ScanSourceOpDesc (${other.getClass.getName})"
+        )
+    }
+
+  /**
+    * Every configuration of one source worth running, as (label, op, its own
+    * directory).
+    *
+    * Each variant gets a directory of its own holding its OWN copy of the fixture,
+    * because the generated script reads the file by bare name (`pd.read_csv(
+    * "sample.csv")`) out of the directory it runs in. Two variants wanting two
+    * different `sample.csv` files cannot share one.
+    */
+  private def variantsFor(
+      opDescClass: Class[_ <: LogicalOp],
+      testRoot: Path
+  ): Seq[(String, LogicalOp, Path)] = {
+    // Punctuation collapses to '_', so two labels differing only in punctuation
+    // would name the same directory and share one fixture and one output dir. Fail
+    // loudly instead of letting a variant quietly run someone else's file.
+    val taken = mutable.Set.empty[String]
+    def dirFor(label: String): Path = {
+      val name = label.replaceAll("[^A-Za-z0-9]+", "_")
+      require(taken.add(name), s"two variants of $opDescClass both map to the directory '$name'")
+      Files.createDirectories(testRoot.resolve(name))
+    }
+
+    curatedHandlersByClass.get(opDescClass) match {
+      case Some(handler) =>
+        val baseDir = dirFor("default")
+        val base = handler.makeOpDesc(baseDir)
+        // Every variant calls the handler AGAIN rather than reusing `base`: the handler
+        // writes its fixture into the directory it is given, and a second op carrying
+        // the first one's `fileName` would read a file outside the directory it runs in.
+        // No enum sweep: both curated sources are the text family, whose `attributeType`
+        // says how to PARSE the fixture (`alice` is not an integer) and whose
+        // `fileEncoding` describes its BYTES — flipping either without rewriting the
+        // fixture compares nothing but how the two paths fail. The auto branch below
+        // rewrites its fixture per variant and does sweep them.
+        ConfigGenerator
+          .fullVariantEditsOf(base, Map.empty, handler.rowCount, sweepEnums = false)
+          .fold(
+            reason =>
+              throw new IllegalStateException(
+                s"cannot vary ${opDescClass.getSimpleName}: $reason"
+              ),
+            identity
+          )
+          .map { variant =>
+            if (variant.at.isEmpty) ("default", base, baseDir)
+            else {
+              val dir = dirFor(variant.label)
+              val op = ConfigGenerator
+                .applyVariant(handler.makeOpDesc(dir), variant)
+                .fold(
+                  reason =>
+                    throw new IllegalStateException(
+                      s"cannot build ${opDescClass.getSimpleName} variant '${variant.label}': $reason"
+                    ),
+                  identity
+                )
+              (variant.label, op, dir)
+            }
+          }
+      case None =>
+        val fileType = declaredFileType(opDescClass).getOrElse("")
+        val encoder = encoderByFileType.getOrElse(
+          fileType,
+          throw new IllegalArgumentException(
+            s"No encoder for ${opDescClass.getSimpleName} (fileTypeName='$fileType')"
+          )
+        )
+        val base = {
+          val dir = dirFor("default")
+          val op = newScanSource(opDescClass)
+          op.fileName = Some(encoder(dir, op.fileEncoding.getCharset).toUri.toString)
+          ("default", op: LogicalOp, dir)
+        }
+        base +: generatedVariants(opDescClass, encoder, dirFor)
+    }
+  }
+
+  /**
+    * The variants the shared [[ConfigGenerator]] derives from the operator's own
+    * fields — the base config with every knob filled, plus one per enum branch
+    * (`hasHeader`, JSONL's `flatten`). Nothing to register per operator: a knob
+    * added to a source is swept the day it is added.
+    *
+    * `fileEncoding` is swept like any other enum, and the fixture FOLLOWS it: each
+    * variant's file is written in the charset that variant declares. Encoding is a
+    * statement about the bytes, so a UTF_16 config over a file left in UTF-8 would
+    * only compare how each path fails.
+    *
+    * Variants that serialize identically are dropped — an operator that ignores
+    * `fileEncoding` (Arrow declares `@JsonIgnoreProperties`) would otherwise run the
+    * same config three times.
+    */
+  private def generatedVariants(
+      opDescClass: Class[_ <: LogicalOp],
+      encoder: (Path, Charset) => Path,
+      dirFor: String => Path
+  ): Seq[(String, LogicalOp, Path)] = {
+    val seen = mutable.Set.empty[String]
+    ConfigGenerator
+      .generateVariants(opDescClass, Map.empty, CanonicalSourceFixture.rows.size)
+      .fold(
+        reason =>
+          throw new IllegalStateException(
+            s"cannot auto-configure ${opDescClass.getSimpleName}: $reason"
+          ),
+        identity
+      )
+      .flatMap {
+        case (label, op) =>
+          val scan = op.asInstanceOf[ScanSourceOpDesc]
+          val shape = objectMapper.valueToTree[ObjectNode](scan)
+          shape.remove("fileName") // every variant reads its own copy of the file
+          if (!seen.add(shape.toString)) None
+          else {
+            // "default" is already the bare newInstance config above; this one is
+            // the generator's, which additionally fills limit and offset.
+            val name = if (label == "default") "auto-base" else label
+            val dir = dirFor(name)
+            scan.fileName = Some(encoder(dir, scan.fileEncoding.getCharset).toUri.toString)
+            Some((name, scan: LogicalOp, dir))
+          }
+      }
+  }
+
+  /** Runs the parity test for the operator, once per variant. Throws on mismatch. */
+  def run(opDescClass: Class[_ <: LogicalOp]): Unit = {
+    val testRoot = Files.createTempDirectory(s"op-behavior-${opDescClass.getSimpleName}-")
+    variantsFor(opDescClass, testRoot).foreach {
+      case (label, opDesc, workDir) =>
+        try runVariant(opDesc, workDir)
+        catch {
+          case e: Throwable =>
+            throw new AssertionError(s"[variant: $label] ${e.getMessage}", e)
+        }
+    }
+  }
+
+  /** Drive one configured source through both paths inside `workDir`, which holds
+    * that variant's fixture, and assert the two tables match.
+    */
+  private def runVariant(opDesc: LogicalOp, workDir: Path): Unit = {
+    val actualDir = workDir.resolve("actual")
+    Files.createDirectories(actualDir)
+
+    val pathA = OpExecHarness.execute(opDesc, inputs = Map.empty, outputDir = actualDir)
+    val pathB = StandaloneRunner.run(
+      opDesc = opDesc,
+      inputs = Map.empty,
+      outputPortCount = 1,
+      workDir = workDir
+    )
+
+    val actual = pathA.outputs(PortIdentity(0))
+    val expected = pathB.outputs(1)
+    Comparator.assertEqual(actual, expected)
+  }
+}
+
+/**
+  * A hand-written recipe for one source that can't use the auto tier
+  * (fileTypeName + [[CanonicalSourceFixture]] encoder): which OpDesc class it
+  * handles and how to fixture a working instance. Used for the text-family
+  * sources ([[TextInputHandler]], [[FileScanSourceHandler]]).
+  */
+trait SourceHandler {
+
+  /** The concrete OpDesc class this handler tests. */
+  def opDescClass: Class[_ <: LogicalOp]
+
+  /**
+    * Generate the fixture file inside `testRoot` and return a configured
+    * OpDesc instance whose `fileName` (or analogous URI field) points at it.
+    */
+  def makeOpDesc(testRoot: Path): LogicalOp
+
+  /** How many rows the fixture holds. Only the handler knows — it writes its own,
+    * rather than the shared [[CanonicalSourceFixture]]. A row-window knob the
+    * variants fill (`limit`, `offset`) is sized against this, so that the value they
+    * take keeps some rows and drops some instead of landing past the end.
+    */
+  def rowCount: Int
+}
+
+/**
+  * The rows every structured-file source reads: [[CanonicalFixture]]'s, whole.
+  *
+  * A source has no input port, so the fixture is delivered not as an input JSONL
+  * but as a file the operator opens itself. Each `writeXxx` encodes these rows
+  * into one on-disk format (CSV / JSONL / Arrow); a source handler picks the
+  * encoder its operator understands and points `fileName` at the result. So CSV,
+  * CSVOld, JSONL and Arrow all verify that the operator reconstructs one shared
+  * table, instead of each asserting against its own ad-hoc sample.
+  *
+  * It reads the canonical table rather than a narrow one of its own. A source
+  * fixture picked for the types that survive a round trip would be choosing not
+  * to ask the question this suite exists to ask: these files carry no types, both
+  * readers infer, and where they infer differently is exactly what should show. A
+  * date column does part them, and [[StandaloneRunner.sourceCasts]] is where that
+  * is settled — on Path B's reading, not by leaving the column out.
+  */
+object CanonicalSourceFixture {
+
+  val schema: Schema = CanonicalFixture.schema
+
+  val rows: Vector[Tuple] = CanonicalFixture.allRows
+
+  /** Write the rows as a header-first, comma-delimited CSV encoded in `charset`.
+    *
+    * The charset is a parameter because it describes the BYTES, not the config: a
+    * variant declaring `fileEncoding = UTF_16` over a file left in UTF-8 would
+    * compare nothing but how each path fails.
+    */
+  def writeCsv(dir: Path, charset: Charset): Path = {
+    val path = dir.resolve("sample.csv")
+    val header = schema.getAttributes.map(a => csvField(a.getName)).mkString(",")
+    val body = rows.map { t =>
+      schema.getAttributes
+        .map(a => csvField(Option(t.getField[AnyRef](a.getName)).map(_.toString).orNull))
+        .mkString(",")
+    }
+    Files.write(path, ((header +: body).mkString("\n") + "\n").getBytes(charset))
+    path
+  }
+
+  /** One CSV field, quoted per RFC 4180.
+    *
+    * The table carries commas inside values — a bracketed edge pair, a
+    * comma-delimited list, an ordinary English sentence — and writing those raw
+    * shifts every column after them. What the two paths then disagree about is a
+    * broken file rather than anything either of them does.
+    */
+  private def csvField(value: String): String =
+    if (value == null) ""
+    else if (value.exists(c => c == ',' || c == '"' || c == '\n' || c == '\r'))
+      "\"" + value.replace("\"", "\"\"") + "\""
+    else value
+
+  /** Write the rows as JSON Lines (one object per line, keys in schema order).
+    * Reuses [[TupleIO.writeTuples]] — the same writer the transform fixtures
+    * use; it also drops a `.schema.json` sidecar the source ignores.
+    *
+    * That writer is shared and always writes UTF-8, so a variant asking for another
+    * charset gets the bytes transcoded afterwards rather than a second writer.
+    */
+  def writeJsonl(dir: Path, charset: Charset): Path = {
+    val path = dir.resolve("sample.jsonl")
+    TupleIO.writeTuples(path, rows.iterator, schema)
+    if (charset != StandardCharsets.UTF_8) {
+      val text = new String(Files.readAllBytes(path), StandardCharsets.UTF_8)
+      Files.write(path, text.getBytes(charset))
+    }
+    path
+  }
+
+  /** Write the rows as an uncompressed Arrow IPC ("file" format) stream — the
+    * format both `ArrowFileReader` (Path A) and `pd.read_feather` (Path B)
+    * read.
+    */
+  def writeArrow(dir: Path): Path = {
+    val path = dir.resolve("sample.arrow")
+    // Texera's own Schema-to-Arrow mapping and tuple writer, so the file carries
+    // exactly the types `ArrowUtils.toTexeraSchema` reads back on the other side.
+    // Hand-listing the fields is what let the table outgrow them unnoticed: the
+    // columns past the list were simply not written, and both paths went on
+    // agreeing about the few that were.
+    val arrowSchema = ArrowUtils.fromTexeraSchema(schema)
+    Using.Manager { use =>
+      val allocator = use(new RootAllocator())
+      val root = use(VectorSchemaRoot.create(arrowSchema, allocator))
+      root.allocateNew()
+      rows.zipWithIndex.foreach { case (t, i) => ArrowUtils.setTexeraTuple(t, i, root) }
+      root.setRowCount(rows.size)
+      val channel = use(
+        FileChannel.open(path, StandardOpenOption.CREATE, StandardOpenOption.WRITE)
+      )
+      val writer = use(new ArrowFileWriter(root, null, channel))
+      writer.start()
+      writer.writeBatch()
+      writer.end()
+    }.get
+    path
+  }
+}
+
+/** Handler for `TextInputSourceOpDesc`. The text lives in the config — no fixture file. */
+object TextInputHandler extends SourceHandler {
+
+  override val opDescClass: Class[_ <: LogicalOp] = classOf[TextInputSourceOpDesc]
+
+  override val rowCount: Int = 3
+
+  override def makeOpDesc(testRoot: Path): LogicalOp = {
+    val desc = new TextInputSourceOpDesc()
+    desc.textInput = "alice\nbob\ncarol"
+    desc // defaults: attributeType STRING (one row per line), attributeName "line"
+  }
+}
+
+/** Handler for `FileScanSourceOpDesc`. Plain text file read in default line mode. */
+object FileScanSourceHandler extends SourceHandler {
+
+  override val opDescClass: Class[_ <: LogicalOp] = classOf[FileScanSourceOpDesc]
+
+  override val rowCount: Int = 3
+
+  override def makeOpDesc(testRoot: Path): LogicalOp = {
+    val txtPath = testRoot.resolve("sample.txt")
+    Files.write(txtPath, "alice\nbob\ncarol\n".getBytes(StandardCharsets.UTF_8))
+
+    val desc = new FileScanSourceOpDesc()
+    desc.fileName = Some(txtPath.toUri.toString)
+    desc // defaults: attributeType STRING (one row per line), attributeName "line"
+  }
+}
diff --git a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/StandaloneEscapingCheck.scala b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/StandaloneEscapingCheck.scala
new file mode 100644
index 00000000000..46564ada47a
--- /dev/null
+++ b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/StandaloneEscapingCheck.scala
@@ -0,0 +1,156 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.texera.amber.translator.verify
+
+import com.fasterxml.jackson.databind.JsonNode
+import com.fasterxml.jackson.databind.node.{ArrayNode, ObjectNode}
+import org.apache.texera.amber.operator.source.SourceOperatorDescriptor
+import org.apache.texera.amber.operator.{LogicalOp, StandaloneCodeGenerator}
+import org.apache.texera.amber.util.JSONUtils.objectMapper
+
+import java.nio.charset.StandardCharsets.UTF_8
+import java.nio.file.{Files, Path}
+import scala.jdk.CollectionConverters._
+import scala.sys.process.{Process, ProcessLogger}
+import scala.util.Try
+
+/**
+  * Configures every operator against hostile column names and parses what
+  * `generateStandaloneCode` produces. A value spliced in without
+  * `pyStringLiteral` is silent until someone names a column `a"b`, and then the
+  * exported script will not compile.
+  *
+  * The check has to be behavioural. A generator that builds its quoted literal
+  * inside a helper shows no quotes in its template, which is how RadarPlot and
+  * Aggregate survived a source-level sweep that reported zero remaining sites.
+  */
+object StandaloneEscapingCheck {
+
+  private val schemas = CanonicalFixture.schemasByPort
+  private val columns = CanonicalFixture.schema.getAttributes.map(_.getName).toSet
+
+  /** Every problem found, empty when the suite is clean. An operator whose config
+    * cannot be built lands here too: dropping it would check less while still
+    * passing.
+    */
+  def run(): Seq[String] = {
+    // Sources have no input schema and so no column knobs. Their free-text knobs
+    // already get a hostile variant in the normal verify run.
+    val operators = OperatorBehaviorSpec
+      .discoverStandaloneOperators()
+      .filterNot(classOf[SourceOperatorDescriptor].isAssignableFrom)
+    val dir = Files.createTempDirectory("standalone-escaping-")
+    dir.toFile.deleteOnExit()
+
+    val (unconfigurable, code) = operators
+      .map(op => op.getSimpleName -> codeFor(op, dir))
+      .partitionMap {
+        case (name, Left(why)) => Left(s"$name: $why")
+        case (name, Right(variants)) =>
+          Right(variants.map { case (label, src) => s"$name/$label" -> src })
+      }
+    unconfigurable ++ parse(code.flatten, dir)
+  }
+
+  /** Holds the four characters that end a Python literal, and carries the column
+    * it replaces so that two knobs never collide on one name.
+    */
+  private def hostile(column: String): String = "a\"b'c\\d\ne_" + column
+
+  private def hostilize(node: JsonNode): Unit =
+    node match {
+      case obj: ObjectNode =>
+        obj.fields().asScala.toSeq.foreach { e =>
+          if (isColumn(e.getValue)) obj.put(e.getKey, hostile(e.getValue.asText))
+          else hostilize(e.getValue)
+        }
+      case arr: ArrayNode =>
+        (0 until arr.size).foreach { i =>
+          if (isColumn(arr.get(i)))
+            arr.set(i, objectMapper.getNodeFactory.textNode(hostile(arr.get(i).asText)))
+          else hostilize(arr.get(i))
+        }
+      case _ => ()
+    }
+
+  private def isColumn(n: JsonNode): Boolean = n.isTextual && columns.contains(n.asText)
+
+  /** Split the way the runner splits: a curated operator's config is hand-written
+    * because the generator cannot derive one. Auto operators contribute every
+    * variant, since a knob only `optionals` fills is a site the base never reaches.
+    */
+  private[verify] def codeFor(
+      opClass: Class[_ <: LogicalOp],
+      dir: Path
+  ): Either[String, Seq[(String, String)]] = {
+    val configs = CuratedHandlers.byClass.get(opClass) match {
+      // The handler's config, not its enum sweep: sweeping moves an enum value,
+      // never a column name, and would need schemas only the runner holds.
+      case Some(h) =>
+        Try(Seq("curated" -> h.fixture(dir)._1)).toEither.left.map(e => s"curated: ${e.getMessage}")
+      case None => ConfigGenerator.generateVariants(opClass, schemas)
+    }
+    configs.flatMap { variants =>
+      val results = variants.map {
+        case (label, op) =>
+          val node = objectMapper.valueToTree[ObjectNode](op)
+          hostilize(node)
+          Try(
+            objectMapper
+              .treeToValue(node, opClass)
+              .asInstanceOf[StandaloneCodeGenerator]
+              .generateStandaloneCode()
+          ).toEither.left.map(e => s"$label: $e").map(label -> _)
+      }
+      results.collectFirst { case Left(why) => why }.toLeft(results.collect { case Right(r) => r })
+    }
+  }
+
+  /** One Python process for all of them; the cost is startup, not parsing. The
+    * snippets stay in `dir` so a reported operator can be opened as generated.
+    */
+  private def parse(snippets: Seq[(String, String)], dir: Path): Seq[String] = {
+    val payload = objectMapper.createObjectNode()
+    snippets.foreach { case (name, src) => payload.put(name, src) }
+    val input = write(dir, "snippets.json", payload.toString)
+    val script = write(
+      dir,
+      "parse_all.py",
+      """import ast, json, sys
+        |for name, code in json.load(open(sys.argv[1])).items():
+        |    try:
+        |        ast.parse(code)
+        |    except SyntaxError as e:
+        |        print(f"{name}: line {e.lineno}: {e.msg}")
+        |""".stripMargin
+    )
+    val out = Seq.newBuilder[String]
+    val err = Seq.newBuilder[String]
+    val python = sys.env.get("UDF_PYTHON_PATH").filter(_.nonEmpty).getOrElse("python3.12")
+    val exit = Process(Seq(python, script.toString, input.toString))
+      .!(ProcessLogger(out += _, err += _))
+    // Without this a parser that never ran reads as "nothing failed".
+    require(exit == 0, s"parse_all.py exited $exit: ${err.result().mkString("\n")}")
+    out.result()
+  }
+
+  private def write(dir: Path, name: String, content: String): Path =
+    Files.write(dir.resolve(name), content.getBytes(UTF_8))
+}
diff --git a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/StandaloneInputCheck.scala b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/StandaloneInputCheck.scala
new file mode 100644
index 00000000000..a4de776f07a
--- /dev/null
+++ b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/StandaloneInputCheck.scala
@@ -0,0 +1,88 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.texera.amber.translator.verify
+
+import org.apache.texera.amber.operator.source.SourceOperatorDescriptor
+
+import java.nio.file.Files
+
+/**
+  * Reads every operator's generated code and reports one that writes to the
+  * frame it was handed.
+  *
+  * The translator names a variable per output PORT, not per reader, so two
+  * operators drawn from one upstream are handed the same name. An operator that
+  * drops rows into that name, or assigns a column through it, changes what the
+  * other branch goes on to read: a bar chart that drops its own nulls would
+  * leave a pie chart beside it drawing a table it never received.
+  *
+  * Both spellings of the mistake matter, and one of them does not look like a
+  * mistake. `inplace=True` mutates the frame outright. `in1df = in1df.dropna()`
+  * reads like a local rebinding, but the operator bodies are concatenated at
+  * module scope, so the name it rebinds is the shared one.
+  *
+  * A single-branch workflow never notices either, which is why this reads the
+  * code rather than waiting for a run to disagree. Every fixture the runner
+  * builds has one reader, so a comparison of the two paths agrees while the
+  * frame is being altered underneath a branch that the fixture does not have.
+  *
+  * What reading the code cannot see is a mutation through another name: bind
+  * the frame to something else first and the writes below are invisible here.
+  * No operator does that today, and an operator that starts to would be saying
+  * something a reader has to work out anyway.
+  */
+object StandaloneInputCheck {
+
+  // `.loc` and `.iloc` are here because they are the other way to write the
+  // assignment, and a pattern that reads only the plain subscript would call an
+  // operator clean for choosing the accessor.
+  private val Assignment = """(?m)^\s*in\d+df(\.i?loc|\.i?at)?\s*\[[^\]]*\]\s*=""".r
+  private val InPlace = """in\d+df[^\n]*inplace\s*=\s*True""".r
+  private val Rebind = """(?m)^\s*in\d+df\s*=""".r
+
+  /** Every operator that writes to its input, empty when the suite is clean. */
+  def run(): Seq[String] = {
+    val operators = OperatorBehaviorSpec
+      .discoverStandaloneOperators()
+      .filterNot(classOf[SourceOperatorDescriptor].isAssignableFrom)
+    val dir = Files.createTempDirectory("standalone-input-")
+    dir.toFile.deleteOnExit()
+
+    operators.flatMap { op =>
+      StandaloneEscapingCheck.codeFor(op, dir) match {
+        // An operator whose config cannot be built is reported by the escaping
+        // check, which builds the same ones; repeating it here would say the
+        // same thing twice.
+        case Left(_) => Seq.empty
+        case Right(variants) =>
+          variants.flatMap {
+            case (label, code) =>
+              val how = Seq(
+                Assignment.findFirstIn(code).map(_ => "assigns a column through it"),
+                InPlace.findFirstIn(code).map(_ => "mutates it with inplace=True"),
+                Rebind.findFirstIn(code).map(_ => "rebinds the name it was given")
+              ).flatten
+              if (how.isEmpty) None
+              else Some(s"${op.getSimpleName}/$label: ${how.mkString(", ")}")
+          }
+      }
+    }
+  }
+}