diff --git a/workflow-compiling-service/src/test/resources/python/standalone_worker.py b/workflow-compiling-service/src/test/resources/python/standalone_worker.py
new file mode 100644
index 00000000000..ff4d2aa23f5
--- /dev/null
+++ b/workflow-compiling-service/src/test/resources/python/standalone_worker.py
@@ -0,0 +1,122 @@
+#!/usr/bin/env python3
+#
+# 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.
+"""
+Persistent worker for the Path B (standalone) verify path.
+
+Motivation: forking a fresh interpreter per operator pays the pandas/plotly
+import cost (~260-310 ms) on every spawn, while the operator's actual compute
+on the tiny canonical fixtures is ~4 ms. Imports dominate ~96% of the per-spawn
+cost. This worker imports those heavy libraries ONCE at startup, then executes
+many operators' generated scripts over its lifetime — so the import cost is
+paid once, not once per operator.
+
+It is a drop-in replacement for `python <script.py>`: it runs the exact same
+rendered script `StandaloneRunner` already produces (imports + prologue + body
++ epilogue). The script's own top-of-file `import pandas` becomes a ~0 ms
+`sys.modules` cache hit.
+
+Protocol (line-delimited JSON, both directions):
+
+  startup   worker -> parent:  {"ready": true}
+  request   parent -> worker:  {"scriptPath": "<abs>", "workDir": "<abs>"}\n
+  response  worker -> parent:  {"exit": 0, "stdout": "...", "stderr": "..."}\n
+
+`exit` is 0 on success or 1 if the script raised; on 1, `stderr` carries the
+traceback — mirroring a nonzero subprocess exit so the Scala side's
+StandaloneExecutionException path is unchanged. The worker keeps running after
+a script error (only a hard interpreter crash ends it); parent closes stdin
+(EOF) to shut it down.
+
+Isolation trade-off (accepted, per design discussion): all jobs share one
+interpreter, so module-level state (e.g. pandas display options) can leak
+between operators. Each job is exec'd in a FRESH namespace and chdir'd to its
+own workDir to contain the common cases; this is weaker than the old
+process-per-operator isolation.
+"""
+from __future__ import annotations
+
+import io
+import json
+import os
+import sys
+import traceback
+from contextlib import redirect_stderr, redirect_stdout
+
+# --- Pay the heavy import cost ONCE, here, at startup. ----------------------
+# Pre-importing populates sys.modules, so a script's own `import pandas as pd`
+# or `import plotly...` is a cache hit. It does not hand the script the name:
+# each one runs in a fresh namespace, so an operator that draws with plotly and
+# forgot to declare it still fails here with a NameError, which is the point.
+# numpy is left out for the same reason plotly is only a cache warmer: the
+# script has to ask for what it uses (see StandaloneRunner.renderScript).
+import pandas as pd  # noqa: F401
+import plotly.express as px  # noqa: F401
+import plotly.graph_objects as go  # noqa: F401
+import plotly.io  # noqa: F401
+
+
+def _run_one(script_path: str, work_dir: str) -> "dict[str, object]":
+    """Execute one rendered standalone script and capture its output.
+
+    Runs in a fresh namespace with cwd = work_dir (generated code may use
+    relative paths, e.g. CSVScan's `pd.read_csv("sample.csv")`; absolute paths
+    written by the prologue/epilogue are unaffected). The script's stdout /
+    stderr are redirected into buffers so they never corrupt the protocol
+    channel on real stdout.
+    """
+    out_buf, err_buf = io.StringIO(), io.StringIO()
+    # __name__ = "__main__" so scripts with a `if __name__ == "__main__"` guard
+    # still run their body (the translator does not emit one, but it is free
+    # insurance and matches `python script.py` semantics).
+    namespace = {"__name__": "__main__", "__file__": script_path}
+    try:
+        with open(script_path, "r", encoding="utf-8") as f:
+            source = f.read()
+        os.chdir(work_dir)
+        code = compile(source, script_path, "exec")
+        with redirect_stdout(out_buf), redirect_stderr(err_buf):
+            exec(code, namespace)  # noqa: S102 (running generated verify code by design)
+        return {"exit": 0, "stdout": out_buf.getvalue(), "stderr": err_buf.getvalue()}
+    except BaseException:  # noqa: BLE001 — a script error must NOT kill the worker
+        # Match a nonzero subprocess exit: traceback goes to stderr, exit = 1.
+        err = err_buf.getvalue() + traceback.format_exc()
+        return {"exit": 1, "stdout": out_buf.getvalue(), "stderr": err}
+
+
+def main() -> None:
+    # Signal readiness only after the heavy imports above have completed, so the
+    # parent can warm a pool and attribute startup cost deterministically.
+    sys.stdout.write(json.dumps({"ready": True}) + "\n")
+    sys.stdout.flush()
+
+    for line in sys.stdin:
+        line = line.strip()
+        if not line:
+            continue
+        try:
+            req = json.loads(line)
+            result = _run_one(req["scriptPath"], req["workDir"])
+        except Exception:  # malformed request — report, keep serving
+            result = {"exit": 1, "stdout": "", "stderr": traceback.format_exc()}
+        sys.stdout.write(json.dumps(result) + "\n")
+        sys.stdout.flush()
+
+
+if __name__ == "__main__":
+    main()
diff --git a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/HarnessSpec.scala b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/HarnessSpec.scala
new file mode 100644
index 00000000000..685b6f056db
--- /dev/null
+++ b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/HarnessSpec.scala
@@ -0,0 +1,115 @@
+/*
+ * 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.{Attribute, AttributeType, Schema, Tuple}
+import org.apache.texera.amber.core.workflow.PortIdentity
+import org.apache.texera.amber.operator.distinct.DistinctOpDesc
+import org.scalatest.Tag
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+
+import java.nio.file.{Files, Path}
+
+/** The two ways of running one operator, and the file format they meet in.
+  *
+  * `Distinct` is the operator under test throughout, because what is being
+  * tested is the harness rather than the operator: it takes one input, needs no
+  * configuration, and its answer is short enough to state in full.
+  */
+class HarnessSpec extends AnyFlatSpec with Matchers {
+
+  /** Only the standalone run needs an interpreter, so only it is held back from
+    * the job that provisions none. The other two are JVM-side and run there.
+    */
+  private val NeedsPython =
+    Tag("org.apache.texera.amber.translator.verify.tags.IntegrationTest")
+
+  private val schema = new Schema(
+    new Attribute("id", AttributeType.INTEGER),
+    new Attribute("name", AttributeType.STRING)
+  )
+
+  private def tuple(id: Int, name: String): Tuple = {
+    val b = Tuple.builder(schema)
+    b.add(schema.getAttribute("id"), Int.box(id))
+    b.add(schema.getAttribute("name"), name)
+    b.build()
+  }
+
+  /** Four rows, the last a repeat of the second. */
+  private val rows = Seq(tuple(1, "a"), tuple(2, "b"), tuple(3, "c"), tuple(2, "b"))
+
+  private def withInput(test: (Path, Path) => Unit): Unit = {
+    val dir = Files.createTempDirectory("harness-spec-")
+    val input = dir.resolve("input_port_0.jsonl")
+    TupleIO.writeTuples(input, rows.iterator, schema)
+    test(dir, input)
+  }
+
+  "TupleIO" should "read back the rows and the schema it wrote" in {
+    withInput { (_, input) =>
+      // The schema travels in a sidecar rather than in the JSONL, which carries
+      // values alone and so cannot say a column is INTEGER rather than a number.
+      TupleIO.readSchemaSidecar(input) shouldBe schema
+      val read = TupleIO.readTuples(input, schema).toSeq
+      read should have length 4
+      read.map(_.getField[Integer]("id").intValue) shouldBe Seq(1, 2, 3, 2)
+    }
+  }
+
+  "OpExecHarness" should "run an operator and write one file per output port" in {
+    withInput { (dir, input) =>
+      val out = dir.resolve("actual")
+      val result =
+        OpExecHarness.execute(new DistinctOpDesc, Map(PortIdentity(0) -> input), out)
+
+      result.outputs should have size 1
+      val produced = result.outputs(PortIdentity(0))
+      Files.exists(produced) shouldBe true
+
+      val written = TupleIO.readTuples(produced, result.outputSchemas(PortIdentity(0))).toSeq
+      written.map(_.getField[Integer]("id").intValue) shouldBe Seq(1, 2, 3)
+    }
+  }
+
+  "StandaloneRunner" should "run the generated script and reach the same answer" taggedAs NeedsPython in {
+    withInput { (dir, input) =>
+      val work = dir.resolve("standalone")
+      Files.createDirectories(work)
+      val result = StandaloneRunner.run(
+        opDesc = new DistinctOpDesc,
+        inputs = Map(1 -> input),
+        outputPortCount = 1,
+        workDir = work
+      )
+
+      // The script is kept where it ran, so a failing operator can be opened as
+      // generated rather than described second-hand.
+      Files.exists(work.resolve("script.py")) shouldBe true
+
+      val produced = result.outputs(1)
+      val lines = Files.readAllLines(produced)
+      lines should have size 3
+      lines.get(0) should include("\"id\":1")
+      lines.get(2) should include("\"id\":3")
+    }
+  }
+}
diff --git a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/StandaloneRunner.scala b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/StandaloneRunner.scala
new file mode 100644
index 00000000000..e39990c0768
--- /dev/null
+++ b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/StandaloneRunner.scala
@@ -0,0 +1,364 @@
+/*
+ * 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.typesafe.scalalogging.LazyLogging
+import org.apache.texera.amber.core.tuple.AttributeType
+import org.apache.texera.amber.operator.{LogicalOp, StandaloneCodeGenerator}
+import org.apache.texera.amber.util.python.PythonWorkerPool
+
+import java.nio.charset.StandardCharsets
+import java.nio.file.{Files, Path}
+import scala.collection.mutable.ArrayBuffer
+import scala.sys.process._
+
+/**
+  * Executes the Python code an OpDesc's [[StandaloneCodeGenerator]] emits and
+  * captures its DataFrame outputs as JSONL files (compatible with
+  * [[TupleIO]]'s sidecar-schema format on the comparison side).
+  *
+  * The operator's code is wrapped in a prologue that reads each input file into
+  * an `inNdf` and an epilogue that writes each `outNdf` back out, with the
+  * generated body verbatim between them.
+  *
+  * Port indexing matches the placeholder convention used by the translator:
+  * `inNdf`/`outNdf` is 1-based and corresponds to the operator's N-th external
+  * input/output port in declaration order. The harness key (a 1-based Int) is
+  * what the placeholder uses; the caller is responsible for ordering inputs
+  * the same way the operator's `generateStandaloneCode()` expects.
+  *
+  * The subprocess inherits the caller's environment so the Python interpreter
+  * picks up whatever pandas/plotly the test fixture installed.
+  */
+object StandaloneRunner extends LazyLogging {
+
+  /** The value both paths seed numpy's global RNG with. Any fixed number does;
+    * what matters is that the two agree, so it is declared once here and
+    * referenced by name from py_op_driver's comment.
+    */
+  private[verify] val VerifySeed: Int = 20260811
+
+  /**
+    * @param outputs paths to the per-port output JSONL files. Empty map iff
+    *                the operator's `producesDataFrame()` returned false
+    *                (visualizations, etc.) — caller handles those separately.
+    * @param stdout  raw subprocess stdout (useful for failure diagnostics)
+    * @param stderr  raw subprocess stderr
+    */
+  final case class Result(outputs: Map[Int, Path], stdout: String, stderr: String)
+
+  /**
+    * Generate, write, and execute the standalone Python script for `opDesc`.
+    *
+    * @param opDesc must mix in [[StandaloneCodeGenerator]]; otherwise we throw
+    *               since there's nothing to test.
+    * @param inputs map from 1-based port index → JSONL fixture path. The
+    *               script reads each into `inNdf`.
+    * @param outputPortCount how many `outNdf` variables the operator declares.
+    *                Caller derives this from the OpDesc's output ports.
+    * @param workDir directory used for the generated `script.py` and output
+    *                JSONL files. Created if missing.
+    * @param pythonExe path to the Python 3.12 interpreter. Defaults to
+    *                  the env var `UDF_PYTHON_PATH`, then `python3.12`, then
+    *                  `python3`. The same fallback chain used by the rest of
+    *                  the Texera test suite for Python-backed operators.
+    */
+  def run(
+      opDesc: LogicalOp,
+      inputs: Map[Int, Path],
+      outputPortCount: Int,
+      workDir: Path,
+      pythonExe: String = resolvePython()
+  ): Result = {
+    val gen = opDesc match {
+      case g: StandaloneCodeGenerator => g
+      case other =>
+        throw new IllegalArgumentException(
+          s"OpDesc ${other.getClass.getSimpleName} does not implement " +
+            s"StandaloneCodeGenerator; nothing to verify"
+        )
+    }
+
+    Files.createDirectories(workDir)
+    val scriptPath = workDir.resolve("script.py")
+    val outputPaths: Map[Int, Path] =
+      if (gen.producesDataFrame())
+        (1 to outputPortCount).map(i => i -> workDir.resolve(s"output_port_${i - 1}.jsonl")).toMap
+      else Map.empty
+
+    val source =
+      renderScript(
+        gen.generateStandaloneCode(),
+        inputs,
+        outputPaths,
+        gen.standaloneHelpers(),
+        gen.standaloneImports()
+      )
+    Files.write(scriptPath, source.getBytes(StandardCharsets.UTF_8))
+
+    val (exit, stdout, stderr) = execute(scriptPath, workDir, pythonExe)
+    if (exit != 0) {
+      throw new StandaloneExecutionException(exit, scriptPath, source, stdout, stderr)
+    }
+    Result(outputPaths, stdout, stderr)
+  }
+
+  private val WorkerResourcePath = "/python/standalone_worker.py"
+
+  // Run the rendered script and return (exitCode, stdout, stderr). Prefers a
+  // pooled persistent worker (imports pandas/plotly once, ~18x faster per op —
+  // see PythonWorkerPool); a rare hard worker crash falls back to a one-shot
+  // subprocess so behavior is never worse than the original path. Both paths
+  // run with cwd = workDir and read results from files, so they are
+  // interchangeable — the executed script is byte-identical.
+  private def execute(scriptPath: Path, workDir: Path, pythonExe: String): (Int, String, String) = {
+    if (PythonWorkerPool.enabled) {
+      try {
+        val req = org.apache.texera.amber.util.JSONUtils.objectMapper.createObjectNode()
+        req.put("scriptPath", scriptPath.toString)
+        req.put("workDir", workDir.toString)
+        val o = PythonWorkerPool.run(WorkerResourcePath, Seq.empty, pythonExe, req)
+        return (o.exit, o.stdout, o.stderr)
+      } catch {
+        case e: PythonWorkerPool.WorkerDiedException =>
+          logger.warn(
+            s"Standalone worker unavailable; falling back to one-shot subprocess " +
+              s"for $scriptPath: ${e.getMessage}"
+          )
+      }
+    }
+    runSubprocess(scriptPath, workDir, pythonExe)
+  }
+
+  // Original one-process-per-operator path. Retained as the fallback and as the
+  // behavior selected by TEXERA_TEST_PYTHON_WORKER=0.
+  private def runSubprocess(
+      scriptPath: Path,
+      workDir: Path,
+      pythonExe: String
+  ): (Int, String, String) = {
+    // Capture stdout/stderr separately. ProcessLogger's append is called from
+    // the subprocess's I/O thread, so we collect into ArrayBuffer (thread-safe
+    // append is fine for this serial use) and join at the end.
+    val outBuf = ArrayBuffer.empty[String]
+    val errBuf = ArrayBuffer.empty[String]
+    val logger = ProcessLogger(line => outBuf += line, line => errBuf += line)
+    // cwd = workDir so generated code using *relative* paths (e.g. CSVScan's
+    // basename-stripped `pd.read_csv("sample.csv")`) resolves against workDir.
+    // Absolute paths written by the prologue/epilogue are unaffected.
+    val exit = Process(Seq(pythonExe, scriptPath.toString), Some(workDir.toFile)).!(logger)
+    (exit, outBuf.mkString("\n"), errBuf.mkString("\n"))
+  }
+
+  // Builds the full Python source: imports + prologue + verbatim operator body
+  // + epilogue. We intentionally do NOT substitute the inNdf/outNdf placeholders
+  // — the body keeps them so the var-bindings the prologue/epilogue introduce
+  // (also named inNdf/outNdf) reference the same names.
+  private def renderScript(
+      body: String,
+      inputs: Map[Int, Path],
+      outputs: Map[Int, Path],
+      helpers: Seq[String],
+      imports: Seq[String]
+  ): String = {
+    val sb = new StringBuilder
+
+    sb.append("# Auto-generated by StandaloneRunner. Do not commit.\n")
+    sb.append("import json\n")
+    sb.append("import sys\n")
+    sb.append("import base64\n")
+    sb.append("import pickle\n")
+    // NOTE: nothing beyond pandas is injected here. The production translator
+    // (WorkflowToPythonTranslator) emits pandas for every script and then only
+    // what the operators in the plan ask for, so an operator whose standalone
+    // code needs numpy, or plotly, must say so. Injecting either here would mask
+    // that class of bug: the script would run in verify and fail on export.
+    sb.append("import pandas as pd\n")
+    imports.foreach(line => sb.append(line).append("\n"))
+    // Same seed as py_op_driver's run_config, for the reason given there. Bound
+    // under a private name and deleted so the note above still holds: a script
+    // that wants numpy has to import it, and this does not hand it one.
+    sb.append(s"import numpy as _texera_np; _texera_np.random.seed($VerifySeed); del _texera_np\n")
+    sb.append("\n")
+
+    // Object columns holding non-primitive values (e.g. a trained sklearn model
+    // in a BINARY output column) can't go through to_json. Pickle+base64 them so
+    // the JSONL matches py_op_driver's BINARY write path exactly. Primitives
+    // (str/int/float/bool/None) pass through unchanged, so ordinary DataFrame
+    // outputs are unaffected.
+    sb.append("def _texera_encode_obj_cols(df):\n")
+    sb.append("    for _c in df.columns:\n")
+    sb.append("        if df[_c].dtype == object:\n")
+    sb.append(
+      "            df[_c] = df[_c].map(lambda _v: base64.b64encode(pickle.dumps(_v)).decode('ascii') " +
+        "if not isinstance(_v, (str, int, float, bool, type(None))) else _v)\n"
+    )
+    sb.append("    return df\n")
+    sb.append("\n")
+
+    // TIMESTAMP columns are handed to the operator as datetime64 (see the
+    // prologue below) to match the schema-typed runtime path, but the runtime
+    // path serializes a TIMESTAMP back out with java.sql.Timestamp.toString —
+    // "yyyy-mm-dd hh:mm:ss.f", trailing zeros trimmed to at least one digit —
+    // whereas pandas' to_json would emit epoch millis. Convert datetime columns
+    // back to that exact form before writing so both paths' JSONL agree.
+    sb.append("def _texera_ts_str(_v):\n")
+    sb.append("    if pd.isna(_v):\n")
+    sb.append("        return None\n")
+    sb.append("    _s = _v.strftime('%Y-%m-%d %H:%M:%S.%f').rstrip('0')\n")
+    sb.append("    return _s + '0' if _s.endswith('.') else _s\n")
+    sb.append("\n")
+    sb.append("def _texera_encode_ts_cols(df):\n")
+    sb.append("    for _c in df.columns:\n")
+    sb.append("        if pd.api.types.is_datetime64_any_dtype(df[_c]):\n")
+    sb.append("            df[_c] = df[_c].map(_texera_ts_str)\n")
+    sb.append("    return df\n")
+    sb.append("\n")
+
+    // Prologue: load each external input into in{N}df. Note: pd.read_json with
+    // lines=True correctly handles empty files (returns empty DataFrame).
+    // convert_dates=False: pd.read_json otherwise auto-coerces ISO-ish strings
+    // and columns named like dates ("date", "*_at", …) to datetime64, which the
+    // schema-typed runtime path (STRING) does not do — that divergence would
+    // make a plain date string column serialize as "...T00:00:00" on only one
+    // side. Operators that genuinely need datetimes convert explicitly, so both
+    // paths stay in sync.
+    // precise_float=True: pd.read_json's default (ujson) fast double parser is
+    // lossy in the last few ULPs, so a DOUBLE column would load slightly
+    // different values than the schema-typed runtime path (which parses doubles
+    // exactly). Operators that stringify raw cell values (e.g. Radar hover text)
+    // then diverge; precise_float=True keeps both paths bit-identical.
+    // The blanket convert_dates=False also leaves genuine TIMESTAMP columns as
+    // strings, which the runtime path delivers as datetime64 — a divergence for
+    // any operator that renders or computes on them. The fixture's schema
+    // sidecar says which columns those are, so cast exactly those back.
+    inputs.toSeq.sortBy(_._1).foreach {
+      case (n, path) =>
+        sb.append(
+          s"in${n}df = pd.read_json(${py(path.toString)}, lines=True, convert_dates=False, precise_float=True)\n"
+        )
+        timestampColumns(path).foreach { col =>
+          sb.append(s"if ${py(col)} in in${n}df.columns:\n")
+          sb.append(s"    in${n}df[${py(col)}] = pd.to_datetime(in${n}df[${py(col)}])\n")
+        }
+        doubleColumns(path).foreach { col =>
+          sb.append(s"if ${py(col)} in in${n}df.columns:\n")
+          sb.append(s"    in${n}df[${py(col)}] = in${n}df[${py(col)}].astype('float64')\n")
+        }
+    }
+    // The variadic placeholder, bound here for the same reason the numbered ones
+    // are: this script leaves the body's placeholders alone and defines names to
+    // match them, so an operator reading a variadic port finds its list here the
+    // way the translator would have written one out.
+    if (inputs.nonEmpty) {
+      sb.append(
+        inputs.keys.toSeq.sorted.map(n => s"in${n}df").mkString("inAlldf = [", ", ", "]\n")
+      )
+    }
+    sb.append("\n")
+
+    // Body verbatim — placeholders left in place.
+    // Emitted ahead of the body the way the translator does, so an operator that
+    // declares a helper is exercised here exactly as it runs in a real script.
+    helpers.foreach { helper =>
+      sb.append(helper)
+      if (!helper.endsWith("\n")) sb.append('\n')
+      sb.append('\n')
+    }
+
+    sb.append("# ── operator body ──\n")
+    sb.append(body)
+    if (!body.endsWith("\n")) sb.append('\n')
+    sb.append("\n")
+
+    // Epilogue: dump each out{N}df to JSONL. When producesDataFrame() is false
+    // (visualization ops), `outputs` is empty and this block is a no-op — the
+    // caller is expected to verify viz outputs by other means.
+    outputs.toSeq.sortBy(_._1).foreach {
+      case (n, path) =>
+        sb.append(
+          s"_texera_encode_obj_cols(_texera_encode_ts_cols(out${n}df))" +
+            s".to_json(${py(path.toString)}, orient='records', lines=True)\n"
+        )
+    }
+
+    sb.toString
+  }
+
+  // TIMESTAMP-typed column names from a fixture's `.jsonl.schema.json` sidecar.
+  // A missing or unreadable sidecar means no casts — the prologue then behaves
+  // exactly as before.
+  private def timestampColumns(input: Path): Seq[String] =
+    columnsOfType(input, AttributeType.TIMESTAMP)
+
+  // DOUBLE-typed column names. pd.read_json narrows a float column whose values
+  // are all integral to int64, while the runtime path keeps the schema's DOUBLE,
+  // so a column like 7.0 stringifies as "7" on one side and "7.0" on the other —
+  // invisible to numeric comparison, visible the moment an operator uses the
+  // column as a label (a trace name, a legend entry, hover text).
+  private def doubleColumns(input: Path): Seq[String] =
+    columnsOfType(input, AttributeType.DOUBLE)
+
+  private def columnsOfType(input: Path, attributeType: AttributeType): Seq[String] =
+    scala.util
+      .Try(TupleIO.readSchemaSidecar(input))
+      .toOption
+      .toSeq
+      .flatMap(
+        _.getAttributes.filter(_.getType == attributeType).map(_.getName)
+      )
+
+  // Python string literal, single-quoted with backslashes escaped. We
+  // deliberately don't use repr() in Scala (no such thing) — JSON.toString
+  // would also work but introduces double-quote escaping when the path has
+  // spaces.
+  private def py(s: String): String =
+    "'" + s.replace("\\", "\\\\").replace("'", "\\'") + "'"
+
+  // Resolution chain mirrors the rest of the Texera test infra: env var first
+  // (set by CI / the shared-venv setup), then conventional names.
+  private def resolvePython(): String = {
+    val fromEnv = sys.env.get("UDF_PYTHON_PATH").filter(_.nonEmpty)
+    fromEnv.getOrElse {
+      // We don't try to probe `which` here — if neither env var nor a literal
+      // `python3.12` is on PATH, the subprocess invocation will fail and the
+      // error path below surfaces it.
+      "python3.12"
+    }
+  }
+}
+
+final class StandaloneExecutionException(
+    val exitCode: Int,
+    val scriptPath: Path,
+    val source: String,
+    val stdout: String,
+    val stderr: String
+) extends RuntimeException(
+      // The script path goes first in the message so a failing CI log makes it
+      // immediately obvious which file to open. stderr ends the message because
+      // the Python traceback (if any) is the most actionable signal.
+      s"""Standalone Python script exited with code $exitCode.
+         |Script: $scriptPath
+         |--- stdout ---
+         |$stdout
+         |--- stderr ---
+         |$stderr""".stripMargin
+    )