diff --git a/pathwaysutils/experimental/gke/jobset.py b/pathwaysutils/experimental/gke/jobset.py
index 0fc89c4..dfdec0c 100644
--- a/pathwaysutils/experimental/gke/jobset.py
+++ b/pathwaysutils/experimental/gke/jobset.py
@@ -15,8 +15,11 @@
 import json
 import logging
 import math
+import time
 from typing import Any, Mapping, Sequence
 from kubernetes import client
+from kubernetes import config as k8s_config
+import yaml
 
 # GKE sidecar containers restartPolicy compatibility placeholder.
 
@@ -179,6 +182,14 @@ def __init__(
           "targetReplicatedJobs": [PATHWAYS_HEAD_JOB_NAME],
       }
 
+  @property
+  def head_job_template(self) -> client.V1JobTemplateSpec:
+    return self._head_job_template
+
+  @property
+  def worker_job_template(self) -> client.V1JobTemplateSpec:
+    return self._worker_job_template
+
   def _build_head_job_template(
       self,
       pathways_dir: str,
@@ -696,29 +707,43 @@ def _compile_config(self) -> dict[str, Any]:
           self._worker_job_template
       )
 
-    replicated_jobs = [
-        {
-            "name": PATHWAYS_HEAD_JOB_NAME,
-            "replicas": 1,
-            "template": serialized_head,
-        },
-        {
-            "name": PATHWAYS_WORKER_JOB_NAME,
-            "replicas": self._worker_replicas,
-            "template": serialized_worker,
-        },
-    ]
+    head_job = {
+        "name": PATHWAYS_HEAD_JOB_NAME,
+        "replicas": 1,
+        "template": serialized_head,
+    }
+    worker_job = {
+        "name": PATHWAYS_WORKER_JOB_NAME,
+        "replicas": self._worker_replicas,
+        "template": serialized_worker,
+    }
+
+    coordinator = {
+        "replicatedJob": PATHWAYS_HEAD_JOB_NAME,
+    }
+
+    failure_policy: dict[str, Any] = {
+        "restartStrategy": "Recreate",
+    }
+    if self._max_restarts > 0:
+      failure_policy["maxRestarts"] = self._max_restarts
 
     jobset_config = {
-        "apiVersion": f"jobset.sigs.k8s.io/{self._jobset_api_version}",
+        "apiVersion": f"jobset.x-k8s.io/{self._jobset_api_version}",
         "kind": "JobSet",
         "metadata": {
             "name": self._name,
             "namespace": self._namespace,
         },
         "spec": {
-            "failurePolicy": {"maxRestarts": self._max_restarts},
-            "replicatedJobs": replicated_jobs,
+            "startupPolicy": {"startupPolicyOrder": "InOrder"},
+            "failurePolicy": failure_policy,
+            "network": {
+                "enableDNSHostnames": True,
+                "publishNotReadyAddresses": True,
+            },
+            "coordinator": coordinator,
+            "replicatedJobs": [head_job, worker_job],
         },
     }
     if self._labels:
@@ -733,3 +758,153 @@ def _compile_config(self) -> dict[str, Any]:
   def to_dict(self) -> dict[str, Any]:
     """Returns the JobSet configuration as a dictionary."""
     return self._compile_config()
+
+  def export_yaml(self, filepath: str) -> None:
+    """Exports the JobSet configuration to a YAML file."""
+    with open(filepath, "w") as f:
+      yaml.dump(self.to_dict(), f, default_flow_style=False)
+
+  @classmethod
+  def import_yaml(cls, filepath: str) -> "PathwaysJobSet":
+    """Imports a JobSet configuration from a YAML file."""
+    with open(filepath, "r") as f:
+      config = yaml.safe_load(f)
+
+    cls._validate_config(config)
+
+    instance = cls.__new__(cls)
+    instance._name = config["metadata"]["name"]
+    instance._namespace = config["metadata"].get("namespace", "default")
+    api_version_parts = config.get("apiVersion", "").split("/")
+    instance._jobset_api_version = (
+        api_version_parts[-1] if len(api_version_parts) > 1 else "v1alpha2"
+    )
+    instance._max_restarts = (
+        config["spec"].get("failurePolicy", {}).get("maxRestarts", 0)
+    )
+    instance._labels = config["metadata"].get("labels", {})
+    instance._annotations = config["metadata"].get("annotations", {})
+
+    # Extract replicated jobs and deserialize.
+    head_job_template: client.V1JobTemplateSpec | None = None
+    worker_job_template: client.V1JobTemplateSpec | None = None
+
+    with client.ApiClient() as api_client:
+      for job in config["spec"]["replicatedJobs"]:
+        if job["name"] == PATHWAYS_HEAD_JOB_NAME:
+          head_job_template = _deserialize_dict(
+              api_client, job["template"], client.V1JobTemplateSpec
+          )
+        elif job["name"] in ("worker", PATHWAYS_WORKER_JOB_NAME):
+          worker_job_template = _deserialize_dict(
+              api_client, job["template"], client.V1JobTemplateSpec
+          )
+          instance._worker_replicas = job["replicas"]
+
+    if head_job_template is None:
+      raise ValueError(f"Missing head job ({PATHWAYS_HEAD_JOB_NAME}) in config")
+    if worker_job_template is None:
+      raise ValueError(
+          f"Missing worker job ({PATHWAYS_WORKER_JOB_NAME}) in config"
+      )
+
+    instance._head_job_template = head_job_template
+    instance._worker_job_template = worker_job_template
+
+    instance._success_policy = config["spec"].get("successPolicy")
+    return instance
+
+  @classmethod
+  def _validate_config(cls, config: dict[str, Any]) -> None:
+    """Validates that the config is a valid Pathways JobSet."""
+    if config.get("kind") != "JobSet":
+      raise ValueError("Resource kind is not JobSet")
+    jobs = {
+        j["name"]: j for j in config.get("spec", {}).get("replicatedJobs", [])
+    }
+    if "head" not in jobs and PATHWAYS_HEAD_JOB_NAME not in jobs:
+      raise ValueError(
+          f"Missing head replicated job ('head' or '{PATHWAYS_HEAD_JOB_NAME}')"
+      )
+    if "worker" not in jobs and PATHWAYS_WORKER_JOB_NAME not in jobs:
+      raise ValueError(
+          "Missing worker replicated job ('worker' or"
+          f" '{PATHWAYS_WORKER_JOB_NAME}')"
+      )
+
+  def apply(
+      self, recreate: bool = False, field_manager: str = "pathwaysutils"
+  ) -> None:
+    """Applies the JobSet to the GKE cluster."""
+
+    try:
+      k8s_config.load_kube_config()
+    except Exception:  # pylint: disable=broad-except
+      try:
+        k8s_config.load_incluster_config()
+      except Exception as e:
+        raise RuntimeError("Failed to load Kubernetes configuration") from e
+
+    api = client.CustomObjectsApi()
+    group = "jobset.x-k8s.io"
+    version = self._jobset_api_version
+    plural = "jobsets"
+
+    exists = False
+    try:
+      api.get_namespaced_custom_object(
+          group, version, self._namespace, plural, self._name
+      )
+      exists = True
+    except client.rest.ApiException as e:
+      if e.status != 404:
+        raise
+
+    if exists:
+      if recreate:
+        _logger.info(
+            "JobSet %s already exists. Deleting it first...", self._name
+        )
+        api.delete_namespaced_custom_object(
+            group, version, self._namespace, plural, self._name
+        )
+
+        # Poll for deletion.
+        max_retries = 30
+        for i in range(max_retries):
+          try:
+            api.get_namespaced_custom_object(
+                group, version, self._namespace, plural, self._name
+            )
+            _logger.info(
+                "Waiting for JobSet %s to be deleted... (%d/%d)",
+                self._name,
+                i + 1,
+                max_retries,
+            )
+            time.sleep(2)
+          except client.rest.ApiException as e:
+            if e.status == 404:
+              _logger.info("JobSet %s deleted.", self._name)
+              break
+            raise
+        else:
+          raise RuntimeError(
+              f"Timeout waiting for JobSet {self._name} to be deleted"
+          )
+      else:
+        raise RuntimeError(
+            f"JobSet {self._name} already exists. Use recreate=True to"
+            " overwrite."
+        )
+
+    _logger.info("Creating JobSet %s...", self._name)
+    api.create_namespaced_custom_object(
+        group=group,
+        version=version,
+        namespace=self._namespace,
+        plural=plural,
+        body=self.to_dict(),
+        field_manager=field_manager,
+    )
+    _logger.info("JobSet %s created successfully.", self._name)
diff --git a/pathwaysutils/test/experimental/gke/jobset_test.py b/pathwaysutils/test/experimental/gke/jobset_test.py
index 9372a71..dc9d290 100644
--- a/pathwaysutils/test/experimental/gke/jobset_test.py
+++ b/pathwaysutils/test/experimental/gke/jobset_test.py
@@ -1,10 +1,39 @@
 import hashlib
 import itertools
+import os
 from typing import Any
+from unittest import mock
 from absl.testing import absltest
 from absl.testing import parameterized
 from kubernetes import client
 from pathwaysutils.experimental.gke import jobset
+import yaml
+
+
+def normalize_k8s_spec(spec: Any) -> Any:
+  if isinstance(spec, dict):
+    result = {}
+    for k, v in spec.items():
+      if k == "env" and isinstance(v, list):
+        result[k] = sorted(
+            [normalize_k8s_spec(x) for x in v], key=lambda x: x.get("name", "")
+        )
+      elif k == "ports" and isinstance(v, list):
+        result[k] = sorted(
+            [normalize_k8s_spec(x) for x in v],
+            key=lambda x: x.get("containerPort", 0),
+        )
+      elif k == "volumeMounts" and isinstance(v, list):
+        result[k] = sorted(
+            [normalize_k8s_spec(x) for x in v],
+            key=lambda x: x.get("mountPath", ""),
+        )
+      else:
+        result[k] = normalize_k8s_spec(v)
+    return result
+  elif isinstance(spec, list):
+    return [normalize_k8s_spec(x) for x in spec]
+  return spec
 
 
 class JobSetManifestHelper:
@@ -475,6 +504,40 @@ def test_add_gcsfuse_preserves_existing_metadata(self):
     self.assertEqual(pod_meta.get("annotations", {}).get("existing-pod-anno"), "value")
     self.assertEqual(pod_meta.get("annotations", {}).get("gke-gcsfuse/volumes"), "true")
 
+  def test_add_gcsfuse_preserves_existing_volumes(self):
+    pw_jobset = self._create_jobset(topology="2x2", num_slices=1)
+
+    # Pre-populate volumes in head and worker pod specs.
+    pw_jobset._head_job_template.spec.template.spec.volumes = [
+        client.V1Volume(name="preexisting-head-vol")
+    ]
+    pw_jobset._worker_job_template.spec.template.spec.volumes = [
+        client.V1Volume(name="preexisting-worker-vol")
+    ]
+
+    pw_jobset.add_gcsfuse(
+        containers="all",
+        mount_path="/gcs/data",
+        bucket="my-bucket",
+    )
+    helper = JobSetManifestHelper(pw_jobset.to_dict())
+
+    # Verify existing volumes and newly added GCSFuse volumes coexist.
+    self.assertIn("preexisting-head-vol", helper.volumes["pathways-head"])
+    self.assertIn("preexisting-worker-vol", helper.volumes["pathways-worker"])
+
+  def test_add_colocated_python_handles_none_volumes(self):
+    pw_jobset = self._create_jobset(topology="2x2", num_slices=1)
+    
+    # Force volumes to be None
+    pw_jobset._worker_job_template.spec.template.spec.volumes = None
+
+    # Should not crash and should correctly add volume
+    pw_jobset.add_colocated_python(image="gcr.io/my-project/colocated-python:custom")
+    helper = JobSetManifestHelper(pw_jobset.to_dict())
+    
+    self.assertIn("shared-memory", helper.volumes["pathways-worker"])
+
   def test_add_colocated_python_sidecar(self):
     pw_jobset = self._create_jobset(topology="2x2", num_slices=1)
 
@@ -620,5 +683,244 @@ def test_colocated_python_with_jax_command(self):
 
     self.assertIn("pathways-worker", helper.jobs)
     self.assertIn("colocated-python-sidecar", helper.init_containers["pathways-worker"])
+
+  # Reference similar GKE / JobSet custom object unit test suites:
+  # - Google3 GKE JobSet test suite: //depot/google3/cloud/ai/map/catmint/supervisor/client/python/orchestrators/gke_callbacks_test.py
+  # - Upstream Kubernetes SIGs JobSet unit tests: https://github.com/kubernetes-sigs/jobset/blob/main/pkg/controllers/jobset_controller_test.go
+  @mock.patch("kubernetes.config.load_kube_config")
+  @mock.patch("kubernetes.config.load_incluster_config")
+  @mock.patch("kubernetes.client.CustomObjectsApi")
+  def test_apply_create(
+      self, mock_custom_objects_api, mock_load_incluster, mock_load_kube
+  ):
+    """Tests deploying a JobSet to GKE via Kubernetes CustomObjectsApi.
+
+    Modeled after official GKE JobSet unit test patterns (e.g. gke_callbacks_test.py).
+    """
+    mock_api = mock_custom_objects_api.return_value
+    # Mock GET to return 404 (not exists).
+    from kubernetes.client.rest import ApiException
+
+    mock_api.get_namespaced_custom_object.side_effect = ApiException(status=404)
+
+    pw_jobset = jobset.PathwaysJobSet(
+        name="test-workload",
+        namespace="default",
+        pathways_dir="gs://bucket/scratch",
+        tpu_type="v5e",
+        topology="2x2",
+        num_slices=1,
+    )
+    pw_jobset.apply()
+
+    mock_api.create_namespaced_custom_object.assert_called_once_with(
+        group="jobset.x-k8s.io",
+        version="v1alpha2",
+        namespace="default",
+        plural="jobsets",
+        body=pw_jobset.to_dict(),
+        field_manager="pathwaysutils",
+    )
+
+  @mock.patch("kubernetes.config.load_kube_config")
+  @mock.patch("kubernetes.config.load_incluster_config")
+  @mock.patch("kubernetes.client.CustomObjectsApi")
+  def test_apply_exists_recreate(
+      self, mock_custom_objects_api, mock_load_incluster, mock_load_kube
+  ):
+    mock_api = mock_custom_objects_api.return_value
+    # Mock GET to return success (exists).
+    mock_api.get_namespaced_custom_object.return_value = {}
+    # Mock GET after delete to return 404.
+    from kubernetes.client.rest import ApiException
+
+    mock_api.get_namespaced_custom_object.side_effect = [
+        {},
+        ApiException(status=404),
+    ]
+
+    pw_jobset = jobset.PathwaysJobSet(
+        name="test-workload",
+        namespace="default",
+        pathways_dir="gs://bucket/scratch",
+        tpu_type="v5e",
+        topology="2x2",
+        num_slices=1,
+    )
+    pw_jobset.apply(recreate=True)
+
+    mock_api.delete_namespaced_custom_object.assert_called_once_with(
+        "jobset.x-k8s.io", "v1alpha2", "default", "jobsets", "test-workload"
+    )
+    mock_api.create_namespaced_custom_object.assert_called_once_with(
+        group="jobset.x-k8s.io",
+        version="v1alpha2",
+        namespace="default",
+        plural="jobsets",
+        body=pw_jobset.to_dict(),
+        field_manager="pathwaysutils",
+    )
+
+  @mock.patch("kubernetes.config.load_kube_config")
+  @mock.patch("kubernetes.config.load_incluster_config")
+  @mock.patch("kubernetes.client.CustomObjectsApi")
+  def test_apply_exists_no_recreate_fails(
+      self, mock_custom_objects_api, mock_load_incluster, mock_load_kube
+  ):
+    mock_api = mock_custom_objects_api.return_value
+    # Mock GET to return success (exists).
+    mock_api.get_namespaced_custom_object.return_value = {}
+
+    pw_jobset = jobset.PathwaysJobSet(
+        name="test-workload",
+        namespace="default",
+        pathways_dir="gs://bucket/scratch",
+        tpu_type="v5e",
+        topology="2x2",
+        num_slices=1,
+    )
+    with self.assertRaises(RuntimeError):
+      pw_jobset.apply(recreate=False)
+
+  def test_export_import_roundtrip(self):
+    pw_jobset = jobset.PathwaysJobSet(
+        name="test-workload",
+        namespace="default",
+        pathways_dir="gs://bucket/scratch",
+        tpu_type="v5e",
+        topology="2x2",
+        num_slices=1,
+    )
+    pw_jobset.add_colocated_python(image="gcr.io/my-project/colocated-python:custom")
+    pw_jobset.add_gcsfuse(
+        containers="pathways-worker", mount_path="/tmp/gcs", bucket="my-bucket"
+    )
+
+    # Export.
+    temp_filepath = os.path.join(self.create_tempdir().full_path, "jobset.yaml")
+    pw_jobset.export_yaml(temp_filepath)
+
+    # Import.
+    imported_jobset = jobset.PathwaysJobSet.import_yaml(temp_filepath)
+
+    # Verify they are semantically identical.
+    self.assertEqual(
+        normalize_k8s_spec(pw_jobset.to_dict()),
+        normalize_k8s_spec(imported_jobset.to_dict()),
+    )
+
+  def test_import_validation_failures(self):
+    temp_dir = self.create_tempdir().full_path
+
+    # 1. Missing kind.
+    invalid_config1 = {
+        "apiVersion": "jobset.x-k8s.io/v1alpha2",
+        "metadata": {"name": "test"},
+        "spec": {"replicatedJobs": []},
+    }
+    path1 = os.path.join(temp_dir, "invalid1.yaml")
+    with open(path1, "w") as f:
+      yaml.dump(invalid_config1, f)
+    with self.assertRaisesRegex(ValueError, "Resource kind is not JobSet"):
+      jobset.PathwaysJobSet.import_yaml(path1)
+
+    # 2. Missing head.
+    invalid_config2 = {
+        "apiVersion": "jobset.x-k8s.io/v1alpha2",
+        "kind": "JobSet",
+        "metadata": {"name": "test"},
+        "spec": {
+            "replicatedJobs": [
+                {"name": "worker", "replicas": 1, "template": {}}
+            ]
+        },
+    }
+    path2 = os.path.join(temp_dir, "invalid2.yaml")
+    with open(path2, "w") as f:
+      yaml.dump(invalid_config2, f)
+    with self.assertRaisesRegex(ValueError, "Missing head replicated job"):
+      jobset.PathwaysJobSet.import_yaml(path2)
+
+    # 3. Missing worker.
+    invalid_config3 = {
+        "apiVersion": "jobset.x-k8s.io/v1alpha2",
+        "kind": "JobSet",
+        "metadata": {"name": "test"},
+        "spec": {
+            "replicatedJobs": [
+                {"name": "pathways-head", "replicas": 1, "template": {}}
+            ]
+        },
+    }
+    path3 = os.path.join(temp_dir, "invalid3.yaml")
+    with open(path3, "w") as f:
+      yaml.dump(invalid_config3, f)
+    with self.assertRaisesRegex(ValueError, "Missing worker replicated job"):
+      jobset.PathwaysJobSet.import_yaml(path3)
+
+  def test_labels_and_annotations(self):
+    pw_jobset = jobset.PathwaysJobSet(
+        name="test-workload",
+        namespace="default",
+        pathways_dir="gs://bucket/scratch",
+        tpu_type="v5e",
+        topology="2x2",
+        num_slices=1,
+        labels={"key1": "val1"},
+        annotations={"key2": "val2"},
+    )
+    config = pw_jobset.to_dict()
+    self.assertEqual(config["metadata"]["labels"], {"key1": "val1"})
+    self.assertEqual(config["metadata"]["annotations"], {"key2": "val2"})
+
+  def test_direct_mutation(self):
+    """Verifies that directly mutating Kubernetes JobTemplateSpec objects is preserved in to_dict()."""
+    pw_jobset = jobset.PathwaysJobSet(
+        name="test-workload",
+        namespace="default",
+        pathways_dir="gs://bucket/scratch",
+        tpu_type="v5e",
+        topology="2x2",
+        num_slices=1,
+    )
+    # Directly modify the Kubernetes template spec objects on the instance.
+    head_spec = pw_jobset.head_job_template.spec.template.spec
+    head_spec.active_deadline_seconds = 100
+
+    worker_spec = pw_jobset.worker_job_template.spec.template.spec
+    worker_spec.active_deadline_seconds = 200
+
+    # Verify that the direct mutations are preserved and serialized in to_dict().
+    config = pw_jobset.to_dict()
+    replicated_jobs = {
+        job["name"]: job for job in config["spec"]["replicatedJobs"]
+    }
+    self.assertEqual(
+        replicated_jobs["pathways-head"]["template"]["spec"]["template"]["spec"][
+            "activeDeadlineSeconds"
+        ],
+        100,
+    )
+    self.assertEqual(
+        replicated_jobs["pathways-worker"]["template"]["spec"]["template"]["spec"][
+            "activeDeadlineSeconds"
+        ],
+        200,
+    )
+
+  def test_failure_policy(self):
+    pw_jobset = jobset.PathwaysJobSet(
+        name="test-workload",
+        namespace="default",
+        pathways_dir="gs://bucket/scratch",
+        tpu_type="v5e",
+        topology="2x2",
+        num_slices=1,
+        max_restarts=5,
+    )
+    config = pw_jobset.to_dict()
+    self.assertEqual(config["spec"]["failurePolicy"]["maxRestarts"], 5)
+
+
 if __name__ == "__main__":
   absltest.main()
diff --git a/pathwaysutils/test/experimental/gke/testdata/model_lite_tpuv5e_4x8.yaml b/pathwaysutils/test/experimental/gke/testdata/model_lite_tpuv5e_4x8.yaml
new file mode 100644
index 0000000..ad72f17
--- /dev/null
+++ b/pathwaysutils/test/experimental/gke/testdata/model_lite_tpuv5e_4x8.yaml
@@ -0,0 +1,228 @@
+apiVersion: jobset.x-k8s.io/v1alpha2
+kind: JobSet
+metadata:
+  labels:
+    kueue.x-k8s.io/queue-name: multislice-queue
+  name: lukebaumann-model-v5e
+  namespace: default
+spec:
+  coordinator:
+    replicatedJob: pathways-head
+  failurePolicy:
+    restartStrategy: Recreate
+  network:
+    enableDNSHostnames: true
+    publishNotReadyAddresses: true
+  replicatedJobs:
+  - name: pathways-head
+    replicas: 1
+    template:
+      metadata:
+        annotations:
+          alpha.jobset.sigs.k8s.io/exclusive-topology: kubernetes.io/hostname
+      spec:
+        backoffLimit: 0
+        completionMode: Indexed
+        completions: 1
+        parallelism: 1
+        template:
+          metadata:
+            annotations:
+              alpha.jobset.sigs.k8s.io/exclusive-topology: kubernetes.io/hostname
+            labels:
+              kueue.x-k8s.io/podset: pathways-head
+          spec:
+            containers:
+            - command:
+              - sleep
+              - infinity
+              env:
+              - name: PATHWAYS_HEAD
+                valueFrom:
+                  fieldRef:
+                    fieldPath: metadata.labels['jobset.sigs.k8s.io/coordinator']
+              - name: JAX_PLATFORMS
+                value: proxy
+              - name: XCLOUD_ENVIRONMENT
+                value: GCP
+              # - name: TPU_STDERR_LOG_LEVEL
+              #   value: "0"
+              # - name: TPU_MIN_LOG_LEVEL
+              #   value: "0"
+              # - name: TF_CPP_MIN_LOG_LEVEL
+              #   value: "0"
+              - name: TPU_VMODULE
+                value: "real_program_continuator=1"
+              - name: ENABLE_PATHWAYS_PERSISTENCE
+                value: "1"
+              - name: ENABLE_PERSISTENCE_API
+                value: "1"
+              - name: ENABLE_PJRT_COMPATIBILITY
+                value: "true"
+              - name: JAX_BACKEND_TARGET
+                value: grpc://$(PATHWAYS_HEAD):29000
+              image: ubuntu:latest
+              imagePullPolicy: Always
+              name: jax-tpu
+              resources:
+                limits:
+                  cpu: "24"
+                  memory: 100G
+              securityContext:
+                privileged: true
+              volumeMounts:
+              - mountPath: /tmp
+                name: shared-tmp
+            dnsPolicy: ClusterFirstWithHostNet
+            hostNetwork: true
+            initContainers:
+            - args:
+              - --server_port=29001
+              - --gcs_scratch_location=gs://fake-bucket/scratch
+              - --node_type=resource_manager
+              - --instance_count=2
+              - --instance_type=tpuv5e:4x8
+              env:
+              - name: REPLICATED_JOB_NAME
+                valueFrom:
+                  fieldRef:
+                    fieldPath: metadata.annotations['jobset.sigs.k8s.io/replicatedjob-name']
+              - name: JOBSET_NAME
+                valueFrom:
+                  fieldRef:
+                    fieldPath: metadata.annotations['jobset.sigs.k8s.io/jobset-name']
+              - name: HOST_ADDRESS
+                valueFrom:
+                  fieldRef:
+                    fieldPath: metadata.labels['jobset.sigs.k8s.io/coordinator']
+              - name: TPU_SKIP_MDS_QUERY
+                value: "true"
+              image: "us-docker.pkg.dev/cloud-tpu-v2-images/pathways/server:latest"
+              imagePullPolicy: Always
+              name: pathways-rm
+              ports:
+              - containerPort: 29001
+                protocol: TCP
+              - containerPort: 29002
+                protocol: TCP
+              resources:
+                limits:
+                  cpu: "8"
+                  memory: 32G
+              restartPolicy: Always
+            - args:
+              - --server_port=29000
+              - --resource_manager_address=$(PATHWAYS_HEAD):29001
+              - --gcs_scratch_location=gs://fake-bucket/scratch
+              - --num_elastic_slices=2
+              env:
+              - name: PATHWAYS_HEAD
+                valueFrom:
+                  fieldRef:
+                    fieldPath: metadata.labels['jobset.sigs.k8s.io/coordinator']
+              image: "us-docker.pkg.dev/cloud-tpu-v2-images/pathways/proxy_server:latest"
+              imagePullPolicy: Always
+              name: pathways-proxy
+              ports:
+              - containerPort: 29000
+                protocol: TCP
+              resources:
+                limits:
+                  cpu: "16"
+                  memory: 100G
+              restartPolicy: Always
+            nodeSelector:
+              cloud.google.com/gke-nodepool: cpu-np
+            restartPolicy: Never
+            volumes:
+            - hostPath:
+                path: /tmp
+                type: DirectoryOrCreate
+              name: shared-tmp
+  - name: pathways-worker
+    replicas: 2 # number of slices
+    template:
+      metadata: {}
+      spec:
+        backoffLimit: 32000000
+        completionMode: Indexed
+        completions: 8 # number of vms per slice
+        parallelism: 8 # number of vms per slice
+        template:
+          metadata:
+            annotations:
+              alpha.jobset.sigs.k8s.io/exclusive-topology: cloud.google.com/gke-nodepool
+          spec:
+            containers:
+            - args:
+              - --resource_manager_address=$(PATHWAYS_HEAD):29001
+              - --server_port=29005
+              - --gcs_scratch_location=gs://fake-bucket/scratch
+              env:
+              - name: TPU_MIN_LOG_LEVEL
+                value: "0"
+              - name: TF_CPP_MIN_LOG_LEVEL
+                value: "0"
+              - name: XCLOUD_ENVIRONMENT
+                value: GCP
+              - name: MEGASCALE_GRPC_ENABLE_XOR_TRACER
+                value: "false"
+              - name: MEGASCALE_NUM_SLICES
+                valueFrom:
+                  fieldRef:
+                    fieldPath: metadata.labels['jobset.sigs.k8s.io/replicatedjob-replicas']
+              - name: JOBSET_NAME
+                valueFrom:
+                  fieldRef:
+                    fieldPath: metadata.annotations['jobset.sigs.k8s.io/jobset-name']
+              - name: REPLICATED_JOB_NAME
+                valueFrom:
+                  fieldRef:
+                    fieldPath: metadata.annotations['jobset.sigs.k8s.io/replicatedjob-name']
+              - name: MEGASCALE_SLICE_ID
+                valueFrom:
+                  fieldRef:
+                    fieldPath: metadata.labels['jobset.sigs.k8s.io/job-index']
+              - name: PATHWAYS_HEAD
+                valueFrom:
+                  fieldRef:
+                    fieldPath: metadata.labels['jobset.sigs.k8s.io/coordinator']
+              - name: MEGASCALE_COORDINATOR_ADDRESS
+                valueFrom:
+                  fieldRef:
+                    fieldPath: metadata.labels['jobset.sigs.k8s.io/coordinator']
+              image: "us-docker.pkg.dev/cloud-tpu-v2-images/pathways/server:latest"
+              imagePullPolicy: Always
+              name: pathways-worker
+              ports:
+              - containerPort: 29005
+                protocol: TCP
+              - containerPort: 29006
+                protocol: TCP
+              - containerPort: 8471
+                protocol: TCP
+              - containerPort: 8080
+                protocol: TCP
+              resources:
+                limits:
+                  google.com/tpu: "4"
+              volumeMounts:
+              - mountPath: /tmp
+                name: shared-tmp
+            dnsPolicy: ClusterFirstWithHostNet
+            hostNetwork: true
+            nodeSelector:
+              cloud.google.com/gke-tpu-accelerator: tpu-v5-lite-podslice
+              cloud.google.com/gke-tpu-topology: 4x8
+            restartPolicy: OnFailure
+            volumes:
+            - hostPath:
+                path: /tmp
+                type: DirectoryOrCreate
+              name: shared-tmp
+  startupPolicy:
+    startupPolicyOrder: InOrder
+  successPolicy:
+    operator: All
+    targetReplicatedJobs:
+    - pathways-head