From 6f84341efbf1d6d068c3991e7487fd5acaaaa0a4 Mon Sep 17 00:00:00 2001
From: Zach Maddox <zmaddox@salesforce.com>
Date: Wed, 12 Aug 2026 19:27:10 -0400
Subject: [PATCH 01/10] jupyter bind loopback, use token

---
 README.md                                     |  4 +-
 .../templates/script/jupyterlab.sh            | 22 ++++--
 tests/test_jupyterlab_script.py               | 67 +++++++++++++++++++
 3 files changed, 87 insertions(+), 6 deletions(-)
 create mode 100644 tests/test_jupyterlab_script.py

diff --git a/README.md b/README.md
index 3cf73b9..3452d04 100644
--- a/README.md
+++ b/README.md
@@ -472,7 +472,7 @@ exploration.  Instead of running an entire script, one can run one code cell at
 
 You can read more about Jupyter Notebooks here: https://jupyter.org/
 
-1. Within the root project of your package folder, run `./jupyterlab.sh start`
+1. Within the root project of your package folder, run `./jupyterlab.sh start`. This prints an access token and opens an already-authenticated JupyterLab session in your browser. If the browser doesn't open automatically, copy the printed `http://localhost:8888/?token=...` URL into your browser.
 1. Double-click on "account.ipynb" file, which provides a starting point for a notebook
 1. Use shift+enter to execute each cell within the notebook.  Add/edit/delete cells of code as needed for your data exploration.
 1. Don't forget to run `./jupyterlab.sh stop` to stop the docker container
@@ -563,4 +563,4 @@ If you're using OAuth Tokens authentication, the initial configure will retrieve
 ## Other docs
 
 - [Troubleshooting](./docs/troubleshooting.md)
-- [For Contributors](./FOR_CONTRIBUTORS.md)
+- [Contributing](./CONTRIBUTING.md)
diff --git a/src/datacustomcode/templates/script/jupyterlab.sh b/src/datacustomcode/templates/script/jupyterlab.sh
index e8445fc..55829d0 100755
--- a/src/datacustomcode/templates/script/jupyterlab.sh
+++ b/src/datacustomcode/templates/script/jupyterlab.sh
@@ -45,13 +45,24 @@ check_docker() {
     echo "Docker daemon is running"
 }
 
+# Function to check if openssl is installed
+check_openssl() {
+    if ! command -v openssl &> /dev/null; then
+        echo "openssl is not installed. It is required to generate a secure JupyterLab access token."
+        exit 1
+    fi
+}
+
 # Function to start Jupyter server
 start_jupyter() {
     echo "Building the docker image"
     docker build -t datacloud-customcode .
 
+    local TOKEN
+    TOKEN=$(openssl rand -hex 32)
+
     echo "Running the docker container"
-    docker run -d --rm -p 8888:8888 \
+    docker run -d --rm -p 127.0.0.1:8888:8888 \
         -v $(pwd):/workspace \
         --name jupyter-server \
         datacloud-customcode jupyter lab \
@@ -59,12 +70,14 @@ start_jupyter() {
         --port=8888 \
         --no-browser \
         --allow-root \
-        --NotebookApp.token='' \
-        --NotebookApp.password='' \
+        --NotebookApp.token="$TOKEN" \
         --notebook-dir=/workspace
 
     sleep 3  # Wait for server to start
-    open_browser "http://localhost:8888"
+    local URL
+    URL="http://localhost:8888/?token=$TOKEN"
+    echo "Opening $URL"
+    open_browser $URL
 }
 
 # Function to stop Jupyter server
@@ -82,6 +95,7 @@ stop_jupyter() {
 case "$1" in
     "start")
         check_docker
+        check_openssl
         start_jupyter
         ;;
     "stop")
diff --git a/tests/test_jupyterlab_script.py b/tests/test_jupyterlab_script.py
new file mode 100644
index 0000000..082f583
--- /dev/null
+++ b/tests/test_jupyterlab_script.py
@@ -0,0 +1,67 @@
+from __future__ import annotations
+
+import os
+import subprocess
+
+from datacustomcode.template import script_template_dir
+
+JUPYTERLAB_SH = os.path.join(script_template_dir, "jupyterlab.sh")
+
+# These tests don't actually run the jupyter script. They simply verify
+# certain specific configurations of the script for things like syntax
+# and security correctness.
+#
+# These were added when fixing a bug that could have allowed for RCE
+# over the local network on the user's device due to previous insufficient
+# network config. While not perfect, they do offer a bit of assurance that
+# the script is configured correctly.
+
+
+class TestJupyterlabScript:
+    def _read(self) -> str:
+        with open(JUPYTERLAB_SH) as f:
+            return f.read()
+
+    def test_jupyterlab_sh_syntax_is_valid(self):
+        """`bash -n` should accept the script without syntax errors."""
+        result = subprocess.run(
+            ["bash", "-n", JUPYTERLAB_SH],
+            capture_output=True,
+            text=True,
+            check=False,
+        )
+        assert result.returncode == 0, result.stderr
+
+    def test_start_jupyter_binds_loopback_host_port(self):
+        content = self._read()
+        assert "-p 127.0.0.1:8888:8888" in content
+        assert "-p 8888:8888" not in content
+
+    def test_start_jupyter_binds_container_to_all_interfaces(self):
+        content = self._read()
+        assert "--ip=0.0.0.0" in content
+        assert "--ip=127.0.0.1" not in content
+
+    def test_start_jupyter_generates_token_not_empty_auth(self):
+        content = self._read()
+        assert "--NotebookApp.token=''" not in content
+        assert "--NotebookApp.password=''" not in content
+        assert "openssl rand -hex 32" in content
+
+    def test_start_jupyter_uses_dynamic_token_variable(self):
+        content = self._read()
+        assert "local TOKEN" in content
+        assert "TOKEN=$(openssl rand -hex 32)" in content
+        assert '--NotebookApp.token="$TOKEN"' in content
+
+    def test_open_browser_url_includes_token_param(self):
+        content = self._read()
+        assert 'URL="http://localhost:8888/?token=$TOKEN"' in content
+        assert "open_browser $URL" in content
+
+    def test_token_never_written_to_file(self):
+        content = self._read()
+        assert "credentials.ini" not in content
+        for line in content.splitlines():
+            if "TOKEN" in line:
+                assert ">" not in line, f"Line writes TOKEN to a file: {line!r}"

From c017cc8f9d5e1dff1dafb029f163e33360325074 Mon Sep 17 00:00:00 2001
From: Steffan Byrne <sbyrne@salesforce.com>
Date: Mon, 17 Aug 2026 14:19:47 -0400
Subject: [PATCH 02/10] Added auto_write* methods to writers and clients.

These new methods will be used with streaming transforms when running in
specific modes where the system must control the write mode.
---
 src/datacustomcode/client.py                  | 36 ++++++++++++++++
 src/datacustomcode/io/writer/base.py          | 12 ++++++
 src/datacustomcode/io/writer/csv.py           |  6 +++
 src/datacustomcode/io/writer/print.py         |  7 +++
 .../examples/streaming_deltas/entrypoint.py   | 40 +++++++++++------
 tests/spark/test_session_provider.py          |  6 +++
 tests/test_client.py                          | 43 +++++++++++++++++++
 7 files changed, 138 insertions(+), 12 deletions(-)

diff --git a/src/datacustomcode/client.py b/src/datacustomcode/client.py
index 56c0588..7836c70 100644
--- a/src/datacustomcode/client.py
+++ b/src/datacustomcode/client.py
@@ -15,6 +15,7 @@
 from __future__ import annotations
 
 from enum import Enum
+import os
 from typing import (
     TYPE_CHECKING,
     Any,
@@ -598,3 +599,38 @@ def write_dlo_deltas(
         """
         self._validate_data_layer_history_does_not_contain(DataCloudObjectType.DMO)
         return self._writer.write_dlo_deltas(name, dataframe, **kwargs)  # type: ignore[no-any-return]
+
+    def auto_write_to_dlo(self, name: str, dataframe: PySparkDataFrame) -> None:
+        """Write a PySpark DataFrame to a DLO in Data Cloud automatically picking
+        the WriteMode.
+        For use with streaming transforms when running in rebuild or initial sync mode.
+        Args:
+            name: The name of the DLO to write to.
+            dataframe: The PySpark DataFrame to write.
+        """
+        self._validate_data_layer_history_does_not_contain(DataCloudObjectType.DMO)
+        return self._writer.auto_write_to_dlo(name, dataframe)
+
+    def auto_write_to_dmo(self, name: str, dataframe: PySparkDataFrame) -> None:
+        """Write a PySpark DataFrame to a DMO in Data Cloud automatically picking
+        the WriteMode.
+        For use with streaming transforms when running in rebuild or initial sync mode.
+        Args:
+            name: The name of the DMO to write to.
+            dataframe: The PySpark DataFrame to write.
+        """
+        self._validate_data_layer_history_does_not_contain(DataCloudObjectType.DLO)
+        return self._writer.auto_write_to_dmo(name, dataframe)
+
+
+class RunMode(Enum):
+    BATCH = "BATCH"
+    INITIAL_SYNC = "INITIAL_SYNC"
+    REBUILD = "REBUILD"
+    DELTA_SYNC = "DELTA_SYNC"
+
+
+def get_run_mode() -> RunMode:
+    """Read and validate the BYOC_RUN_MODE env var; default to BATCH when unset."""
+    run_mode = os.getenv("BYOC_RUN_MODE", "BATCH").upper()
+    return RunMode(run_mode)
diff --git a/src/datacustomcode/io/writer/base.py b/src/datacustomcode/io/writer/base.py
index 47a7bd2..a37e226 100644
--- a/src/datacustomcode/io/writer/base.py
+++ b/src/datacustomcode/io/writer/base.py
@@ -59,6 +59,18 @@ def write_to_dmo(
         self, name: str, dataframe: PySparkDataFrame, write_mode: WriteMode
     ) -> None: ...
 
+    @abstractmethod
+    def auto_write_to_dlo(self, name: str, dataframe: PySparkDataFrame) -> None:
+        """Write to a DLO automatically picking the write mode.
+        For use with streaming transforms when running in rebuild or initial sync mode.
+        """
+
+    @abstractmethod
+    def auto_write_to_dmo(self, name: str, dataframe: PySparkDataFrame) -> None:
+        """Write to a DMO automatically picking the write mode.
+        For use with streaming transforms when running in rebuild or initial sync mode.
+        """
+
     def write_dlo_deltas(
         self, name: str, dataframe: PySparkDataFrame
     ) -> StreamingQuery:
diff --git a/src/datacustomcode/io/writer/csv.py b/src/datacustomcode/io/writer/csv.py
index 3d037d9..7505026 100644
--- a/src/datacustomcode/io/writer/csv.py
+++ b/src/datacustomcode/io/writer/csv.py
@@ -36,6 +36,9 @@ def write_to_dlo(
             name = f"{name}{SUFFIX}"
         dataframe.write.csv(name, mode=write_mode)
 
+    def auto_write_to_dlo(self, name: str, dataframe: PySparkDataFrame) -> None:
+        self.write_to_dlo(name, dataframe, WriteMode.OVERWRITE)
+
     def write_to_dmo(
         self, name: str, dataframe: PySparkDataFrame, write_mode: WriteMode
     ) -> None:
@@ -43,3 +46,6 @@ def write_to_dmo(
         if not name.lower().endswith(SUFFIX):
             name = f"{name}{SUFFIX}"
         dataframe.write.csv(name, mode=write_mode)
+
+    def auto_write_to_dmo(self, name: str, dataframe: PySparkDataFrame) -> None:
+        self.write_to_dmo(name, dataframe, WriteMode.OVERWRITE)
diff --git a/src/datacustomcode/io/writer/print.py b/src/datacustomcode/io/writer/print.py
index c4d2a75..19ef117 100644
--- a/src/datacustomcode/io/writer/print.py
+++ b/src/datacustomcode/io/writer/print.py
@@ -122,6 +122,10 @@ def write_to_dlo(
 
         dataframe.show()
 
+    def auto_write_to_dlo(self, name: str, dataframe: PySparkDataFrame) -> None:
+        self.validate_dataframe_columns_against_dlo(dataframe, name)
+        dataframe.show()
+
     def write_to_dmo(
         self, name: str, dataframe: PySparkDataFrame, write_mode: WriteMode
     ) -> None:
@@ -130,3 +134,6 @@ def write_to_dmo(
         # so just show the dataframe.
 
         dataframe.show()
+
+    def auto_write_to_dmo(self, name: str, dataframe: PySparkDataFrame) -> None:
+        dataframe.show()
diff --git a/src/datacustomcode/templates/script/examples/streaming_deltas/entrypoint.py b/src/datacustomcode/templates/script/examples/streaming_deltas/entrypoint.py
index 97dea40..fc1b6cc 100644
--- a/src/datacustomcode/templates/script/examples/streaming_deltas/entrypoint.py
+++ b/src/datacustomcode/templates/script/examples/streaming_deltas/entrypoint.py
@@ -22,27 +22,43 @@
 ``NotImplementedError`` for the delta methods.
 """
 
+from pyspark.sql import DataFrame
 from pyspark.sql.functions import col, upper
 
-from datacustomcode.client import StreamingClient
+from datacustomcode.client import (
+    Client,
+    RunMode,
+    StreamingClient,
+    get_run_mode,
+)
 
 
 def main():
-    client = StreamingClient()
+    source_dlo = "Account_std__dll"
+    target_dlo = "Account_std_copy__dll"
+    if get_run_mode() == RunMode.DELTA_SYNC:
+        client = StreamingClient()
+        # Streaming DataFrame over the source DLO's change feed.
+        dataframe = client.read_dlo_deltas()
+        # Ordinary PySpark transform.
+        transformed = transform(dataframe)
 
-    # Streaming DataFrame over the source DLO's change feed.
-    deltas = client.read_dlo_deltas()
+        # Start the streaming write. write_dlo_deltas returns the StreamingQuery;
+        # the trigger and checkpoint location are provided by the runtime.
+        query = client.write_dlo_deltas(target_dlo, transformed)
 
-    # Ordinary PySpark transform.
-    transformed = deltas.withColumn("description__c", upper(col("description__c")))
+        # Drive the query's lifecycle. In the streaming runtime this blocks until
+        # the job is stopped by the platform.
+        query.awaitTermination()
+    else:
+        client = Client()
+        dataframe = client.read_dlo(source_dlo)
+        transformed = transform(dataframe)
+        client.auto_write_to_dlo(target_dlo, transformed)
 
-    # Start the streaming write. write_dlo_deltas returns the StreamingQuery;
-    # the trigger and checkpoint location are provided by the runtime.
-    query = client.write_dlo_deltas("Account_std_copy__dll", transformed)
 
-    # Drive the query's lifecycle. In the streaming runtime this blocks until
-    # the job is stopped by the platform.
-    query.awaitTermination()
+def transform(dataframe: DataFrame) -> DataFrame:
+    return dataframe.withColumn("description__c", upper(col("description__c")))
 
 
 if __name__ == "__main__":
diff --git a/tests/spark/test_session_provider.py b/tests/spark/test_session_provider.py
index 71f0e70..b0c7d90 100644
--- a/tests/spark/test_session_provider.py
+++ b/tests/spark/test_session_provider.py
@@ -52,11 +52,17 @@ def write_to_dlo(
     ) -> None:  # type: ignore[override]
         raise NotImplementedError
 
+    def auto_write_to_dlo(self, name: str, dataframe: PySparkDataFrame) -> None:
+        raise NotImplementedError
+
     def write_to_dmo(
         self, name: str, dataframe: PySparkDataFrame, write_mode: WriteMode
     ) -> None:  # type: ignore[override]
         raise NotImplementedError
 
+    def auto_write_to_dmo(self, name: str, dataframe: PySparkDataFrame) -> None:
+        raise NotImplementedError
+
 
 class FakeProvider(BaseSparkSessionProvider):
     CONFIG_NAME = "FakeProvider"
diff --git a/tests/test_client.py b/tests/test_client.py
index c40a995..042f352 100644
--- a/tests/test_client.py
+++ b/tests/test_client.py
@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import os
 from unittest.mock import MagicMock, patch
 
 from pyspark.sql import DataFrame, SparkSession
@@ -9,9 +10,11 @@
     Client,
     DataCloudAccessLayerException,
     DataCloudObjectType,
+    RunMode,
     StreamingClient,
     _BaseClient,
     einstein_predict_col,
+    get_run_mode,
     llm_gateway_generate_text_col,
 )
 from datacustomcode.config import (
@@ -48,11 +51,17 @@ def write_to_dlo(
     ) -> None:
         pass
 
+    def auto_write_to_dlo(self, name: str, dataframe: DataFrame) -> None:
+        pass
+
     def write_to_dmo(
         self, name: str, dataframe: DataFrame, write_mode: WriteMode, **kwargs
     ) -> None:
         pass
 
+    def auto_write_to_dmo(self, name: str, dataframe: DataFrame) -> None:
+        pass
+
 
 @pytest.fixture
 def mock_spark():
@@ -265,6 +274,16 @@ def test_read_pattern_flow(self, reset_client, mock_spark):
 
         assert "source_dmo" in client._data_layer_history[DataCloudObjectType.DMO]
 
+    @patch.dict(os.environ, {}, clear=True)
+    def test_get_run_mode_default_batch(self, reset_client, mock_spark):
+
+        assert get_run_mode() == RunMode.BATCH
+
+    @patch.dict(os.environ, {"BYOC_RUN_MODE": "INITIAL_SYNC"})
+    def test_get_run_mode(self, reset_client, mock_spark):
+
+        assert get_run_mode() == RunMode.INITIAL_SYNC
+
 
 class TestStreamingClient:
 
@@ -395,6 +414,30 @@ def test_streaming_read_write_flow(self, reset_client, mock_spark):
         writer.write_dlo_deltas.assert_called_once_with("target_dll", stream_df)
         assert "source_dll" in client._data_layer_history[DataCloudObjectType.DLO]
 
+    def test_auto_write_to_dlo(self, reset_client, mock_spark):
+        reader = MagicMock(spec=BaseDataCloudReader)
+        writer = MagicMock(spec=BaseDataCloudWriter)
+        mock_df = MagicMock(spec=DataFrame)
+
+        client = StreamingClient(reader=reader, writer=writer)
+        client._record_dlo_access("some_dlo")
+
+        client.auto_write_to_dlo("test_dlo", mock_df)
+
+        writer.auto_write_to_dlo.assert_called_once_with("test_dlo", mock_df)
+
+    def test_auto_write_to_dmo(self, reset_client, mock_spark):
+        reader = MagicMock(spec=BaseDataCloudReader)
+        writer = MagicMock(spec=BaseDataCloudWriter)
+        mock_df = MagicMock(spec=DataFrame)
+
+        client = StreamingClient(reader=reader, writer=writer)
+        client._record_dmo_access("some_dmo")
+
+        client.auto_write_to_dmo("test_dmo", mock_df)
+
+        writer.auto_write_to_dmo.assert_called_once_with("test_dmo", mock_df)
+
 
 class TestSharedSparkSession:
     """Both client types must share a single Spark session (one connection)."""

From 62b02b04e198eb72cb56284752a97dbd4f477172 Mon Sep 17 00:00:00 2001
From: Mark DeLaVergne <mdelavergne@salesforce.com>
Date: Tue, 18 Aug 2026 14:36:20 -0400
Subject: [PATCH 03/10] Fix SF CLI integration mock server

---
 .github/workflows/sf_cli_integration.yml | 17 +++++++--
 scripts/mock_sf_server.py                | 44 +++++++++++++++++++++---
 2 files changed, 54 insertions(+), 7 deletions(-)

diff --git a/.github/workflows/sf_cli_integration.yml b/.github/workflows/sf_cli_integration.yml
index f3c5fb3..7580095 100644
--- a/.github/workflows/sf_cli_integration.yml
+++ b/.github/workflows/sf_cli_integration.yml
@@ -10,6 +10,12 @@ jobs:
     env:
       SF_AUTOUPDATE_DISABLE: true
       NO_COLOR: '1'
+      # The mock server serves TLS with a self-signed cert written to this path.
+      # Point both the CLI (Node) and the SDK (Python requests) at it so they
+      # trust the specific cert rather than disabling verification wholesale.
+      MOCK_SF_CERT_FILE: ${{ github.workspace }}/mock_sf_cert.pem
+      NODE_EXTRA_CA_CERTS: ${{ github.workspace }}/mock_sf_cert.pem
+      REQUESTS_CA_BUNDLE: ${{ github.workspace }}/mock_sf_cert.pem
 
     steps:
       # ── Setup ─────────────────────────────────────────────────────────────────
@@ -56,7 +62,14 @@ jobs:
       # ── Mock Salesforce server + fake org auth ────────────────────────────────
 
     - name: Start mock Salesforce server
-      run: python scripts/mock_sf_server.py &
+      run: |
+        python scripts/mock_sf_server.py &
+        # Wait for the TLS cert the server writes at startup so clients can trust it.
+        for _ in $(seq 1 30); do
+          [ -f "$MOCK_SF_CERT_FILE" ] && break
+          sleep 0.2
+        done
+        test -f "$MOCK_SF_CERT_FILE" || { echo "::error::mock server never wrote $MOCK_SF_CERT_FILE"; exit 1; }
       env:
         MOCK_SF_PORT: '8888'
 
@@ -73,7 +86,7 @@ jobs:
         sfdx_dir.mkdir(exist_ok=True)
         auth = {
             "accessToken": "00D000000000001AAA!fakeTokenForCITesting",
-            "instanceUrl": "http://localhost:8888",
+            "instanceUrl": "https://localhost:8888",
             "loginUrl": "https://login.salesforce.com",
             "orgId": "00D000000000001AAA",
             "userId": "005000000000001AAA",
diff --git a/scripts/mock_sf_server.py b/scripts/mock_sf_server.py
index b296577..0fa279e 100644
--- a/scripts/mock_sf_server.py
+++ b/scripts/mock_sf_server.py
@@ -40,14 +40,21 @@
     python scripts/mock_sf_server.py          # listens on port 8888
     MOCK_SF_PORT=9000 python scripts/mock_sf_server.py
     python scripts/mock_sf_server.py 9000
+
+Serves TLS with a throwaway self-signed cert (the deploy path requires an HTTPS
+upload URL). Set ``MOCK_SF_CERT_FILE`` to a path the clients can trust via
+``NODE_EXTRA_CA_CERTS`` (CLI) and ``REQUESTS_CA_BUNDLE`` (SDK).
 """
 
 from __future__ import annotations
 
-from http.server import BaseHTTPRequestHandler, HTTPServer
 import json
 import os
+import ssl
+import subprocess
 import sys
+import tempfile
+from http.server import BaseHTTPRequestHandler, HTTPServer
 
 PORT = (
     int(sys.argv[1])
@@ -55,6 +62,28 @@
     else int(os.environ.get("MOCK_SF_PORT", "8888"))
 )
 
+
+def _self_signed_cert(dirpath: str) -> tuple[str, str]:
+    """Generate a throwaway self-signed cert for localhost via openssl.
+
+    The plugin's deploy path requires an HTTPS upload URL, so the server must
+    speak TLS. Set ``MOCK_SF_CERT_FILE`` to write the cert to a known path so
+    clients can trust it (``REQUESTS_CA_BUNDLE`` / ``NODE_EXTRA_CA_CERTS``).
+    """
+    cert_path = os.environ.get("MOCK_SF_CERT_FILE") or os.path.join(dirpath, "cert.pem")
+    key_path = os.path.join(dirpath, "key.pem")
+    subprocess.run(
+        [
+            "openssl", "req", "-x509", "-newkey", "rsa:2048", "-nodes",
+            "-keyout", key_path, "-out", cert_path, "-days", "1",
+            "-subj", "/CN=localhost",
+            "-addext", "subjectAltName=DNS:localhost,IP:127.0.0.1",
+        ],
+        check=True,
+        capture_output=True,
+    )
+    return cert_path, key_path
+
 _USERINFO = {
     "sub": "https://test.salesforce.com/id/00D000000000001AAA/005000000000001AAA",
     "user_id": "005000000000001AAA",
@@ -68,7 +97,7 @@
 
 _TOKEN_RESPONSE = {
     "access_token": "00D000000000001AAA!fakeAccessTokenForCITesting",
-    "instance_url": f"http://localhost:{PORT}",
+    "instance_url": f"https://localhost:{PORT}",
     "token_type": "Bearer",
     "scope": "api",
 }
@@ -135,7 +164,7 @@ def do_POST(self) -> None:
         elif path == _DATA_CUSTOM_CODE_PATH:
             # create_deployment() — return a presigned upload URL
             self._send_json(
-                {"fileUploadUrl": f"http://localhost:{PORT}/upload/fake-deployment.zip"}
+                {"fileUploadUrl": f"https://localhost:{PORT}/upload/fake-deployment.zip"}
             )
         elif path == _DATA_TRANSFORMS_PATH:
             # create_data_transform() — script packages only
@@ -152,5 +181,10 @@ def do_PUT(self) -> None:
 if __name__ == "__main__":
     server = HTTPServer(("localhost", PORT), MockSFHandler)
     server.allow_reuse_address = True
-    print(f"[MOCK SF] Listening on http://localhost:{PORT}", flush=True)
-    server.serve_forever()
+    with tempfile.TemporaryDirectory() as certdir:
+        cert_path, key_path = _self_signed_cert(certdir)
+        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
+        ctx.load_cert_chain(cert_path, key_path)
+        server.socket = ctx.wrap_socket(server.socket, server_side=True)
+        print(f"[MOCK SF] Listening on https://localhost:{PORT}", flush=True)
+        server.serve_forever()

From 9fe6d768804ef51e3addd44cc6456e8952d7dca1 Mon Sep 17 00:00:00 2001
From: Mark DeLaVergne <mdelavergne@salesforce.com>
Date: Tue, 18 Aug 2026 14:42:07 -0400
Subject: [PATCH 04/10] Selective fix for SF CLI mock server

---
 .github/workflows/sf_cli_integration.yml | 21 ++++++++++++++++-----
 1 file changed, 16 insertions(+), 5 deletions(-)

diff --git a/.github/workflows/sf_cli_integration.yml b/.github/workflows/sf_cli_integration.yml
index 7580095..d309d0e 100644
--- a/.github/workflows/sf_cli_integration.yml
+++ b/.github/workflows/sf_cli_integration.yml
@@ -10,12 +10,11 @@ jobs:
     env:
       SF_AUTOUPDATE_DISABLE: true
       NO_COLOR: '1'
-      # The mock server serves TLS with a self-signed cert written to this path.
-      # Point both the CLI (Node) and the SDK (Python requests) at it so they
-      # trust the specific cert rather than disabling verification wholesale.
+      # Path the mock server writes its self-signed TLS cert to. The CA-trust
+      # env vars (NODE_EXTRA_CA_CERTS / REQUESTS_CA_BUNDLE) are set per-step on
+      # only the run/deploy steps — setting them job-wide would point pip/poetry
+      # at a cert file that does not exist yet during setup, breaking installs.
       MOCK_SF_CERT_FILE: ${{ github.workspace }}/mock_sf_cert.pem
-      NODE_EXTRA_CA_CERTS: ${{ github.workspace }}/mock_sf_cert.pem
-      REQUESTS_CA_BUNDLE: ${{ github.workspace }}/mock_sf_cert.pem
 
     steps:
       # ── Setup ─────────────────────────────────────────────────────────────────
@@ -177,6 +176,9 @@ jobs:
       # ── Script: run ───────────────────────────────────────────────────────────
 
     - name: '[script] run — sf data-code-extension script run --entrypoint testScript/payload/entrypoint.py -o dev1'
+      env:
+        NODE_EXTRA_CA_CERTS: ${{ github.workspace }}/mock_sf_cert.pem
+        REQUESTS_CA_BUNDLE: ${{ github.workspace }}/mock_sf_cert.pem
       run: |
         sf data-code-extension script run \
           --entrypoint testScript/payload/entrypoint.py \
@@ -188,6 +190,9 @@ jobs:
       # ── Script: deploy ───────────────────────────────────────────────────────
 
     - name: '[script] deploy — sf data-code-extension script deploy'
+      env:
+        NODE_EXTRA_CA_CERTS: ${{ github.workspace }}/mock_sf_cert.pem
+        REQUESTS_CA_BUNDLE: ${{ github.workspace }}/mock_sf_cert.pem
       run: |
         sf data-code-extension script deploy \
           --name test-script-deploy \
@@ -275,6 +280,9 @@ jobs:
       # ── Function: run ─────────────────────────────────────────────────────────
 
     - name: '[function] run — sf data-code-extension function run --entrypoint testFunction/payload/entrypoint.py  --test-with testFunction/payload/tests/test.json -o dev1'
+      env:
+        NODE_EXTRA_CA_CERTS: ${{ github.workspace }}/mock_sf_cert.pem
+        REQUESTS_CA_BUNDLE: ${{ github.workspace }}/mock_sf_cert.pem
       run: |
         sf data-code-extension function run \
           --entrypoint testFunction/payload/entrypoint.py \
@@ -286,6 +294,9 @@ jobs:
       # ── Function: deploy ─────────────────────────────────────────────────────
 
     - name: '[function] deploy — sf data-code-extension function deploy'
+      env:
+        NODE_EXTRA_CA_CERTS: ${{ github.workspace }}/mock_sf_cert.pem
+        REQUESTS_CA_BUNDLE: ${{ github.workspace }}/mock_sf_cert.pem
       run: |
         sf data-code-extension function deploy \
           --name test-function-deploy \

From e35cdb1e44544c01359147070442c3f166516b3a Mon Sep 17 00:00:00 2001
From: Mark DeLaVergne <mdelavergne@salesforce.com>
Date: Tue, 18 Aug 2026 14:58:46 -0400
Subject: [PATCH 05/10] Fix lint issues

---
 scripts/mock_sf_server.py | 27 +++++++++++++++++++++------
 1 file changed, 21 insertions(+), 6 deletions(-)

diff --git a/scripts/mock_sf_server.py b/scripts/mock_sf_server.py
index 0fa279e..0f48d5d 100644
--- a/scripts/mock_sf_server.py
+++ b/scripts/mock_sf_server.py
@@ -48,13 +48,13 @@
 
 from __future__ import annotations
 
+from http.server import BaseHTTPRequestHandler, HTTPServer
 import json
 import os
 import ssl
 import subprocess
 import sys
 import tempfile
-from http.server import BaseHTTPRequestHandler, HTTPServer
 
 PORT = (
     int(sys.argv[1])
@@ -74,16 +74,29 @@ def _self_signed_cert(dirpath: str) -> tuple[str, str]:
     key_path = os.path.join(dirpath, "key.pem")
     subprocess.run(
         [
-            "openssl", "req", "-x509", "-newkey", "rsa:2048", "-nodes",
-            "-keyout", key_path, "-out", cert_path, "-days", "1",
-            "-subj", "/CN=localhost",
-            "-addext", "subjectAltName=DNS:localhost,IP:127.0.0.1",
+            "openssl",
+            "req",
+            "-x509",
+            "-newkey",
+            "rsa:2048",
+            "-nodes",
+            "-keyout",
+            key_path,
+            "-out",
+            cert_path,
+            "-days",
+            "1",
+            "-subj",
+            "/CN=localhost",
+            "-addext",
+            "subjectAltName=DNS:localhost,IP:127.0.0.1",
         ],
         check=True,
         capture_output=True,
     )
     return cert_path, key_path
 
+
 _USERINFO = {
     "sub": "https://test.salesforce.com/id/00D000000000001AAA/005000000000001AAA",
     "user_id": "005000000000001AAA",
@@ -164,7 +177,9 @@ def do_POST(self) -> None:
         elif path == _DATA_CUSTOM_CODE_PATH:
             # create_deployment() — return a presigned upload URL
             self._send_json(
-                {"fileUploadUrl": f"https://localhost:{PORT}/upload/fake-deployment.zip"}
+                {
+                    "fileUploadUrl": f"https://localhost:{PORT}/upload/fake-deployment.zip"
+                }
             )
         elif path == _DATA_TRANSFORMS_PATH:
             # create_data_transform() — script packages only

From 76de8a14d75b9a2639790669f8accc56c238601f Mon Sep 17 00:00:00 2001
From: Steffan Byrne <sbyrne@salesforce.com>
Date: Tue, 18 Aug 2026 17:12:56 -0400
Subject: [PATCH 06/10] Added read methods to streaming client.

These are similar to the delta methods but are meant for the other run
modes.
---
 src/datacustomcode/client.py                  | 17 ++++++++++++
 src/datacustomcode/io/writer/csv.py           |  2 ++
 .../examples/streaming_deltas/entrypoint.py   | 27 +++++++++----------
 3 files changed, 31 insertions(+), 15 deletions(-)

diff --git a/src/datacustomcode/client.py b/src/datacustomcode/client.py
index 7836c70..579e9d2 100644
--- a/src/datacustomcode/client.py
+++ b/src/datacustomcode/client.py
@@ -559,6 +559,15 @@ class StreamingClient(_BaseClient):
 
     _instance: ClassVar[Optional[StreamingClient]] = None
 
+    def read_dlo(self) -> PySparkDataFrame:
+        """Read the streamingSource
+
+        Returns:
+            A standard PySpark DataFrame from the streaming source DLO
+        """
+        self._record_dlo_access(_streaming_source_name())
+        return self._reader.read_dlo(_streaming_source_name())
+
     def read_dlo_deltas(self) -> PySparkDataFrame:
         """Read the streaming change feed (deltas) for a DLO from Data Cloud.
 
@@ -572,6 +581,14 @@ def read_dlo_deltas(self) -> PySparkDataFrame:
         self._record_dlo_access(_streaming_source_name())
         return self._reader.read_dlo_deltas()  # type: ignore[no-any-return]
 
+    def read_dmo(self) -> PySparkDataFrame:
+        """Read the streamingSource
+
+        Returns a standard PySpark DataFrame from the streaming source DMO
+        """
+        self._record_dmo_access(_streaming_source_name())
+        return self._reader.read_dmo(_streaming_source_name())
+
     def read_dmo_deltas(self) -> PySparkDataFrame:
         """Read the streaming change feed (deltas) for a DMO from Data Cloud.
 
diff --git a/src/datacustomcode/io/writer/csv.py b/src/datacustomcode/io/writer/csv.py
index 7505026..292d92d 100644
--- a/src/datacustomcode/io/writer/csv.py
+++ b/src/datacustomcode/io/writer/csv.py
@@ -37,6 +37,7 @@ def write_to_dlo(
         dataframe.write.csv(name, mode=write_mode)
 
     def auto_write_to_dlo(self, name: str, dataframe: PySparkDataFrame) -> None:
+        # use overwrite since this is a local only writer
         self.write_to_dlo(name, dataframe, WriteMode.OVERWRITE)
 
     def write_to_dmo(
@@ -48,4 +49,5 @@ def write_to_dmo(
         dataframe.write.csv(name, mode=write_mode)
 
     def auto_write_to_dmo(self, name: str, dataframe: PySparkDataFrame) -> None:
+        # use overwrite since this is a local only writer
         self.write_to_dmo(name, dataframe, WriteMode.OVERWRITE)
diff --git a/src/datacustomcode/templates/script/examples/streaming_deltas/entrypoint.py b/src/datacustomcode/templates/script/examples/streaming_deltas/entrypoint.py
index fc1b6cc..abe7981 100644
--- a/src/datacustomcode/templates/script/examples/streaming_deltas/entrypoint.py
+++ b/src/datacustomcode/templates/script/examples/streaming_deltas/entrypoint.py
@@ -3,22 +3,19 @@
 This example is the streaming counterpart to a normal batch entrypoint. Instead
 of a batch ``Client`` with ``read_dlo`` / ``write_to_dlo`` (which read and write
 a bounded snapshot), it uses a :class:`StreamingClient` and its streaming delta
-methods:
+methods.
 
-* ``client.read_dlo_deltas()`` returns a *streaming* DataFrame over the
-  Change Data Feed of the source DLO. Each row carries the source columns plus
-  change-feed metadata columns (``_record_type``, ``_commit_*``).
-* ``client.write_dlo_deltas(name, df)`` starts a streaming query that writes
-  each micro-batch to the target DLO and returns the ``StreamingQuery`` handle.
-  The runtime owns the trigger, and checkpoint location — the caller only
-  chooses the table.
+The first run of a streaming job will use the run mode INITIAL_SYNC which behaves
+like a batch run on the streaming source. A streaming transform can also use run
+mode REBUILD to do the same thing on demand. Note that these will process all
+source rows and overwrite the target.
 
 The transform in between is ordinary PySpark. Because the source is a change
 feed, keep the metadata columns on the DataFrame you hand to
 ``write_dlo_deltas`` — the sink relies on them to merge changes correctly.
 
-This entrypoint only runs inside the Data Cloud streaming (``DELTA_SYNC``)
-runtime; the local ``datacustomcode run`` readers/writers raise
+This entrypoint only runs inside the Data Cloud runtime;
+ the local ``datacustomcode run`` readers/writers raise
 ``NotImplementedError`` for the delta methods.
 """
 
@@ -26,7 +23,6 @@
 from pyspark.sql.functions import col, upper
 
 from datacustomcode.client import (
-    Client,
     RunMode,
     StreamingClient,
     get_run_mode,
@@ -34,10 +30,10 @@
 
 
 def main():
-    source_dlo = "Account_std__dll"
     target_dlo = "Account_std_copy__dll"
+    client = StreamingClient()
+
     if get_run_mode() == RunMode.DELTA_SYNC:
-        client = StreamingClient()
         # Streaming DataFrame over the source DLO's change feed.
         dataframe = client.read_dlo_deltas()
         # Ordinary PySpark transform.
@@ -51,8 +47,9 @@ def main():
         # the job is stopped by the platform.
         query.awaitTermination()
     else:
-        client = Client()
-        dataframe = client.read_dlo(source_dlo)
+        # initial sync and rebuild read the entire streaming source DLO and
+        # write using a server-decided mode based on the run mode
+        dataframe = client.read_dlo()
         transformed = transform(dataframe)
         client.auto_write_to_dlo(target_dlo, transformed)
 

From 650be6bcdf554658be141a69cf7dd710f53a97b4 Mon Sep 17 00:00:00 2001
From: Mark DeLaVergne <mdelavergne@salesforce.com>
Date: Wed, 19 Aug 2026 10:28:25 -0400
Subject: [PATCH 07/10] Generate mock certs as a dedicated step

---
 .github/workflows/sf_cli_integration.yml | 38 ++++++-------
 scripts/mock_sf_server.py                | 70 ++++++++----------------
 2 files changed, 41 insertions(+), 67 deletions(-)

diff --git a/.github/workflows/sf_cli_integration.yml b/.github/workflows/sf_cli_integration.yml
index d309d0e..7dbf65b 100644
--- a/.github/workflows/sf_cli_integration.yml
+++ b/.github/workflows/sf_cli_integration.yml
@@ -10,11 +10,9 @@ jobs:
     env:
       SF_AUTOUPDATE_DISABLE: true
       NO_COLOR: '1'
-      # Path the mock server writes its self-signed TLS cert to. The CA-trust
-      # env vars (NODE_EXTRA_CA_CERTS / REQUESTS_CA_BUNDLE) are set per-step on
-      # only the run/deploy steps — setting them job-wide would point pip/poetry
-      # at a cert file that does not exist yet during setup, breaking installs.
+      # Mock server's TLS cert/key pair.
       MOCK_SF_CERT_FILE: ${{ github.workspace }}/mock_sf_cert.pem
+      MOCK_SF_KEY_FILE: ${{ github.workspace }}/mock_sf_key.pem
 
     steps:
       # ── Setup ─────────────────────────────────────────────────────────────────
@@ -22,6 +20,13 @@ jobs:
     - name: Checkout code
       uses: actions/checkout@v4
 
+    - name: Generate mock server TLS cert
+      run: |
+        openssl req -x509 -newkey rsa:2048 -nodes \
+          -keyout "$MOCK_SF_KEY_FILE" -out "$MOCK_SF_CERT_FILE" \
+          -days 1 -subj "/CN=localhost" \
+          -addext "subjectAltName=DNS:localhost,IP:127.0.0.1"
+
     - name: Set up Python 3.11
       uses: actions/setup-python@v5
       with:
@@ -61,14 +66,7 @@ jobs:
       # ── Mock Salesforce server + fake org auth ────────────────────────────────
 
     - name: Start mock Salesforce server
-      run: |
-        python scripts/mock_sf_server.py &
-        # Wait for the TLS cert the server writes at startup so clients can trust it.
-        for _ in $(seq 1 30); do
-          [ -f "$MOCK_SF_CERT_FILE" ] && break
-          sleep 0.2
-        done
-        test -f "$MOCK_SF_CERT_FILE" || { echo "::error::mock server never wrote $MOCK_SF_CERT_FILE"; exit 1; }
+      run: python scripts/mock_sf_server.py &
       env:
         MOCK_SF_PORT: '8888'
 
@@ -177,8 +175,8 @@ jobs:
 
     - name: '[script] run — sf data-code-extension script run --entrypoint testScript/payload/entrypoint.py -o dev1'
       env:
-        NODE_EXTRA_CA_CERTS: ${{ github.workspace }}/mock_sf_cert.pem
-        REQUESTS_CA_BUNDLE: ${{ github.workspace }}/mock_sf_cert.pem
+        NODE_EXTRA_CA_CERTS: ${{ env.MOCK_SF_CERT_FILE }}
+        REQUESTS_CA_BUNDLE: ${{ env.MOCK_SF_CERT_FILE }}
       run: |
         sf data-code-extension script run \
           --entrypoint testScript/payload/entrypoint.py \
@@ -191,8 +189,8 @@ jobs:
 
     - name: '[script] deploy — sf data-code-extension script deploy'
       env:
-        NODE_EXTRA_CA_CERTS: ${{ github.workspace }}/mock_sf_cert.pem
-        REQUESTS_CA_BUNDLE: ${{ github.workspace }}/mock_sf_cert.pem
+        NODE_EXTRA_CA_CERTS: ${{ env.MOCK_SF_CERT_FILE }}
+        REQUESTS_CA_BUNDLE: ${{ env.MOCK_SF_CERT_FILE }}
       run: |
         sf data-code-extension script deploy \
           --name test-script-deploy \
@@ -281,8 +279,8 @@ jobs:
 
     - name: '[function] run — sf data-code-extension function run --entrypoint testFunction/payload/entrypoint.py  --test-with testFunction/payload/tests/test.json -o dev1'
       env:
-        NODE_EXTRA_CA_CERTS: ${{ github.workspace }}/mock_sf_cert.pem
-        REQUESTS_CA_BUNDLE: ${{ github.workspace }}/mock_sf_cert.pem
+        NODE_EXTRA_CA_CERTS: ${{ env.MOCK_SF_CERT_FILE }}
+        REQUESTS_CA_BUNDLE: ${{ env.MOCK_SF_CERT_FILE }}
       run: |
         sf data-code-extension function run \
           --entrypoint testFunction/payload/entrypoint.py \
@@ -295,8 +293,8 @@ jobs:
 
     - name: '[function] deploy — sf data-code-extension function deploy'
       env:
-        NODE_EXTRA_CA_CERTS: ${{ github.workspace }}/mock_sf_cert.pem
-        REQUESTS_CA_BUNDLE: ${{ github.workspace }}/mock_sf_cert.pem
+        NODE_EXTRA_CA_CERTS: ${{ env.MOCK_SF_CERT_FILE }}
+        REQUESTS_CA_BUNDLE: ${{ env.MOCK_SF_CERT_FILE }}
       run: |
         sf data-code-extension function deploy \
           --name test-function-deploy \
diff --git a/scripts/mock_sf_server.py b/scripts/mock_sf_server.py
index 0f48d5d..8012f89 100644
--- a/scripts/mock_sf_server.py
+++ b/scripts/mock_sf_server.py
@@ -41,9 +41,16 @@
     MOCK_SF_PORT=9000 python scripts/mock_sf_server.py
     python scripts/mock_sf_server.py 9000
 
-Serves TLS with a throwaway self-signed cert (the deploy path requires an HTTPS
-upload URL). Set ``MOCK_SF_CERT_FILE`` to a path the clients can trust via
-``NODE_EXTRA_CA_CERTS`` (CLI) and ``REQUESTS_CA_BUNDLE`` (SDK).
+Serves TLS (the deploy path requires an HTTPS upload URL) using a pre-generated
+cert/key pair — this script does not generate one. Set ``MOCK_SF_CERT_FILE`` /
+``MOCK_SF_KEY_FILE`` to the pair's paths; generate a throwaway one with:
+
+    openssl req -x509 -newkey rsa:2048 -nodes -keyout key.pem -out cert.pem \\
+      -days 1 -subj "/CN=localhost" \\
+      -addext "subjectAltName=DNS:localhost,IP:127.0.0.1"
+
+Point clients at the cert so they trust it: ``NODE_EXTRA_CA_CERTS`` (CLI) and
+``REQUESTS_CA_BUNDLE`` (SDK).
 """
 
 from __future__ import annotations
@@ -52,9 +59,7 @@
 import json
 import os
 import ssl
-import subprocess
 import sys
-import tempfile
 
 PORT = (
     int(sys.argv[1])
@@ -62,41 +67,6 @@
     else int(os.environ.get("MOCK_SF_PORT", "8888"))
 )
 
-
-def _self_signed_cert(dirpath: str) -> tuple[str, str]:
-    """Generate a throwaway self-signed cert for localhost via openssl.
-
-    The plugin's deploy path requires an HTTPS upload URL, so the server must
-    speak TLS. Set ``MOCK_SF_CERT_FILE`` to write the cert to a known path so
-    clients can trust it (``REQUESTS_CA_BUNDLE`` / ``NODE_EXTRA_CA_CERTS``).
-    """
-    cert_path = os.environ.get("MOCK_SF_CERT_FILE") or os.path.join(dirpath, "cert.pem")
-    key_path = os.path.join(dirpath, "key.pem")
-    subprocess.run(
-        [
-            "openssl",
-            "req",
-            "-x509",
-            "-newkey",
-            "rsa:2048",
-            "-nodes",
-            "-keyout",
-            key_path,
-            "-out",
-            cert_path,
-            "-days",
-            "1",
-            "-subj",
-            "/CN=localhost",
-            "-addext",
-            "subjectAltName=DNS:localhost,IP:127.0.0.1",
-        ],
-        check=True,
-        capture_output=True,
-    )
-    return cert_path, key_path
-
-
 _USERINFO = {
     "sub": "https://test.salesforce.com/id/00D000000000001AAA/005000000000001AAA",
     "user_id": "005000000000001AAA",
@@ -194,12 +164,18 @@ def do_PUT(self) -> None:
 
 
 if __name__ == "__main__":
+    cert_path = os.environ.get("MOCK_SF_CERT_FILE")
+    key_path = os.environ.get("MOCK_SF_KEY_FILE")
+    if not cert_path or not key_path:
+        sys.exit(
+            "MOCK_SF_CERT_FILE and MOCK_SF_KEY_FILE must both be set to an "
+            "existing TLS cert/key pair — see the module docstring."
+        )
+
     server = HTTPServer(("localhost", PORT), MockSFHandler)
     server.allow_reuse_address = True
-    with tempfile.TemporaryDirectory() as certdir:
-        cert_path, key_path = _self_signed_cert(certdir)
-        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
-        ctx.load_cert_chain(cert_path, key_path)
-        server.socket = ctx.wrap_socket(server.socket, server_side=True)
-        print(f"[MOCK SF] Listening on https://localhost:{PORT}", flush=True)
-        server.serve_forever()
+    ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
+    ctx.load_cert_chain(cert_path, key_path)
+    server.socket = ctx.wrap_socket(server.socket, server_side=True)
+    print(f"[MOCK SF] Listening on https://localhost:{PORT}", flush=True)
+    server.serve_forever()

From a847b20dbf196a9948401a18b40d25743b04fa34 Mon Sep 17 00:00:00 2001
From: Mark DeLaVergne <mdelavergne@salesforce.com>
Date: Wed, 19 Aug 2026 10:31:30 -0400
Subject: [PATCH 08/10] Utilize temp dir instead of workspace

---
 .github/workflows/sf_cli_integration.yml | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/.github/workflows/sf_cli_integration.yml b/.github/workflows/sf_cli_integration.yml
index 7dbf65b..d2bde49 100644
--- a/.github/workflows/sf_cli_integration.yml
+++ b/.github/workflows/sf_cli_integration.yml
@@ -11,8 +11,8 @@ jobs:
       SF_AUTOUPDATE_DISABLE: true
       NO_COLOR: '1'
       # Mock server's TLS cert/key pair.
-      MOCK_SF_CERT_FILE: ${{ github.workspace }}/mock_sf_cert.pem
-      MOCK_SF_KEY_FILE: ${{ github.workspace }}/mock_sf_key.pem
+      MOCK_SF_CERT_FILE: ${{ runner.temp }}/mock_sf_cert.pem
+      MOCK_SF_KEY_FILE: ${{ runner.temp }}/mock_sf_key.pem
 
     steps:
       # ── Setup ─────────────────────────────────────────────────────────────────

From 448188d7d573d796f69e1f87bf5dc842510a4d53 Mon Sep 17 00:00:00 2001
From: Mark DeLaVergne <mdelavergne@salesforce.com>
Date: Wed, 19 Aug 2026 10:34:24 -0400
Subject: [PATCH 09/10] Fix runner temp usage

---
 .github/workflows/sf_cli_integration.yml | 8 +++++---
 1 file changed, 5 insertions(+), 3 deletions(-)

diff --git a/.github/workflows/sf_cli_integration.yml b/.github/workflows/sf_cli_integration.yml
index d2bde49..c320f8b 100644
--- a/.github/workflows/sf_cli_integration.yml
+++ b/.github/workflows/sf_cli_integration.yml
@@ -10,9 +10,6 @@ jobs:
     env:
       SF_AUTOUPDATE_DISABLE: true
       NO_COLOR: '1'
-      # Mock server's TLS cert/key pair.
-      MOCK_SF_CERT_FILE: ${{ runner.temp }}/mock_sf_cert.pem
-      MOCK_SF_KEY_FILE: ${{ runner.temp }}/mock_sf_key.pem
 
     steps:
       # ── Setup ─────────────────────────────────────────────────────────────────
@@ -20,6 +17,11 @@ jobs:
     - name: Checkout code
       uses: actions/checkout@v4
 
+    - name: Set mock server TLS cert paths
+      run: |
+        echo "MOCK_SF_CERT_FILE=$RUNNER_TEMP/mock_sf_cert.pem" >> "$GITHUB_ENV"
+        echo "MOCK_SF_KEY_FILE=$RUNNER_TEMP/mock_sf_key.pem" >> "$GITHUB_ENV"
+
     - name: Generate mock server TLS cert
       run: |
         openssl req -x509 -newkey rsa:2048 -nodes \

From 7ba0f1a08f3b808c1d90416c32d0fed4570673c7 Mon Sep 17 00:00:00 2001
From: Steffan Byrne <sbyrne@salesforce.com>
Date: Wed, 19 Aug 2026 10:49:48 -0400
Subject: [PATCH 10/10] Fixed a few more things.

---
 src/datacustomcode/client.py         | 5 ++++-
 src/datacustomcode/io/writer/base.py | 4 ++--
 tests/test_client.py                 | 5 +++++
 3 files changed, 11 insertions(+), 3 deletions(-)

diff --git a/src/datacustomcode/client.py b/src/datacustomcode/client.py
index 579e9d2..740b8b9 100644
--- a/src/datacustomcode/client.py
+++ b/src/datacustomcode/client.py
@@ -650,4 +650,7 @@ class RunMode(Enum):
 def get_run_mode() -> RunMode:
     """Read and validate the BYOC_RUN_MODE env var; default to BATCH when unset."""
     run_mode = os.getenv("BYOC_RUN_MODE", "BATCH").upper()
-    return RunMode(run_mode)
+    try:
+        return RunMode(run_mode)
+    except ValueError as exc:
+        raise ValueError("Set BYOC_RUN_MODE to a valid value") from exc
diff --git a/src/datacustomcode/io/writer/base.py b/src/datacustomcode/io/writer/base.py
index a37e226..d24a33b 100644
--- a/src/datacustomcode/io/writer/base.py
+++ b/src/datacustomcode/io/writer/base.py
@@ -59,17 +59,17 @@ def write_to_dmo(
         self, name: str, dataframe: PySparkDataFrame, write_mode: WriteMode
     ) -> None: ...
 
-    @abstractmethod
     def auto_write_to_dlo(self, name: str, dataframe: PySparkDataFrame) -> None:
         """Write to a DLO automatically picking the write mode.
         For use with streaming transforms when running in rebuild or initial sync mode.
         """
+        raise NotImplementedError
 
-    @abstractmethod
     def auto_write_to_dmo(self, name: str, dataframe: PySparkDataFrame) -> None:
         """Write to a DMO automatically picking the write mode.
         For use with streaming transforms when running in rebuild or initial sync mode.
         """
+        raise NotImplementedError
 
     def write_dlo_deltas(
         self, name: str, dataframe: PySparkDataFrame
diff --git a/tests/test_client.py b/tests/test_client.py
index 042f352..05399eb 100644
--- a/tests/test_client.py
+++ b/tests/test_client.py
@@ -284,6 +284,11 @@ def test_get_run_mode(self, reset_client, mock_spark):
 
         assert get_run_mode() == RunMode.INITIAL_SYNC
 
+    @patch.dict(os.environ, {"BYOC_RUN_MODE": "INVALID"})
+    def test_get_run_mode_throws(self):
+        with pytest.raises(ValueError, match="Set BYOC_RUN_MODE to a valid value"):
+            get_run_mode()
+
 
 class TestStreamingClient: