jason810496 · Oct 9, 2026 · Oct 7, 2026 · Oct 7, 2026 · Oct 7, 2026 · Oct 8, 2026
@@ -22,8 +22,15 @@
 # image can exec it directly -- no extra runtime needs to be installed. We only
 # bind-mount the bundle into the directory the ExecutableCoordinator scans and
 # point the worker at the "golang" queue where @task.stub tasks are routed.
+# The native-Dag bundle sits in the Dags folder, so the Dag processor execs it
+# to parse its Dags, with the same USER and HOME as the worker.
 ---
 services:
+  airflow-dag-processor:
+    environment:
+      USER: airflow
+      HOME: /home/airflow
+
   airflow-worker:
     # The bundle is built with CGO_ENABLED=0, so the SDK's user.Current() call at
     # init falls back to Go's pure-Go resolver, which reads $USER / $HOME and

@@ -44,6 +44,8 @@
     GO_SDK_BUNDLE_NAME,
     GO_SDK_DAGS_PATH,
     GO_SDK_EXAMPLE_BUNDLE_PKG,
+    GO_SDK_NATIVE_BUNDLE_NAME,
+    GO_SDK_NATIVE_BUNDLE_PKG,
     GO_SDK_ROOT_PATH,
     GO_SDK_STATE_STORE_RETENTION_DAYS,
     JAVA_COMPOSE_PATH,
@@ -536,10 +538,10 @@ def _setup_java_sdk_integration(dot_env_file, tmp_dir):
     os.environ["ENV_FILE_PATH"] = str(dot_env_file)
 
 
-def _run_go_sdk_pack(output_path, *, capture_output=False, native=False):
+def _run_go_sdk_pack(package, output_path, *, capture_output=False, native=False):
     """Run ``go tool airflow-go-pack`` natively or inside the pinned Go toolchain container.
 
-    ``go tool airflow-go-pack`` builds the bundle package, reads its
+    ``go tool airflow-go-pack`` builds the bundle *package*, reads its
     --airflow-metadata, and appends the source + airflow-metadata.yaml + the
     AFBNDL01 trailer, writing a single self-contained executable bundle.
     CGO_ENABLED=0 yields a fully static binary that runs on the stock worker.
@@ -569,7 +571,7 @@ def _run_go_sdk_pack(output_path, *, capture_output=False, native=False):
             "airflow-go-pack",
             "--output",
             str(output_path),
-            GO_SDK_EXAMPLE_BUNDLE_PKG,
+            package,
         ]
     else:
         cwd = None
@@ -600,20 +602,20 @@ def _run_go_sdk_pack(output_path, *, capture_output=False, native=False):
             "airflow-go-pack",
             "--output",
             f"{container_bin_dir}/{output_path.name}",
-            GO_SDK_EXAMPLE_BUNDLE_PKG,
+            package,
         ]
     return subprocess.run(argv, cwd=cwd, env=env, check=True, capture_output=capture_output, text=True)
 
 
-def _pack_go_sdk_example_bundle(*, native=False):
-    """Build the Go SDK example bundle, capturing output so a failure prints the build log."""
-    output_path = GO_SDK_BIN_PATH / GO_SDK_BUNDLE_NAME
+def _pack_go_sdk_bundle(package, name, *, native=False):
+    """Build a Go SDK bundle, capturing output so a failure prints the build log."""
+    output_path = GO_SDK_BIN_PATH / name
     mode_label = "host toolchain" if native else GO_BUILDER_IMAGE
-    console.print(f"[yellow]Building Go SDK example bundle ({mode_label})...")
+    console.print(f"[yellow]Building Go SDK bundle {package} ({mode_label})...")
     try:
-        completed = _run_go_sdk_pack(output_path, capture_output=True, native=native)
+        completed = _run_go_sdk_pack(package, output_path, capture_output=True, native=native)
     except subprocess.CalledProcessError as e:
-        console.print("[red]Go SDK example bundle build failed:")
+        console.print(f"[red]Go SDK bundle {package} build failed:")
         console.print(e.stdout, e.stderr, sep="\n", markup=False, soft_wrap=True)
         raise
     console.print(completed.stdout, completed.stderr, sep="\n", markup=False, soft_wrap=True)
@@ -627,11 +629,16 @@ def _setup_go_sdk_integration(dot_env_file, tmp_dir):
     as the ``go-task-handlers`` Dag bundle, copies the Python stub Dag, and
     writes the coordinator configuration.
 
+    The native-Dag example is packed too, into the Dags folder, where the Dag
+    processor runs it to parse its Dags. Its tasks run on the ``golang`` queue
+    through the same coordinator.
+
     The packed bundle is a statically linked native executable (built with
     ``CGO_ENABLED=0``), so the stock Airflow worker image can exec it directly
     without a Go toolchain or any extra runtime installed -- see ``go.yml``.
     """
-    _pack_go_sdk_example_bundle(native=LANG_SDK_NATIVE_TOOLCHAIN)
+    _pack_go_sdk_bundle(GO_SDK_EXAMPLE_BUNDLE_PKG, GO_SDK_BUNDLE_NAME, native=LANG_SDK_NATIVE_TOOLCHAIN)
+    _pack_go_sdk_bundle(GO_SDK_NATIVE_BUNDLE_PKG, GO_SDK_NATIVE_BUNDLE_NAME, native=LANG_SDK_NATIVE_TOOLCHAIN)
 
     # Copy the compose override into the temp directory.
     copyfile(GO_COMPOSE_PATH, tmp_dir / "go.yml")
@@ -644,13 +651,25 @@ def _setup_go_sdk_integration(dot_env_file, tmp_dir):
     packed_bundle = go_bundles_dir / GO_SDK_BUNDLE_NAME
     copyfile(GO_SDK_BIN_PATH / GO_SDK_BUNDLE_NAME, packed_bundle)
     os.chmod(packed_bundle, 0o755)
+    # This bundle only registers task handlers, so the Dag processor must not parse it if the
+    # directory is ever mounted there. Tasks do not read .airflowignore.
+    (go_bundles_dir / ".airflowignore").write_text("*\n")
+
+    # The native-Dag bundle goes into the Dags folder, where the Dag processor runs it to parse.
+    native_dags_dir = tmp_dir / "dags" / "go"
+    native_dags_dir.mkdir()
+    packed_native_bundle = native_dags_dir / GO_SDK_NATIVE_BUNDLE_NAME
+    copyfile(GO_SDK_BIN_PATH / GO_SDK_NATIVE_BUNDLE_NAME, packed_native_bundle)
+    os.chmod(packed_native_bundle, 0o755)
 
     # Copy the Go SDK example stub Dag so Airflow can discover and serialize it.
     copyfile(GO_SDK_DAGS_PATH / "go_examples.py", tmp_dir / "dags" / "go_examples.py")
 
     # Coordinator registry: maps the logical name "go-sdk" to ExecutableCoordinator,
     # which scans the go-task-handlers Dag bundle for the packed bundle by dag_id.
-    # Queue mapping: routes tasks on the "golang" queue to "go-sdk".
+    # Queue mapping: routes tasks on the "golang" queue to "go-sdk". It is the only
+    # ExecutableCoordinator, so it also parses the native Dags in the Dags folder, and runs
+    # their tasks from that bundle, not from go-task-handlers.
     dag_bundle_config = _build_dag_bundle_config({"go-task-handlers": "/opt/airflow/go-bundles"})
     coordinator_config = json.dumps(
         {

@@ -92,6 +92,9 @@
 GO_SDK_EXAMPLE_BUNDLE_PKG = "./example/bundle"
 # Name of the packed bundle binary (matches the example bundle's package dir name).
 GO_SDK_BUNDLE_NAME = "example_dags"
+# The native-Dag example: Dags declared in Go and parsed from the binary by the Dag processor.
+GO_SDK_NATIVE_BUNDLE_PKG = "./example/native"
+GO_SDK_NATIVE_BUNDLE_NAME = "go_native_dags"
 # Where airflow-go-pack writes the packed bundle inside the repo (go-sdk/bin is gitignored).
 GO_SDK_BIN_PATH = GO_SDK_ROOT_PATH / "bin"
 GO_COMPOSE_PATH = AIRFLOW_ROOT_PATH / "airflow-e2e-tests" / "docker" / "go.yml"

@@ -19,6 +19,7 @@
 import time
 from datetime import UTC, datetime
 from functools import cached_property
+from http import HTTPStatus
 
 import boto3
 import requests
@@ -143,11 +144,7 @@ def trigger_dag(self, dag_id: str, json=None):
     def wait_for_dag_run(self, dag_id: str, run_id: str, timeout=300, check_interval=5):
         start_time = time.time()
         while time.time() - start_time < timeout:
-            response = self._make_request(
-                method="GET",
-                endpoint=f"dags/{dag_id}/dagRuns/{run_id}",
-            )
-            state = response.get("state")
+            state = self.get_dag_run(dag_id=dag_id, run_id=run_id).get("state")
             if state in {"success", "failed"}:
                 return state
             time.sleep(check_interval)
@@ -170,6 +167,29 @@ def get_task_state_store(self, dag_id: str, run_id: str, task_id: str, key: str)
             endpoint=f"dags/{dag_id}/dagRuns/{run_id}/taskInstances/{task_id}/state-store/{key}",
         )
 
+    def set_variable(self, key: str, value: str, description: str | None = None):
+        """Create or replace an Airflow Variable via API."""
+        body = {"key": key, "value": value, "description": description}
+        try:
+            return self._make_request(method="POST", endpoint="variables", json=body)
+        except requests.HTTPError as exc:
+            # 409 == it already exists, from an earlier run of the same suite.
+            if exc.response is None or exc.response.status_code != HTTPStatus.CONFLICT:
+                raise
+            return self._make_request(method="PATCH", endpoint=f"variables/{key}", json=body)
+
+    def get_tasks(self, dag_id: str):
+        """List a Dag's tasks, with the edges each one carries."""
+        return self._make_request(method="GET", endpoint=f"dags/{dag_id}/tasks")
+
+    def get_dag_source(self, dag_id: str):
+        """Get the source code stored for a Dag's latest version."""
+        return self._make_request(method="GET", endpoint=f"dagSources/{dag_id}")
+
+    def get_dag_run(self, dag_id: str, run_id: str):
+        """Get a Dag run, with its state, run type and conf."""
+        return self._make_request(method="GET", endpoint=f"dags/{dag_id}/dagRuns/{run_id}")
+
     def trigger_dag_and_wait(self, dag_id: str, json=None):
         """Trigger a DAG and wait for it to complete."""
         self.un_pause_dag(dag_id)