Repository: kubernetes-sigs/kueue

Issue:
Introduce a library for mocking in unit tests and consolidate current mocks
<!-- Please only use this template for submitting clean up requests -->

**What would you like to be cleaned**:

Use a library, like uber-go/mock to consolidate the mocking for GenericJob:
- testGenericJob in validation_test.go and (almost identical copy) in base_webhook_test.go

This was first proposed in https://github.com/kubernetes-sigs/kueue/pull/5510#discussion_r2183633446 

**Why is this needed**:

To simplify the test code, and avoid duplication for writing mocks per-test.

Hints from issue comments:
@mimowo: cc @tenzen-y @ichekrygin @tenzen-y: Is it only for Unit Test, right?
@mimowo: Yes, updated title@tenzen-y: In terms of uber-go/mock vs testify, +1 on uber-go/mock since it is originally came from Go std library.@mbobrovskyi: /assign@ichekrygin: Hi @mbobrovskyi, while working on an unrelated PR I came across a need to use a mock for the GenericJob interface.
I’m not sure if you’ve made progress on this issue yet, but if not, I have a PR almost ready that adds this functionality, so I was wondering if you would be open to me taking over this issue in that case?
@ichekrygin: > Hi [@mbobrovskyi](https://github.com/mbobrovskyi), while working on an unrelated PR I came across a need to use a mock for the GenericJob interface.
> I’m not sure if you’ve made progress on this issue yet, but if not, I have a PR almost ready that adds this functionality, so I was wondering if you would be open to me taking over this issue in that case?

Please disregard. I had "outdated" view on the issue, and entirely missed your PR. I just closed mine.

Solution patch:
diff --git a/apis/kueue/v1beta1/constants.go b/apis/kueue/v1beta1/constants.go
index 42d9aae44..9ee27e1fa 100644
--- a/apis/kueue/v1beta1/constants.go
+++ b/apis/kueue/v1beta1/constants.go
@@ -19,6 +19,12 @@ package v1beta1
 const (
 	ResourceInUseFinalizerName                 = "kueue.x-k8s.io/resource-in-use"
 	DefaultPodSetName          PodSetReference = "main"
+
+	// ElasticJobSchedulingGate is the name of the scheduling gate applied to Pods
+	// to delay their scheduling until the associated workload slice has been admitted.
+	// This gate ensures that Pods do not begin scheduling prematurely, maintaining
+	// proper sequencing in workload processing.
+	ElasticJobSchedulingGate = "kueue.x-k8s.io/elastic-job"
 )
 
 type StopPolicy string
diff --git a/apis/kueue/v1beta1/workload_types.go b/apis/kueue/v1beta1/workload_types.go
index 0aadbeb3f..0255d53ab 100644
--- a/apis/kueue/v1beta1/workload_types.go
+++ b/apis/kueue/v1beta1/workload_types.go
@@ -645,6 +645,10 @@ const (
 	// due to non-recoverable node failures.
 	WorkloadEvictedDueToNodeFailures = "NodeFailures"
 
+	// WorkloadSliceReplaced indicates that the workload instance was
+	// replaced with a new workload slice.
+	WorkloadSliceReplaced = "WorkloadSliceReplaced"
+
 	// WorkloadDeactivated indicates that the workload was evicted
 	// because spec.active is set to false.
 	WorkloadDeactivated = "Deactivated"
diff --git a/keps/77-dynamically-sized-jobs/README.md b/keps/77-dynamically-sized-jobs/README.md
index 0d131a539..760ccde8d 100644
--- a/keps/77-dynamically-sized-jobs/README.md
+++ b/keps/77-dynamically-sized-jobs/README.md
@@ -7,168 +7,331 @@
   - [Non-Goals](#non-goals)
 - [Proposal](#proposal)
   - [User Stories](#user-stories)
-    - [Story 1 - RayCluster w/ autoscaling](#story-1---raycluster-w-autoscaling)
+    - [Story 1 - batchv1/Job scale-down](#story-1---batchv1job-scale-down)
+    - [Story 2 - batchv1/Job scale-up](#story-2---batchv1job-scale-up)
+    - [Story 3 – batch/v1.Job in Multi-Cluster Configuration](#story-3--batchv1job-in-multi-cluster-configuration)
   - [Notes/Constraints/Caveats (Optional)](#notesconstraintscaveats-optional)
 - [Design Details](#design-details)
-  - [Workload Slices](#workload-slices)
-  - [Creating Workload Slices](#creating-workload-slices)
+  - [Enablement](#enablement)
+    - [Features](#features)
+    - [WorkloadSliceAnnotation](#workloadsliceannotation)
+  - [Processing of Workload Slices](#processing-of-workload-slices)
+    - [Creation](#creation)
+      - [Naming](#naming)
+    - [Scheduling and Preemption](#scheduling-and-preemption)
+      - [Preemption Disambiguation.](#preemption-disambiguation)
+      - [Resource Flavors](#resource-flavors)
+    - [Garbage Collection of Preempted Workload Slices](#garbage-collection-of-preempted-workload-slices)
   - [Pod Scheduling Gates](#pod-scheduling-gates)
-  - [Garbage Collecting Workload Slices](#garbage-collecting-workload-slices)
+  - [Limitations and Incompatibilities](#limitations-and-incompatibilities)
+    - [PartialAdmission](#partialadmission)
 - [Phases for MVP (alpha)](#phases-for-mvp-alpha)
-  - [Phase 1 - Scale Down](#phase-1---scale-down)
-    - [Job controller](#job-controller)
-  - [Phase 2 - Aggregating Workload Slices](#phase-2---aggregating-workload-slices)
-  - [Phase 3 - Scale up with Workload Slices and Scheduling Gates](#phase-3---scale-up-with-workload-slices-and-scheduling-gates)
-    - [Scheduler](#scheduler)
+  - [Phase 1 - batchv1/Job WorkloadSlices Support in Single-Cluster Configuration.](#phase-1---batchv1job-workloadslices-support-in-single-cluster-configuration)
+    - [Scale Down](#scale-down)
+    - [Scale Up](#scale-up)
+  - [Phase 2 – RayCluster WorkloadSlice Support in Single-Cluster Configuration](#phase-2--raycluster-workloadslice-support-in-single-cluster-configuration)
+  - [Phase 3 – Enabling Workload Slicing for batch/v1.Job in Multi-Cluster Configuration](#phase-3--enabling-workload-slicing-for-batchv1job-in-multi-cluster-configuration)
 - [Additional Details](#additional-details)
   - [Test Plan](#test-plan)
     - [Unit Tests](#unit-tests)
     - [Integration tests](#integration-tests)
+    - [Scale-down](#scale-down-1)
+    - [Scale-Down](#scale-down-2)
+    - [Resource Flavor Handling](#resource-flavor-handling)
   - [Graduation Criteria](#graduation-criteria)
-- [Implementation History](#implementation-history)
-- [Drawbacks](#drawbacks)
+  - [PodScheduling Readiness and ResourceQuota](#podscheduling-readiness-and-resourcequota)
 - [Alternatives](#alternatives)
-  - [Ignore Resize from Kuberay](#ignore-resize-from-kuberay)
 <!-- /toc -->
 
 ## Summary
 
-Enable dynamic sizing of Kueue jobs. For the MVP, we will only focus on supporting RayClusters with autoscaling enabled, but other resources that can benefit from dynamic sizing should be supported eventually.
+Enable dynamic resizing of Kueue-managed workloads by supporting in-place horizontal scaling of job parallelism.
+The primary goal is to allow dynamic adjustment of job parallelism without requiring suspension and requeueing, thereby improving scheduling efficiency and minimizing disruption during resource fluctuations.
 
-See: [Support dynamically sized (elastic) jobs #77](https://github.com/kubernetes-sigs/kueue/issues/77)
+In addition, we aim to extend dynamic resizing support to RayCluster workloads. 
+While the initial focus is on batch/v1.Job and RayCluster, the long-term objective is to generalize support for dynamically sized (elastic) workloads across other Kueue-integrated resource types.
 
 ## Motivation
 
-Kueue currently lacks support for resizing jobs. When a job is resized, Kueue will recreate the Workload representation of the job, leading to a disruptive suspend and requeue process. This limitation hinders the usability of Kueue for resources like RayCluster that sometimes have autoscaling capabilities enabled.
+Kueue currently lacks native support for resizing jobs. Any change in job size leads to the recreation of the associated Workload, resulting in job suspension and requeueing. 
+This disrupts execution and hinders usability for elastic workloads like RayCluster, which rely on in-place autoscaling. 
 
-To properly support RayCluster, Kueue needs to gracefully handle the scale up and scale down of RayCluster nodes. Concretely this means that scaling the resources used by a RayCluster is appropriately reflected in the respective ClusterQueue without needing to suspend the entire RayCluster.
+To support such scenarios, Kueue must gracefully handle horizontal scale-up and scale-down operations without disrupting admitted jobs or re-acquiring quota.
 
 ### Goals
 
-- Gracefully handle resize operations for Kueue jobs (i.e. update quota usage without suspend, enqueue scale ups)
-- Autoscaling RayCluster works with Kueue since it has an autoscaler and there is high demand (MVP)s
-
+- Gracefully handle resize operations for Kueue-managed jobs by updating quota usage without suspending the job or terminating running pods.
+  
 ### Non-Goals
 
-- Vertical scaling of workloads – Kueue will only handle resize operations that scale Pods horizontally 
-- Support resize for other Kueue jobs such as Job, JobSet, etc (future)
-- Resizing of RayJobs
-- Partial Preemption
+- Vertical scaling of workloads – Kueue will only handle resize operations that scale Pods horizontally.
+- Orchestration or initiation of resizing for the parent workload objects (e.g., `batch/v1.Job`, `RayJob`, etc.)
+- Partial Preemption.
 
 ## Proposal
 
 Update the Job framework reconciler and introduce new controllers to orchestrate dynamic resizing of jobs. We are only interested in horizontal scaling of jobs (e.g. scaling more replicas). At a high level, this will be accomplished by:
-- Creating Workload Slice objects that represent incremental scale up of jobs. This gives us per-replica control of workload admission. Workload Slices will be garbage collected and consolidated with their parent workloads after successful admission.
-- Adding default scheduling gates to control the scheduling of new pods based on their admission status.
-- Dynamically adjust quotas in ClusterQueue based on scaling events.
+- Creating WorkloadSlice objects that represent incremental scale up of jobs. This gives us per-replica control of workload admission. Workload Slices will be garbage collected and consolidated with their parent workloads after successful admission.
+  - Adding default scheduling gates to control the scheduling of new pods based on their admission status.
+  - Dynamically adjust quotas in ClusterQueue based on scaling events.
 
-For the MVP, we will only focus on admission of **RayClusters** with autoscaling enabled.
+For the MVP, dynamic resizing will be available through an opt-in mechanism for selected, supported Kueue frameworks—starting with batch/v1.Job and, optionally, RayCluster workloads with autoscaling enabled. 
+Additionally, WorkloadSlice will be supported in the multiKueue context as part of the MVP scope.
 
 ### User Stories
 
-#### Story 1 - RayCluster w/ autoscaling
+#### Story 1 - batchv1/Job scale-down
+
+1. The user creates a `batch/v1.Job` with workload-slice enablement explicitly opted in.
+2. Kueue admits the Job based on its requested resources, creating the corresponding Workload and Job's Pods. The Pods begin running as usual.
+3. The user updates the Job by reducing its parallelism, triggering a scale-down event:
+   1. The associated Workload is updated to reflect the new, lower pod count in its spec. 
+   2. The local queue’s capacity is updated accordingly to reflect the reduced resource usage. 
+   3. One running Job pod is terminated, while the remaining Pods continue running uninterrupted.
+
+#### Story 2 - batchv1/Job scale-up
+1. The user creates a `batch/v1.Job` with workload-slice enablement explicitly opted in.
+2. Kueue admits the Job based on its requested resources, creating the corresponding Workload and Job’s Pods. The Pods begin running as usual.
+3. The user updates the Job by increasing its parallelism, triggering a scale-up event:
+   1. A new WorkloadSlice is created to represent the additional pod capacity requested.
+   2. The new Pods are created in a gated state (via PodSchedulingGates) and held from scheduling until the slice is admitted. 
+   3. Kueue evaluates available resources; if sufficient, the slice is admitted and the scheduling gates on the new Pods are removed. 
+   4. The newly admitted Pods begin scheduling and running, resulting in the Job operating with the increased parallelism level.
 
-1. The user creates a RayCluster with `enableInTreeAutoscaling: true`.
-2. Kueue admits the RayCluster based on the requested resources in the head pod and worker pod. 
-3. Given the demand of the job, the Ray Autoscaler adjusts the replicas field as it adds or removes Pods from the cluster.
-4. Kueue will not suspend and requeue the RayCluster, instead it will dynamically update ClusterQueue usage based on the scaling event. 
+#### Story 3 – batch/v1.Job in Multi-Cluster Configuration
+
+This story mirrors the behaviors in Stories 1 and 2 but applies them within a multi-cluster (MultiKueue) setup, 
+where WorkloadSlices are propagated, admitted, and scheduled across multiple clusters.
 
 ### Notes/Constraints/Caveats (Optional)
 
-If Kueue needs to preempt the resized RayCluster, it would preempt it as a whole, regardless of whether the RayCluster was previously scaled up.
+If Kueue needs to preempt a resized Job, it will preempt the entire Job as a single unit, regardless of whether the Job has undergone a scale-up operation.
 
 ## Design Details
 
-### Workload Slices
-To support horizontal scaling of jobs, we will introduce the concept of a "Workload Slice”. A Workload Slice is a Workload object with an owner reference to the original Workload for a job. Workload Slices represent per-replica changes to a job that were not initially accounted for when the job was created. **It's an internal state, not an extra CRD.**
+To support horizontal scaling of Jobs, we will introduce the concept of a "WorkloadSlice". A WorkloadSlice is a Workload object with an owner reference to the original Workload for a Job. 
+Workload Slices represent per-replica changes to a Job that were not initially accounted for when the Job was created. **It's an internal state, not an extra CRD.**
 
-The benefit of Workload Slices is that Kueue can evaluate admission on a per-replica basis without changing the existing semantics of the Workload API. Once a Workload Slice is admitted, it will be garbage collected and its resources will be aggregated into the admission status of the parent workload.
-- Workload Slices will be submitted to the same LocalQueue that's referenced by the top-level Workload. 
+The benefit of Workload Slices is that Kueue can evaluate admission on a per-replica basis without changing the existing semantics of the Workload API. 
+Once a WorkloadSlice is admitted, it will replace the original/old WorkloadSlice. 
+The replaced slice will be marked as finished.
+- Each WorkloadSlice inherits the resource flavor assignment of the original admitted slice, enforcing flavor consistency across scale-ups. 
 - In MultiKueue, Workload Slices would go into both clusters (management and workload) in a multi-cluster environment.
-- Workload Slices will be created by Kueue and use identical PodTemplates (which is already enforced by Kuberay in the case for RayCluster).
-- Workload Slices will belong to the same resource flavor as the top-level Workload that was initially admitted.
+- Workload Slices will be created by Kueue and use identical PodTemplate.
+- Workload Slices will belong to the same resource flavor as the top-level Workload that was initially admitted (for more details see "Resource Flavors" subsection)
+
 
-The parent Workload should have a condition that reflects the scaling progression status. 
+### Enablement
+
+WorkloadSlices in Kueue are enabled through a combination of a Kubernetes feature gate and an opt-in annotation on individual Workload objects. 
+At the cluster level, the WorkloadSlices feature must be enabled via the corresponding Kueue feature gate, which controls whether the controller logic for slicing is active. 
+
+Once the feature gate is enabled, individual Workload objects can opt into slicing by including the `kueue.x-k8s.io/elastic-job: "true"` annotation. 
+When both conditions are met, Kueue treats the Workload as eligible for partitioning into one or more WorkloadSlice objects, enabling fine-grained scheduling and 
+execution across multiple clusters or within a single cluster. If the feature gate is disabled or the annotation is omitted (or set to "false"), 
+the system defaults to the traditional single-Workload scheduling path for full backward compatibility.
+
+#### Features
+```golang
+// owner: @ichekrygin
+// kep: https://github.com/kubernetes-sigs/kueue/tree/main/keps/77-dynamically-sized-jobs
+//
+// ElasticJobsViaWorkloadSlices enables support for horizontal-scale in jobs.
+ElasticJobsViaWorkloadSlices featuregate.Feature = "ElasticJobsViaWorkloadSlices"
+```
 
+#### WorkloadSliceAnnotation
 ```golang
+// Workload slicing can be enabled for a specific integration job instance,
+// provided that the integration supports workload slicing.
 const (
-  ...
-  ...
-	// WorkloadResizeRequested means that the Workload is in the process of scaling up or down
-	WorkloadResizeRequested = "ResizeRequested"
+  // EnabledAnnotationKey refers to the annotation key present on Job's that support
+  // workload slicing.
+  // This annotation is alpha-level.
+  EnabledAnnotationKey = "kueue.x-k8s.io/elastic-job"
 )
-
 ```
 
-### Creating Workload Slices
+### Processing of Workload Slices
+
+#### Creation
+The creation of WorkloadSlices in Kueue is initiated when a Kueue Job with slicing enabled, via both annotation and feature gate, is admitted for execution. 
+Upon admission, the `jobframework.JobReconciler` evaluates the Job’s PodSets and resource requirements, then creates a new or additional (slice) Workload object 
+that represents a partition of the original workload. Each slice contains a `PodSet.Count` identical to the Job’s definition, along with scheduling metadata needed for handling preemption of the old workload (slice).
+
+At most two active workload slices can exist for any single Job in Kueue: the current admitted slice and a newly created slice during a scaling operation. 
+This ensures a controlled transition between slices while maintaining consistency and avoiding resource contention and race conditions.
 
-The [GenericJob interface (7e778f5)](https://github.com/kubernetes-sigs/kueue/blob/main/pkg/controller/jobframework/interface.go#L30-L55) will be updated to handle resize operations of jobs.
+To facilitate precise resource scheduling and effective workload preemption, every new job configured with slice enablement will be appropriately annotated using `EnabledAnnotationKey/Value` mentioned above.
+Furthermore, each individual workload slice will record the identity of any preempting slice it encounters.
 
 ```golang
-type GenericJob interface {
-  ...
-  ...
-  ResizeJob(wl *kueue.Workload) error
-}
+
+const (
+    // WorkloadSliceReplacementFor is the annotation key used to capture an "old" workload slice key
+    // that will be preempted by the "new", e.g., this workload slice with annotation.
+    WorkloadSliceReplacementFor = "kueue.x-k8s.io/workload-slice-replacement-for"
+)
+```
+
+##### Naming
+Currently, Kueue uses a deterministic Workload naming strategy by generating consistent names for the same Job instance via `GetWorkloadNameForOwnerWithGVK`.
+As a result, Workload names are not unique across different revisions of the same owner object (e.g., a Job), which makes this approach unsuitable for representing Workload slices.
+
+To address this, a new naming strategy will be introduced specifically for Workload slices, incorporating the owner object's generation value to ensure name uniqueness across revisions.
+For example, `bachv1/Job`: 
+```yaml
+annotations:
+  kueue.x-k8s.io/elastic-job: "true"
+creationTimestamp: "2025-06-12T21:32:41Z"
+generation: 1
+labels:
+  kueue.x-k8s.io/queue-name: demo
+name: demo-slice
+namespace: demo
+resourceVersion: "190175"
+uid: 78d1e61a-c399-4c09-800f-a0bbfb5fe71e
+```
+
+will result in a Workload name similar to: `job-demo-slice-75e0c`
+Whereas, changing `generation: 2`: will result in a different name similar to: `job-demo-slice-e7080`
+
+#### Scheduling and Preemption
+Kueue’s scheduler and preemptor components will be augmented to support workload slice processing, specifically by enforcing 
+preemption of the old workload slice regardless of queue capacity to make room for the new slice. 
+
+Scheduling assignment is based on pod count capacity, accounting for the existing workload slice's allocated pods. 
+For example, scaling a job up from `3` pods to `10` pods results in the creation of a new Workload representing all `10` pods; 
+however, the scheduling assignment will reflect only the `7` additional pods needed to admit the new slice, since `3` pods are already accounted for by the existing slice.
+
+While preemption and admission are not strictly atomic operations, they occur within the same scheduling cycle, specifically within the same closure of workload processing. Barring transient errors, for every candidate workload, preemption is immediately followed by admission. We could think of this as quasi-atomic behavior from a scheduling perspective.
+
+What’s important to emphasize is that while the workload object is preempted, the pods associated with the old workload are not. More importantly, those existing pods are effectively absorbed into the new workload.
+
+In the example at hand, preempting an old workload with 3 running pods and admitting a new workload requesting 10 pods (7 of which are pending and schedule-gated), the new workload will effectively consist of 3 running pods and 7 un-gated pods. This transition avoids disruption while maintaining quota integrity.
+
+##### Preemption Disambiguation.
+
+Here, we’re using the term “preemption” to describe the process of replacing an old workload slice with a new one. 
+At the same time, Kueue already uses the concept of workload preemption to mean evicting a workload and deferring its scheduling until capacity becomes available.
+
+In the context of workload slices, it may be more helpful to think of this replacement process as **aggregation** rather than preemption. 
+Aggregation involves terminating the old slice and activating the new one. In the strict sense, neither eviction nor preemption (as traditionally defined) actually occurs during this process.
+
+As such, we mark the old slice with the `Finished` condition. This ensures it is removed from the in-memory ClusterQueue snapshot, and its resource usage is released and attributed to the new slice that replaces it.
+
+A Workload that is preempted by a WorkloadSlice will be correctly marked with a Finished status condition, including an appropriate reason and message.
+```yaml
+- type: Finished
+  status: True
+  reason: WorkloadSliceReplaced
+  message: 'Removed to accommodate a workload (UID: a112b551-a595-4356-875d-e075aee52bac, JobUID: 88d348ca-317a-47a0-bd92-786f95f2608e) due to workload slice aggregation'
+  observedGeneration: 1
+  lastTransitionTime: "2025-07-08T20:09:24Z"
 ```
-Jobs implementing the ResizeJob method will create a Workload Slice for every new replica of a job. 
 
-On scale down to M we will change the original Workload's resources and then on scale up to N we will create N-M WorkloadSlice objects that go through admission and scheduling gate checks, i.e. the Workload can "lose" the original (first time admission) quota when scaling down before scaling up.
+##### Resource Flavors
+
+In the context of WorkloadSlice scheduling, when a workload qualifies for multiple resource flavors, i.e., more than one flavor can satisfy the resource requirements and selector constraints, the scheduler must determine which flavor to assign. 
+Kueue currently does not support combining resources from multiple flavors within a single PodSet; however, multiple PodSets within the same workload may be admitted using different resource flavors.
+
+Flavor selection is influenced by factors such as quota availability, taint tolerations, node label selectors, and, where applicable, topology-aware scheduling policies.
+
+When scaling up a previously admitted workload slice, the current implementation enforces a sticky flavor constraint, meaning that the new slice must reuse the originally assigned flavor, even if other eligible flavors have available capacity. 
+This approach avoids unpredictable flavor switching but may lead to the new slice remaining in a Pending state if the original flavor lacks sufficient resources.
+
+Future enhancements may introduce support for controlled flavor reassignment or user-defined policies to express preferences or tolerances for such transitions.
+
+#### Garbage Collection of Preempted Workload Slices
+Currently, a Workload can become outdated, marked by Kueue as "out-of-sync", and subsequently transitioned to the "Finished" state.
+This also applies to outdated Workload slices: when multiple successive updates are issued for a given Job, any superseded slices are marked as "Finished" to reflect their obsolescence.
+
+Preempted (i.e., aggregated) Workload slices are marked as Finished.
+Under the current design proposal, all finished workload slices are retained indefinitely and are not garbage collected.
+To address potential resource buildup, a `PreemptedWorkloadSliceHistory` mechanism whether as a configuration option or an API field on the Workload, could be introduced to limit the number of retained inactive slices, 
+similar to how `revisionHistoryLimit` is used in Kubernetes Deployments to manage ReplicaSet history.
+
+Preempted Workload slices that fail admission requirements are not marked as Finished, as they were never actually started (i.e., never admitted or executed). 
+
+Similar to current functionality, all Workloads associated with a Job (including slices) will be deleted when the parent Job is removed.
 
 ### Pod Scheduling Gates
 
-Inside the job's webhook, implement schedulingGate injection for pods on creation time.
-The Pods will be ungated following a similar behavior as to how a job is suspended and then unsuspended in the when admitted.
-When the job scales up, the new pods will be gated due to the schedulingGates injection in the webhook.
+Workload slices operate independently of the `spec.suspend` mechanism where applicable, relying instead on pod scheduling gates for admission control. 
+This approach is automatically enabled for supported Job types through defaulting. When a Workload (either initial or a slice) is created, all associated pods are initialized with scheduling gates, effectively "gated" from scheduling. 
+These gates are removed only upon the admission of the corresponding Workload, ensuring controlled and deliberate execution.
+
+```golang
+const (
+...
+  // WorkloadSliceSchedulingGate is the name of the scheduling gate applied to Pods
+  // to delay their scheduling until the associated workload slice has been admitted.
+  // This gate ensures that Pods do not begin scheduling prematurely, maintaining
+  // proper sequencing in workload processing.
+  // This scheduling gate is alpha-level.
+  ElasticJobSchedulingGate = "kueue.x-k8s.io/elastic-job"
+)
+```
 
-After the creation of each individual Workload Slice and admission of a Workload Slice, the **workload_scheduling_gates_controller** should be in charge of removing the scheduling gates from each pod. All worker pods from the same worker group share the same pod template, so we only need to ungate the number of pods to match the number of admitted pods, this should be a counter. We don’t want to accidentally ungate too many pods since race conditions could happen and we also don’t want to double count. It's worth mentioning that for the case of recreated pods (i.e. machine failure for example), these pods will go through the admission/scheduling check again, Kueue is responsible fo removing the scheduling gates when there's available quota and resources to spend on the Job. 
+### Limitations and Incompatibilities
 
-### Garbage Collecting Workload Slices
+This section captures all known limitations and incompatibilities introduced by, or resulting from, enabling the ElasticJobs feature.
 
-The logic of folding and deleting would be isolated in this controller. We don’t necessarily have to say that this Workload Slice belongs to a specific pod. This controller will look at the Workload objects and check whether they have the Admitted condition or not.
+#### PartialAdmission
 
-1. The controller increments the `.status.admission.podSetAssignments.count` and passes the UID of the workload you are folding to the parent workload.
-If we still don’t see the workload being deleted, we at least know it has been counted towards the parent workload and the controller shouldn't fold it again.
+A given Job instance cannot have both `PartialAdmission` and `ElasticJob` enabled. This is a **per-Job limitation**, not a global feature-level conflict. It is entirely valid to have both features enabled in the system, just not simultaneously on the same Job.
 
-## Phases for MVP (alpha)
+**Rationale**
 
-### Phase 1 - Scale Down
-Scaling down will be the first phase towards MVP because it can be implemented without introducing the Workload Slices. 
+* `PartialAdmission` depends on the static nature of a workload. Specifically, it assumes, reasonably, given the current model, that workloads are immutable in terms of `podSets[].count`. Based on this assumption, it adjusts the Job’s spec to reflect the minimum allowed parallelism at scheduling time. This value is fixed and will not change, even if the ClusterQueue’s capacity increases later.
 
-Scaling down a Job won’t involve the creation of Workload Slices, instead it’ll involve an update to the current workload, no requeuing.
+* In practice, `PartialAdmission` takes ownership of `job.spec.parallelism`, and Kueue enforces this immutability through the admission webhook. For example:
 
-1. Compare Pod Counts: Within the job framework, check if the PodSet.Count of the job matches the Workload.Spec.PodSets[1].Count (worker group).
-2. Synchronize Workload: If the pod counts don't match, update the workload to align with the job's PodSet.Count.
-3. Trigger Resource Updates:  By updating the workload in cache and in queue, we'll signal the following:
-- The cluster queue resource usage should recalculate, ensuring accurate resource management.
-- The scheduler should re-evaluate and update PodSetAssignments related to the workload.
+  ```text
+  admission webhook "vjob.kb.io" denied the request: spec.parallelism: Forbidden: cannot change when partial admission is enabled and the job is not suspended
+  ```
 
+**Validation**
 
-#### Job controller
+To avoid delayed or implicit validation errors, any Kueue-integrated Job that supports `ElasticJobs` should include admission validation logic to reject the following annotation combination:
 
-Rework *equivalentToWorkload()* and *reconcile()* to account for potential differences between the job’s number of replicas and the running workload’s PodSet.Count. 
+```yaml
+annotations:
+  kueue.x-k8s.io/dynamically-sized-job: "true"
+  kueue.x-k8s.io/job-min-parallelism: "1"
+```
 
-Given these changes, check if the delta is positive or negative indicating a scaleup/scaledown. If it’s a scaledown the reconciler should trigger an update on the workload to match the new job’s spec and no quota needs to be checked or accounted for, the cluster queue should update the workload resource usage. 
+## Phases for MVP (alpha)
 
-**Important:** in Phase 1 if there's a scale up, the behaviour will be suspend and requeue. This will be temporary while Phase 2 and 3 are not completed.
+### Phase 1 - batchv1/Job WorkloadSlices Support in Single-Cluster Configuration.
 
-### Phase 2 - Aggregating Workload Slices
+Scaling up and down for `batch/v1.Job` will be the initial phase of the MVP, as it builds on stable and well-understood Kubernetes components without requiring additional integration work.
 
-In Phase 2, aggregating Workload Slices into the parent workload will be implemented. This doesn't represent a usable feature to end users, but it can be reviewed independently from the phase 3.
+#### Scale Down
 
-### Phase 3 - Scale up with Workload Slices and Scheduling Gates
+1. Create a Job with workload-slice enablement and a parallelism value greater than 1. 
+2. Observe that a corresponding Workload is created and admitted, and that the Job's pods are in the Running state, matching the specified parallelism. 
+3. Update the Job by reducing its parallelism value. 
+4. Confirm that the existing Workload is updated accordingly, with WorkloadSpec.PodSets[0].Count reflecting the new parallelism value. 
+5. Observe that the number of running pods is reduced, while the remaining pods continue to run uninterrupted.
 
-In Phase 3, scale up will be implemented by introducing Workload Slices and adding Pod scheduling gates as part of Kueue’s mutating admission for the job. 
+#### Scale Up
 
-When a job scales, its webhook would be modified to intercept and "gate" all pods. Every time there’s a resize, you create a dependable (child) workload slice and once it's admitted, you increase the count in the original workload, delete the old workload and remove the schedulingGates. 
-- Pros: You are able to hold the pods added by Kuberay
-- Cons: The fact of having schedulingGates, means we need an API call for every pod, because all pods that are created by the job are going to have schedulingGates. We need to remove those gates and for every pod you need to make API calls.
+1. Create a `batch/v1.Job` with workload-slice enablement and an initial `parallelism` value greater than 0.
+2. Verify that a corresponding `Workload` is created and admitted, and that the Job’s pods are in the `Running` state, matching the specified parallelism.
+3. Increase the Job’s `parallelism` value to trigger a scale-up event.
+4. Confirm that a new `Workload` slice is created and admitted, reflecting the increased pod count, while the previous slice is marked as `Finished`.
+5. Observe that the total number of running pods increases to match the updated parallelism, with the original pods continuing to run without disruption.
 
-#### Scheduler
-Since every scale up will have its own individual workload they should proceed to the current scheduling cycle and continue the normal admission process.
-However, we need to ensure that the Workload Slice is assigned the same resource flavor as the parent workload. 
+### Phase 2 – RayCluster WorkloadSlice Support in Single-Cluster Configuration
+This phase mirrors the steps from Phase 1 but is adapted specifically for RayCluster workloads, taking into account their autoscaling behavior and internal lifecycle management.
 
-The `nominate()` returns the workloads with their requirements (resource flavors, borrowing) if they were admitted by the clusterQueues in the snapshot, so we need to return the original workload that was already admitted with the resize information inside *TotalRequests* so that *PodSetAssignments* is locked to the parent Workload assignments.
+### Phase 3 – Enabling Workload Slicing for batch/v1.Job in Multi-Cluster Configuration
+In this phase, Workload Slicing support for batch/v1.Job will be extended to multi-cluster environments using Kueue’s MultiKueue architecture. 
+When a job is scaled in the management cluster, its corresponding WorkloadSlice will be propagated to the appropriate workload cluster(s), respecting existing cluster assignment and resource flavor constraints. 
+Each WorkloadSlice will be subject to independent admission in the target cluster, and only after successful admission will scheduling gates be lifted to allow pod execution. 
+Slice preemption, quota accounting, and garbage collection must be coordinated across clusters to ensure consistency and avoid orphaned resources. 
+This phase will validate correctness and stability of the slicing mechanism in a federated deployment model.
 
 ## Additional Details
 
@@ -209,18 +372,59 @@ extending the production code to implement this enhancement.
 -->
 The code will adhere to regular best practices for unit tests and coverage.
 
-
-
 #### Integration tests
+
 Integration tests will be executed against mocked clients for Jobs 
 that will provide predefined responses and allow to test various scenarios, 
 including situations like:
 
-* Job has a scale up, workload slices get created, pods are gated
-* Job has a scale up, gated pods are admitted, pods get ungated and assigned same flavors as parent workload
-* Workload slices are correctly folded and deleted
-* Job has a scale down, Workload spec reflects podset count values
-* When Kueue preempt a resized Job, it should preempt it as a whole
+#### Scale-down
+* Job scale down, cluster/local queue correctly reflect resource flavor used capacity.
+
+Here’s a revised and polished version of your text with improved grammar, structure, and clarity:
+
+---
+
+#### Integration Tests
+
+Integration tests will be executed against mocked clients for `batchv1/Job` resources. These mocks will simulate predefined responses, enabling validation of various scheduling and workload slice scenarios, including the following cases:
+
+---
+
+#### Scale-Down
+
+* A `Job` is scaled down.
+* The ClusterQueue and/or local queue correctly reflect the reduced resource usage for the assigned resource flavor.
+
+---
+
+#### Scale-Up
+
+* A `batchv1/Job` is updated to increase its `parallelism` value.
+* New pods are created in the `ScheduleGated` state.
+* A new `WorkloadSlice` is created and admitted.
+* The previous `WorkloadSlice` is marked as "Finished."
+* `ScheduleGates` are removed from the new pods, allowing them to transition to the running state.
+
+---
+
+#### Resource Flavor Handling
+
+* Kueue is configured with two resource flavors, each with different CPU capacities.
+* A `batchv1/Job` is created that matches the selection criteria for both flavors.
+* The job is initially admitted using the flavor with the smaller capacity.
+* The job is later updated to increase `parallelism` beyond the capacity of the originally assigned flavor, but within the capacity of the alternative flavor.
+* New pods are created in the `ScheduleGated` state.
+* The new `WorkloadSlice` remains in a `Pending` state due to the sticky flavor constraint. The status condition resembles the following:
+
+```yaml
+- lastTransitionTime: "2025-06-12T21:32:56Z"
+  message: "couldn't assign flavors to pod set main: couldn't change flavor from: smaller-flavor to: larger-flavor, insufficient quota for cpu in flavor smaller-flavor, request > maximum capacity (1100m > 1)"
+  observedGeneration: 1
+  reason: Pending
+  status: "False"
+  type: QuotaReserved 
+```
 
 ### Graduation Criteria
 <!--
@@ -239,10 +443,54 @@ milestones with these graduation criteria:
 [deprecation-policy]: https://kubernetes.io/docs/reference/using-api/deprecation-policy/
 -->
 
-The feature starts at the alpha level, with a feature gate.
+Here’s a structured and detailed **Graduation Criteria** section for KEP-77: *Dynamically Sized Jobs*, tailored to your current implementation of `WorkloadSlice` and phased MVP plan.
+
+---
 
-In the Alpha version, Dynamically Sized Jobs will support RayCluster resizing. 
-- Scale up/down of replica workers
+### Graduation Criteria
+
+#### Alpha
+
+* [x] Feature is gated by the `ElasticJobsViaWorkloadSlices` feature gate.
+* [x] WorkloadSlice is implemented as an internal Kueue construct (no new CRDs).
+* [x] Scale-up events result in the creation of new `WorkloadSlice`s; scale-downs result in updates to the original `Workload`.
+* [x] Pod-level gating using `PodSchedulingGates` is integrated with slice admission.
+* [x] Sticky flavor behavior is enforced: scale-ups must reuse the original assigned flavor.
+* [x] Superseded slices are marked as `Finished` once the new slice is admitted.
+* [x] Support for `batch/v1.Job` in both single-cluster and multi-cluster (multiKueue) configurations.
+* [x] Integration tests cover basic scale-up, scale-down, and resource flavor constraints.
+* [x] Feature opt-in is supported via workload annotation (`kueue.x-k8s.io/elastic-job: "true"`).
+* [x] Documented enablement steps, slice behavior, and known limitations.
+
+#### Beta
+
+* [ ] Feature is enabled by default in Kueue, still guarded by the feature gate for opt-out.
+* [ ] Complete support for garbage collection of preempted and inactive `WorkloadSlice`s, with tunable retention (e.g., `revisionHistoryLimit`-like mechanism).
+* [ ] Metrics and events emitted for slice lifecycle transitions (e.g., created, admitted, failed admission).
+* [ ] Documentation includes examples for users and integrators to adopt the WorkloadSlice model.
+* [ ] Formal conformance tests validate end-to-end behavior for:
+    * Horizontal scale-up and scale-down
+    * Slice replacements
+    * Sticky flavor enforcement
+    * Multi-cluster propagation
+* [ ] At least one additional framework beyond `batch/v1.Job` (e.g., RayCluster) integrates and validates the WorkloadSlice flow.
+* [ ] Slice lifecycle events (e.g., admitted, preempted, finished) are observable via `kubectl describe workload` or equivalent API tools.
+* [ ] Slice preemption is consistently handled and visible in workload status conditions.
+* [ ] All Kueue core controllers (scheduler, preemptor, queue manager) are validated under slice-enabled workloads.
+* [ ] Dynamic resizing is enabled for all Kueue-managed workloads that support the elastic-job feature (including JobSet, RayJob, Kubeflow jobs, etc.)
+* [ ] Re-evaluate the WorkloadSlice implementation to ensure compatibility with elastic workloads, considering all current and emerging alternatives within Kueue.
+* [ ] Re-evaluate currently disallowed per-job-instance combination of enabled PartialAdmission and ElasticJobs.
+* [ ] Re-evaluate the approach for removing PodSchedulingReadiness gate for admitted workload slices to use a dedicated controller rather than calling from Job reconciler (see 3. in [comment](https://github.com/kubernetes-sigs/kueue/pull/5510#issuecomment-3060737465)).
+
+#### GA (Stable)
+
+* [ ] Proven stability under production-scale workloads, verified through internal deployments or community reports.
+* [ ] Full backwards compatibility: workloads that do not opt into `WorkloadSlice` continue to function identically.
+* [ ] API guarantees around slice naming, preemption markers, and flavor enforcement are documented and stable.
+* [ ] No known correctness issues across single- and multi-cluster environments.
+* [ ] User-configurable policies (optional) for flavor migration or slice aggregation behavior are validated.
+* [ ] Feature is permanently enabled and no longer gated.
+* [ ] Associated documentation, examples, and operational best practices are published as part of the GA release.
 
 
 ## Implementation History
@@ -260,18 +508,40 @@ Major milestones might include:
 
 ## Drawbacks
 
-<!--
-Why should this KEP _not_ be implemented?
--->
+- Adds complexity to Kueue’s scheduling and preemption model.
+- Users must understand new slice lifecycle semantics, which differ from traditional Workload behavior.
+- Leveraging PodSchedulingGates together with ResourceQuota enforcement may result out-of-quota namespaces (see section below for more details).
 
+### ResourceQuota
 
-## Alternatives
+`ResourceQuota` in Kubernetes is a **namespace-scoped object** that limits how much compute, storage, and object-count-based resources a namespace can consume. It is used to enforce fair resource usage in multi-tenant environments and to prevent any single team or workload from consuming excessive cluster resources.
 
+Quota enforcement occurs during the **pod admission phase**. When a pod is created, the API server checks whether the requested resources exceed the available quota in the namespace. If they do, the creation request is rejected.
 
-### Ignore Resize from Kuberay
+`ResourceQuota` includes all pods by default, with a few specific exceptions:
+
+| Pod Type / State         | Counted in Quota? | Notes                                |
+| ------------------------ | ----------------- | ------------------------------------ |
+| Running / Pending Pods   | ✅ Yes             | If they belong to the namespace      |
+| Succeeded / Failed Pods  | ❌ No              | Considered completed and inactive    |
+| Terminating Pods         | ❌ No              | Identified by a `deletionTimestamp`  |
+| Mirror Pods              | ❌ No              | Created directly by the kubelet      |
+| Pods outside quota scope | ❌ No              | If the `ResourceQuota` uses `scopes` |
+
+---
+
+### PodScheduling Readiness and ResourceQuota
+
+**PodScheduling Readiness** is a mechanism that ensures a pod is only eligible for scheduling when it meets specific conditions. This is especially useful for multi-stage scheduling, resource reservation, or coordination with external systems.
+
+An important observation: by default, pods that are gated using `PodSchedulingGates` are still counted toward the namespace's `ResourceQuota` if they do not match any of the exclusion conditions listed above.
+
+As a result, when `PodSchedulingGates` are used in a namespace with quota enforcement, it is possible to exhaust the available quota with ScheduleGated pods that are not yet schedulable. This can block additional pods from being admitted, even though none are actively running.
+
+
+
+## Alternatives
 
-- Idea: Ignoring the scale up or rejecting the scale up from the auto scaler and storing the value in an annotation so that Kueue takes a decision based on the annotation. We’d need a signal from Kueue to hold this in the raycluster_webhook and identify when the resize comes from Kueue, it has to be accepted.
-- Pros: No need to intercept the pods and no need of using schedulingGates
-- Cons: There would be a permanent race bewteen Kueue and Kuberay autoscaler to change the counter in the RayCluster replica number for the worker group. 
-- Exploration: See if autoscaler would indicate a desired size in the spec without altering the number of replicas directly. 
-- Discarded: Higher complexity than gating/ungating pods via SchedulingGates
+- Require users to manage resizing manually by recreating jobs.
+- Defer support for elastic workloads to higher-level controllers (e.g., RayOperator), leaving Kueue unaware of scale operations.
+- [WorkloadResize Request](https://github.com/kubernetes-sigs/kueue/issues/5897) an exploration of an alternative approach to elastic jobs.  
\ No newline at end of file
diff --git a/keps/77-dynamically-sized-jobs/kep.yaml b/keps/77-dynamically-sized-jobs/kep.yaml
index 9c6a69fe3..f965cf918 100644
--- a/keps/77-dynamically-sized-jobs/kep.yaml
+++ b/keps/77-dynamically-sized-jobs/kep.yaml
@@ -2,6 +2,7 @@ title:
 kep-number: 77
 authors:
   - "@vicentefb"
+  - "@ichekrygin"
 status: provisional
 creation-date: 2024-03-14
 reviewers:
@@ -9,15 +10,17 @@ reviewers:
   - "@alculquicondor"
   - "@astefanutti"
   - "@tenzen-y"
+  - "@mimowo"
 approvers:
   - "@alculquicondor"
-
+  - "@mimowo"
+  - "@tenzen-y"
 # The target maturity stage in the current dev cycle for this KEP.
 stage: alpha
 
 # The milestone at which this feature was, or is targeted to be, at each stage.
 milestone:
-  alpha: "v0.8"
+  alpha: "v0.13"
 
 feature-gates:
-  - name: DynamicallySizedJobs
\ No newline at end of file
+  - name: ElasticJobsViaWorkloadSlices
\ No newline at end of file
diff --git a/pkg/controller/jobframework/reconciler.go b/pkg/controller/jobframework/reconciler.go
index b5cda36a4..f5f94ae94 100644
--- a/pkg/controller/jobframework/reconciler.go
+++ b/pkg/controller/jobframework/reconciler.go
@@ -61,6 +61,7 @@ import (
 	utilpriority "sigs.k8s.io/kueue/pkg/util/priority"
 	"sigs.k8s.io/kueue/pkg/util/slices"
 	"sigs.k8s.io/kueue/pkg/workload"
+	"sigs.k8s.io/kueue/pkg/workloadslicing"
 )
 
 const (
@@ -394,8 +395,7 @@ func (r *JobReconciler) ReconcileGenericJob(ctx context.Context, req ctrl.Reques
 
 	log.V(2).Info("Reconciling Job")
 
-	// 1. make sure there is only a single existing instance of the workload.
-	// If there's no workload exists and job is unsuspended, we'll stop it immediately.
+	// 1. Attempt to retrieve an existing workload (if any) for this job.
 	wl, err := r.ensureOneWorkload(ctx, job, object)
 	if err != nil {
 		return ctrl.Result{}, err
@@ -601,7 +601,12 @@ func (r *JobReconciler) ReconcileGenericJob(ctx context.Context, req ctrl.Reques
 
 	// 8. handle job is unsuspended.
 	if !workload.IsAdmitted(wl) {
-		// the job must be suspended if the workload is not yet admitted.
+		// The job must be suspended if the workload is not yet admitted,
+		// unless this job is workload-slicing enabled. In workload-slicing we rely
+		// on pod-scheduling gate(s) to pause workload slice pods during the workload admission process.
+		if workloadSliceEnabled(job) {
+			return ctrl.Result{}, nil
+		}
 		log.V(2).Info("Running job is not admitted by a cluster queue, suspending")
 		err := r.stopJob(ctx, job, wl, StopReasonNotAdmitted, "Not admitted by cluster queue")
 		if err != nil {
@@ -610,6 +615,12 @@ func (r *JobReconciler) ReconcileGenericJob(ctx context.Context, req ctrl.Reques
 		return ctrl.Result{}, err
 	}
 
+	if workloadSliceEnabled(job) {
+		// Start workload-slice schedule-gated pods (if any).
+		log.V(3).Info("Job running with admitted workload slice, start pods.")
+		return ctrl.Result{}, workloadslicing.StartWorkloadSlicePods(ctx, r.client, object)
+	}
+
 	// workload is admitted and job is running, nothing to do.
 	log.V(3).Info("Job running with admitted workload, nothing to do")
 	return ctrl.Result{}, nil
@@ -772,6 +783,26 @@ func FindAncestorJobManagedByKueue(ctx context.Context, c client.Client, jobObj
 func (r *JobReconciler) ensureOneWorkload(ctx context.Context, job GenericJob, object client.Object) (*kueue.Workload, error) {
 	log := ctrl.LoggerFrom(ctx)
 
+	// If workload slicing is enabled for this job, use the slice-based processing path.
+	if workloadSliceEnabled(job) {
+		podSets, err := job.PodSets()
+		if err != nil {
+			return nil, fmt.Errorf("failed to retrieve pod sets from job: %w", err)
+		}
+
+		// Workload slices allow modifications only to PodSet.Count.
+		// Any other changes will result in the slice being marked as incompatible,
+		// and the workload will fall back to being processed by the original ensureOneWorkload function.
+		wl, compatible, err := workloadslicing.EnsureWorkloadSlices(ctx, r.client, podSets, object, job.GVK())
+		if err != nil {
+			return nil, err
+		}
+		if compatible {
+			return wl, nil
+		}
+		// Fallback.
+	}
+
 	if prebuiltWorkloadName, usePrebuiltWorkload := PrebuiltWorkloadFor(job); usePrebuiltWorkload {
 		wl := &kueue.Workload{}
 		err := r.client.Get(ctx, types.NamespacedName{Name: prebuiltWorkloadName, Namespace: object.GetNamespace()}, wl)
@@ -1132,6 +1163,17 @@ func (r *JobReconciler) constructWorkload(ctx context.Context, job GenericJob) (
 	return ConstructWorkload(ctx, r.client, job, r.labelKeysToCopy)
 }
 
+// newWorkloadName generates a new workload name for the given job, incorporating the job's name, UID,
+// and GroupVersionKind (GVK). If workload slicing is enabled, it includes the job's generation
+// in the generated workload name.
+func newWorkloadName(job GenericJob) string {
+	object := job.Object()
+	if workloadSliceEnabled(job) {
+		return GetWorkloadNameForOwnerWithGVKAndGeneration(object.GetName(), object.GetUID(), job.GVK(), object.GetGeneration())
+	}
+	return GetWorkloadNameForOwnerWithGVK(object.GetName(), object.GetUID(), job.GVK())
+}
+
 func ConstructWorkload(ctx context.Context, c client.Client, job GenericJob, labelKeysToCopy []string) (*kueue.Workload, error) {
 	log := ctrl.LoggerFrom(ctx)
 	object := job.Object()
@@ -1141,8 +1183,7 @@ func ConstructWorkload(ctx context.Context, c client.Client, job GenericJob, lab
 		return nil, err
 	}
 
-	wl := NewWorkload(GetWorkloadNameForOwnerWithGVK(object.GetName(), object.GetUID(), job.GVK()), object, podSets, labelKeysToCopy)
-
+	wl := NewWorkload(newWorkloadName(job), object, podSets, labelKeysToCopy)
 	if wl.Labels == nil {
 		wl.Labels = make(map[string]string)
 	}
@@ -1163,6 +1204,33 @@ func ConstructWorkload(ctx context.Context, c client.Client, job GenericJob, lab
 	return wl, nil
 }
 
+// prepareWorkloadSlice adds necessary workload slice annotations.
+func prepareWorkloadSlice(ctx context.Context, clnt client.Client, job GenericJob, wl *kueue.Workload) error {
+	// Lookup existing slice for a given job.
+	workloadSlices, err := workloadslicing.FindNotFinishedWorkloads(ctx, clnt, job.Object(), job.GVK())
+	if err != nil {
+		return fmt.Errorf("failure looking up workload slices: %w", err)
+	}
+
+	switch len(workloadSlices) {
+	case 0:
+		// No active workloads found for this job - noop.
+		return nil
+	case 1:
+		// One active workload found for this job - typically, we are in a scale-up event, where previous
+		// workload is the "old" slice.
+		oldSlice := workloadSlices[0]
+		// Annotate new workload slice with the preemptible (old) workload slice.
+		metav1.SetMetaDataAnnotation(&wl.ObjectMeta, workloadslicing.WorkloadSliceReplacementFor, string(workload.Key(&oldSlice)))
+		return nil
+	default:
+		// Any other slices length is invalid. I.E, we expect to have at most 1 "current/old" workload slice.
+		// Failing here, would trigger job re-processing, and hopefully giving a chance to clear up (preempt/deactivate)
+		// old slice.
+		return fmt.Errorf("unexpected workload-slices count: %d", len(workloadSlices))
+	}
+}
+
 // prepareWorkload adds the priority information for the constructed workload
 func (r *JobReconciler) prepareWorkload(ctx context.Context, job GenericJob, wl *kueue.Workload) error {
 	priorityClassName, source, p, err := r.extractPriority(ctx, wl.Spec.PodSets, job)
@@ -1173,9 +1241,11 @@ func (r *JobReconciler) prepareWorkload(ctx context.Context, job GenericJob, wl
 	wl.Spec.PriorityClassName = priorityClassName
 	wl.Spec.Priority = &p
 	wl.Spec.PriorityClassSource = source
-
 	wl.Spec.PodSets = clearMinCountsIfFeatureDisabled(wl.Spec.PodSets)
 
+	if workloadSliceEnabled(job) {
+		return prepareWorkloadSlice(ctx, r.client, job, wl)
+	}
 	return nil
 }
 
@@ -1252,8 +1322,10 @@ func (r *JobReconciler) handleJobWithNoWorkload(ctx context.Context, job Generic
 		}
 	}
 
-	// Wait until there are no active pods.
-	if job.IsActive() {
+	// Wait until there are no active pods, unless this is a workload-slice job.
+	// For workload-slice enabled jobs, we allow the job to remain "Active" to accommodate
+	// the scale-up case, where the new workload slice replaces the old workload slice.
+	if job.IsActive() && !workloadSliceEnabled(job) {
 		log.V(2).Info("Job is suspended but still has active pods, waiting")
 		return nil
 	}
@@ -1401,3 +1473,19 @@ func clearMinCountsIfFeatureDisabled(in []kueue.PodSet) []kueue.PodSet {
 	}
 	return in
 }
+
+// workloadSliceEnabled returns true if all the following conditions are met:
+//   - The ElasticJobsViaWorkloadSlices feature is enabled.
+//   - The provided job is not nil.
+//   - The job's underlying object is not nil.
+//   - The job's object has opted in for WorkloadSlice processing.
+func workloadSliceEnabled(job GenericJob) bool {
+	if !features.Enabled(features.ElasticJobsViaWorkloadSlices) || job == nil {
+		return false
+	}
+	jobObject := job.Object()
+	if jobObject == nil {
+		return false
+	}
+	return workloadslicing.Enabled(jobObject)
+}
diff --git a/pkg/controller/jobframework/workload_names.go b/pkg/controller/jobframework/workload_names.go
index 9a7b92527..7dd9d3df9 100644
--- a/pkg/controller/jobframework/workload_names.go
+++ b/pkg/controller/jobframework/workload_names.go
@@ -19,10 +19,12 @@ package jobframework
 import (
 	"crypto/sha1"
 	"encoding/hex"
+	"strconv"
 	"strings"
 
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/utils/ptr"
 )
 
 const (
@@ -32,14 +34,22 @@ const (
 )
 
 func GetWorkloadNameForOwnerWithGVK(ownerName string, ownerUID types.UID, ownerGVK schema.GroupVersionKind) string {
+	return generateWorkloadName(ownerName, ownerUID, ownerGVK, nil)
+}
+
+func GetWorkloadNameForOwnerWithGVKAndGeneration(ownerName string, ownerUID types.UID, ownerGVK schema.GroupVersionKind, generation int64) string {
+	return generateWorkloadName(ownerName, ownerUID, ownerGVK, &generation)
+}
+
+func generateWorkloadName(ownerName string, ownerUID types.UID, ownerGVK schema.GroupVersionKind, generation *int64) string {
 	prefixedName := strings.ToLower(ownerGVK.Kind) + "-" + ownerName
 	if len(prefixedName) > maxPrefixLength {
 		prefixedName = prefixedName[:maxPrefixLength]
 	}
-	return prefixedName + "-" + getHash(ownerName, ownerUID, ownerGVK)[:hashLength]
+	return prefixedName + "-" + getHash(ownerName, ownerUID, ownerGVK, generation)[:hashLength]
 }
 
-func getHash(ownerName string, ownerUID types.UID, gvk schema.GroupVersionKind) string {
+func getHash(ownerName string, ownerUID types.UID, gvk schema.GroupVersionKind, generation *int64) string {
 	h := sha1.New()
 	h.Write([]byte(gvk.Kind))
 	h.Write([]byte("\n"))
@@ -48,5 +58,9 @@ func getHash(ownerName string, ownerUID types.UID, gvk schema.GroupVersionKind)
 	h.Write([]byte(ownerName))
 	h.Write([]byte("\n"))
 	h.Write([]byte(ownerUID))
+	if generation != nil {
+		h.Write([]byte("\n"))
+		h.Write([]byte(strconv.FormatInt(ptr.Deref(generation, 0), 10)))
+	}
 	return hex.EncodeToString(h.Sum(nil))
 }
diff --git a/pkg/controller/jobs/job/job_controller.go b/pkg/controller/jobs/job/job_controller.go
index c8cc26495..e44067b37 100644
--- a/pkg/controller/jobs/job/job_controller.go
+++ b/pkg/controller/jobs/job/job_controller.go
@@ -370,6 +370,14 @@ func SetupIndexes(ctx context.Context, fieldIndexer client.FieldIndexer) error {
 	if err := fieldIndexer.IndexField(ctx, &batchv1.Job{}, indexer.OwnerReferenceUID, indexer.IndexOwnerUID); err != nil {
 		return err
 	}
+	// Add pod index to be able to list pods for elastic-jobs, needed to remove scheduling gate on
+	// admitted workload slices.
+	if features.Enabled(features.ElasticJobsViaWorkloadSlices) {
+		if err := fieldIndexer.IndexField(ctx, &corev1.Pod{}, indexer.OwnerReferenceUID, indexer.IndexOwnerUID); err != nil {
+			return err
+		}
+	}
+
 	return jobframework.SetupWorkloadOwnerIndex(ctx, fieldIndexer, gvk)
 }
 
diff --git a/pkg/controller/jobs/job/job_webhook.go b/pkg/controller/jobs/job/job_webhook.go
index 008b7336f..fcf911dfe 100644
--- a/pkg/controller/jobs/job/job_webhook.go
+++ b/pkg/controller/jobs/job/job_webhook.go
@@ -19,9 +19,11 @@ package job
 import (
 	"context"
 	"fmt"
+	"slices"
 	"strconv"
 
 	batchv1 "k8s.io/api/batch/v1"
+	corev1 "k8s.io/api/core/v1"
 	apivalidation "k8s.io/apimachinery/pkg/api/validation"
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -31,10 +33,13 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/webhook/admission"
 
+	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta1"
 	"sigs.k8s.io/kueue/pkg/cache"
 	"sigs.k8s.io/kueue/pkg/controller/jobframework"
 	"sigs.k8s.io/kueue/pkg/controller/jobframework/webhook"
+	"sigs.k8s.io/kueue/pkg/features"
 	"sigs.k8s.io/kueue/pkg/queue"
+	"sigs.k8s.io/kueue/pkg/workloadslicing"
 )
 
 var (
@@ -43,6 +48,31 @@ var (
 	replicaMetaPath               = field.NewPath("spec", "template", "metadata")
 )
 
+// applyWorkloadSliceSchedulingGate ensures that the workload slice-specific
+// PodSchedulingGate is present in the Workload-slice enabled Job's pod template.
+// If the scheduling gate is not already included, it appends it to the list of scheduling gates.
+//
+// This function is essential for enabling workload slice-aware scheduling
+// behavior in Kueue-managed Jobs, allowing for fine-grained control over
+// resource allocation and scheduling decisions.
+//
+// Parameters:
+//   - job: Pointer to the Job object to be modified.
+//
+// Note: This function modifies the Job's pod template in-place.
+func applyWorkloadSliceSchedulingGate(job *Job) {
+	if !features.Enabled(features.ElasticJobsViaWorkloadSlices) || !workloadslicing.Enabled(job.Object()) {
+		return
+	}
+	workloadSliceSchedulingGate := corev1.PodSchedulingGate{
+		Name: kueue.ElasticJobSchedulingGate,
+	}
+	if slices.Contains(job.Spec.Template.Spec.SchedulingGates, workloadSliceSchedulingGate) {
+		return
+	}
+	job.Spec.Template.Spec.SchedulingGates = append(job.Spec.Template.Spec.SchedulingGates, workloadSliceSchedulingGate)
+}
+
 type JobWebhook struct {
 	client                       client.Client
 	manageJobsWithoutQueueName   bool
@@ -85,6 +115,8 @@ func (w *JobWebhook) Default(ctx context.Context, obj runtime.Object) error {
 	}
 	jobframework.ApplyDefaultForManagedBy(job, w.queues, w.cache, log)
 
+	applyWorkloadSliceSchedulingGate(job)
+
 	return nil
 }
 
diff --git a/pkg/features/kube_features.go b/pkg/features/kube_features.go
index 80da7e707..dc70666a9 100644
--- a/pkg/features/kube_features.go
+++ b/pkg/features/kube_features.go
@@ -156,6 +156,12 @@ const (
 	// Enable replacement of failed node in TAS.
 	TASFailedNodeReplacement featuregate.Feature = "TASFailedNodeReplacement"
 
+	// owner: @ichekrygin
+	// kep: https://github.com/kubernetes-sigs/kueue/tree/main/keps/77-dynamically-sized-jobs
+	//
+	// ElasticJobsViaWorkloadSlices enables workload-slices support.
+	ElasticJobsViaWorkloadSlices featuregate.Feature = "ElasticJobsViaWorkloadSlices"
+
 	// owner: @pbundyra
 	// kep: https://github.com/kubernetes-sigs/kueue/tree/main/keps/2724-topology-aware-scheduling
 	//
@@ -260,6 +266,9 @@ var defaultVersionedFeatureGates = map[featuregate.Feature]featuregate.Versioned
 	TASFailedNodeReplacement: {
 		{Version: version.MustParse("0.12"), Default: false, PreRelease: featuregate.Alpha},
 	},
+	ElasticJobsViaWorkloadSlices: {
+		{Version: version.MustParse("0.13"), Default: false, PreRelease: featuregate.Alpha},
+	},
 	TASFailedNodeReplacementFailFast: {
 		{Version: version.MustParse("0.13"), Default: false, PreRelease: featuregate.Alpha},
 	},
diff --git a/pkg/metrics/metrics.go b/pkg/metrics/metrics.go
index a4e8c2852..948de6a87 100644
--- a/pkg/metrics/metrics.go
+++ b/pkg/metrics/metrics.go
@@ -280,6 +280,14 @@ The label 'reason' can have the following values:
 		}, []string{"cluster_queue", "reason"},
 	)
 
+	ReplacedWorkloadSlicesTotal = prometheus.NewCounterVec(
+		prometheus.CounterOpts{
+			Subsystem: constants.KueueName,
+			Name:      "replaced_workload_slices_total",
+			Help:      `The number of replaced workload slices per 'cluster_queue'`,
+		}, []string{"cluster_queue"},
+	)
+
 	LocalQueueEvictedWorkloadsTotal = prometheus.NewCounterVec(
 		prometheus.CounterOpts{
 			Subsystem: constants.KueueName,
@@ -533,6 +541,10 @@ func ReportEvictedWorkloads(cqName kueue.ClusterQueueReference, evictionReason s
 	}
 }
 
+func ReportReplacedWorkloadSlices(cqName kueue.ClusterQueueReference) {
+	ReplacedWorkloadSlicesTotal.WithLabelValues(string(cqName)).Inc()
+}
+
 func ReportLocalQueueEvictedWorkloads(lq LocalQueueReference, reason string) {
 	LocalQueueEvictedWorkloadsTotal.WithLabelValues(string(lq.Name), lq.Namespace, reason).Inc()
 }
diff --git a/pkg/scheduler/flavorassigner/flavorassigner.go b/pkg/scheduler/flavorassigner/flavorassigner.go
index 0c0a12d3d..9bdbbd117 100644
--- a/pkg/scheduler/flavorassigner/flavorassigner.go
+++ b/pkg/scheduler/flavorassigner/flavorassigner.go
@@ -384,15 +384,26 @@ type FlavorAssigner struct {
 	resourceFlavors   map[kueue.ResourceFlavorReference]*kueue.ResourceFlavor
 	enableFairSharing bool
 	oracle            preemptionOracle
+
+	// preemptWorkloadSlice identifies the workload slice that will be mandatorily preempted
+	// by this workload. It must be considered during flavor computation and included in the preemption targets.
+	//
+	// Note: This value may be nil in the following cases:
+	//   - Workload slicing is not enabled (either globally or for this specific workload).
+	//   - The current workload does not represent a scale-up slice.
+	// In these scenarios, flavor assignment proceeds as in the original flow—i.e., as for regular,
+	// non-sliced workloads.
+	preemptWorkloadSlice *workload.Info
 }
 
-func New(wl *workload.Info, cq *cache.ClusterQueueSnapshot, resourceFlavors map[kueue.ResourceFlavorReference]*kueue.ResourceFlavor, enableFairSharing bool, oracle preemptionOracle) *FlavorAssigner {
+func New(wl *workload.Info, cq *cache.ClusterQueueSnapshot, resourceFlavors map[kueue.ResourceFlavorReference]*kueue.ResourceFlavor, enableFairSharing bool, oracle preemptionOracle, preemptWorkloadSlice *workload.Info) *FlavorAssigner {
 	return &FlavorAssigner{
-		wl:                wl,
-		cq:                cq,
-		resourceFlavors:   resourceFlavors,
-		enableFairSharing: enableFairSharing,
-		oracle:            oracle,
+		wl:                   wl,
+		cq:                   cq,
+		resourceFlavors:      resourceFlavors,
+		enableFairSharing:    enableFairSharing,
+		oracle:               oracle,
+		preemptWorkloadSlice: preemptWorkloadSlice,
 	}
 }
 
@@ -671,6 +682,24 @@ func (a *FlavorAssigner) findFlavorForPodSetResource(
 		// Calculate representativeMode for this assignment as the worst mode among all requests.
 		representativeMode := fit
 		for rName, val := range requests {
+			// Ensure the same resource flavor is used for the workload slice as in the original admitted slice.
+			if features.Enabled(features.ElasticJobsViaWorkloadSlices) && a.preemptWorkloadSlice != nil {
+				for _, psID := range psIDs {
+					preemptWorkloadRequests := a.preemptWorkloadSlice.TotalRequests[psID]
+
+					// Enforce consistent resource flavor assignment between slices.
+					if originalFlavor := preemptWorkloadRequests.Flavors[rName]; originalFlavor != fName {
+						// Flavor mismatch. Skip further checks for this resource.
+						representativeMode = noFit
+						status.reasons = append(status.reasons, fmt.Sprintf("could not assign %s flavor since the original workload is assigned: %s", fName, originalFlavor))
+						break
+					}
+
+					// Subtract the resource usage of the preempted slice to request only the delta needed.
+					val -= preemptWorkloadRequests.Requests[rName]
+				}
+			}
+
 			resQuota := a.cq.QuotaFor(resources.FlavorResource{Flavor: fName, Resource: rName})
 			// Check considering the flavor usage by previous pod sets.
 			fr := resources.FlavorResource{Flavor: fName, Resource: rName}
diff --git a/pkg/scheduler/scheduler.go b/pkg/scheduler/scheduler.go
index 6046b1424..bca54f904 100644
--- a/pkg/scheduler/scheduler.go
+++ b/pkg/scheduler/scheduler.go
@@ -28,6 +28,7 @@ import (
 	"github.com/go-logr/logr"
 	corev1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/api/meta"
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/tools/record"
@@ -40,6 +41,7 @@ import (
 	kueuealpha "sigs.k8s.io/kueue/apis/kueue/v1alpha1"
 	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta1"
 	"sigs.k8s.io/kueue/pkg/cache"
+	controllerconstants "sigs.k8s.io/kueue/pkg/controller/constants"
 	"sigs.k8s.io/kueue/pkg/features"
 	"sigs.k8s.io/kueue/pkg/metrics"
 	"sigs.k8s.io/kueue/pkg/queue"
@@ -51,6 +53,7 @@ import (
 	"sigs.k8s.io/kueue/pkg/util/routine"
 	"sigs.k8s.io/kueue/pkg/util/wait"
 	"sigs.k8s.io/kueue/pkg/workload"
+	"sigs.k8s.io/kueue/pkg/workloadslicing"
 )
 
 const (
@@ -273,10 +276,16 @@ func (s *Scheduler) schedule(ctx context.Context) wait.SpeedSignal {
 		preemptedWorkloads.Insert(e.preemptionTargets)
 		cq.AddUsage(usage)
 
+		// Filter out the old workload slice from the preemption targets.
+		// The old workload slice is initially included in the preemption targets because it is treated
+		// as a preemptible target during flavor assignment. However, it should be evicted rather than preempted.
+		// Note: it is valid for either or both preemptionTargets and oldWorkloadSlice to be nil.
+		preemptionTargets, oldWorkloadSlice := workloadslicing.FindReplacedSliceTarget(e.Obj, e.preemptionTargets)
+
 		if e.assignment.RepresentativeMode() == flavorassigner.Preempt {
 			// If preemptions are issued, the next attempt should try all the flavors.
 			e.LastAssignment = nil
-			preempted, err := s.preemptor.IssuePreemptions(ctx, &e.Info, e.preemptionTargets)
+			preempted, err := s.preemptor.IssuePreemptions(ctx, &e.Info, preemptionTargets)
 			if err != nil {
 				log.Error(err, "Failed to preempt workloads")
 			}
@@ -298,6 +307,16 @@ func (s *Scheduler) schedule(ctx context.Context) wait.SpeedSignal {
 			s.cache.WaitForPodsReady(ctx)
 			log.V(5).Info("Finished waiting for all admitted workloads to be in the PodsReady condition")
 		}
+
+		// Evict old workload-slice if any. Note: that oldWorkloadSlice is not nil only if
+		// this is a workload-slice enabled workload and there is an old slice to evict.
+		if features.Enabled(features.ElasticJobsViaWorkloadSlices) && oldWorkloadSlice != nil {
+			if err := s.replaceWorkloadSlice(ctx, oldWorkloadSlice.WorkloadInfo.ClusterQueue, e.Obj, oldWorkloadSlice.WorkloadInfo.Obj); err != nil {
+				log.Error(err, "Failed to aggregate workload slice")
+				continue
+			}
+		}
+
 		e.status = nominated
 		if err := s.admit(ctx, e, cq); err != nil {
 			e.inadmissibleMsg = fmt.Sprintf("Failed to admit workload: %v", err)
@@ -462,20 +481,45 @@ func (s *Scheduler) getAssignments(log logr.Logger, wl *workload.Info, snap *cac
 	return assignment, targets
 }
 
+// getInitialAssignments computes the initial resource flavor assignment and any required preemption targets
+// for a workload slice.
+//
+// The function attempts to assign resources to the provided workload slice using the current
+// snapshot of the scheduling state. It proceeds in the following steps:
+//
+//  1. It first checks for any preemptible workload slices that workload may replace, using an annotation-based lookup.
+//  2. It creates a flavor assigner to compute a full assignment scale-adjusted for preemptable workload slice targets
+//     based on either:
+//     - direct fit (no preemption needed), or
+//     - preemption (if needed and possible).
+//  3. If direct assignment isn't possible but preemption is enabled and viable, it includes any additional
+//     preemption targets obtained through the configured preemptor.
+//  4. If partial admission is enabled and the workload allows it, the function attempts to reduce pod counts
+//     across PodSets to find an assignable configuration—again checking for preemption if needed.
+//
+// Returns:
+//   - A flavorassigner.Assignment representing the selected (possibly reduced) flavor allocation.
+//   - A slice of preemption targets, which may include both explicitly annotated slices and those
+//     identified during scheduling.
+//
+// If no valid assignment can be made, returns the original full assignment with no preemption targets.
 func (s *Scheduler) getInitialAssignments(log logr.Logger, wl *workload.Info, snap *cache.Snapshot) (flavorassigner.Assignment, []*preemption.Target) {
 	cq := snap.ClusterQueue(wl.ClusterQueue)
-	flvAssigner := flavorassigner.New(wl, cq, snap.ResourceFlavors, s.fairSharing.Enable, preemption.NewOracle(s.preemptor, snap))
+
+	preemptionTargets, replaceableWorkloadSlice := workloadslicing.ReplacedWorkloadSlice(wl, snap)
+
+	flvAssigner := flavorassigner.New(wl, cq, snap.ResourceFlavors, s.fairSharing.Enable, preemption.NewOracle(s.preemptor, snap), replaceableWorkloadSlice)
 	fullAssignment := flvAssigner.Assign(log, nil)
 
 	arm := fullAssignment.RepresentativeMode()
 	if arm == flavorassigner.Fit {
-		return fullAssignment, nil
+		return fullAssignment, preemptionTargets
 	}
 
 	if arm == flavorassigner.Preempt {
 		faPreemptionTargets := s.preemptor.GetTargets(log, *wl, fullAssignment, snap)
 		if len(faPreemptionTargets) > 0 {
-			return fullAssignment, faPreemptionTargets
+			return fullAssignment, append(preemptionTargets, faPreemptionTargets...)
 		}
 	}
 
@@ -496,7 +540,7 @@ func (s *Scheduler) getInitialAssignments(log logr.Logger, wl *workload.Info, sn
 			return nil, false
 		})
 		if pa, found := reducer.Search(); found {
-			return pa.assignment, pa.preemptionTargets
+			return pa.assignment, append(preemptionTargets, pa.preemptionTargets...)
 		}
 	}
 	return fullAssignment, nil
@@ -751,3 +795,32 @@ func (s *Scheduler) recordWorkloadAdmissionEvents(newWorkload, originalWorkload
 		}
 	}
 }
+
+// replaceWorkloadSlice handles the replacement of a workload slice by deactivating the old slice and
+// marking it as finished. It logs the replacement operation, records an event, and reports metrics.
+//
+// This function performs the following steps:
+//  1. Checks if the old workload slice is already finished by inspecting the "Finished" condition in its status.
+//     If the slice is already finished, the function logs a message and returns early.
+//  2. If the old slice is not finished, it deactivates the old slice and marks it with a "Finished" condition,
+//     indicating that the slice was replaced to accommodate the new workload slice.
+//  3. The function logs details about the replacement, including the reason for the removal and the associated message.
+//  4. An event is recorded for the old slice to indicate that the slice was aggregated (replaced) by the new slice.
+//  5. The function reports metrics for the aggregation of workload slices for the old queue.
+func (s *Scheduler) replaceWorkloadSlice(ctx context.Context, oldQueue kueue.ClusterQueueReference, newSlice, oldSlice *kueue.Workload) error {
+	log := ctrl.LoggerFrom(ctx)
+	if meta.IsStatusConditionTrue(oldSlice.Status.Conditions, kueue.WorkloadFinished) {
+		log.V(3).Info("Workload slice already finished", "old-slice", klog.KObj(oldSlice), "new-slice", klog.KObj(newSlice))
+		return nil
+	}
+	reason := kueue.WorkloadSliceReplaced
+	message := fmt.Sprintf("Replaced to accommodate a workload (UID: %s, JobUID: %s) due to workload slice aggregation", newSlice.UID, newSlice.Labels[controllerconstants.JobUIDLabel])
+	if err := workloadslicing.Finish(ctx, s.client, oldSlice, reason, message); err != nil {
+		return fmt.Errorf("failed to replace workload slice: %w", err)
+	}
+
+	log.V(3).Info("Replaced", "old slice", klog.KObj(oldSlice), "new slice", klog.KObj(newSlice), "reason", reason, "message", message, "old-queue", klog.KRef("", string(oldQueue)))
+	s.recorder.Eventf(oldSlice, corev1.EventTypeNormal, reason, message)
+	metrics.ReportReplacedWorkloadSlices(oldQueue)
+	return nil
+}
diff --git a/pkg/webhooks/workload_webhook.go b/pkg/webhooks/workload_webhook.go
index 04f5a32dd..bb2e8a50e 100644
--- a/pkg/webhooks/workload_webhook.go
+++ b/pkg/webhooks/workload_webhook.go
@@ -19,6 +19,7 @@ package webhooks
 import (
 	"context"
 	"fmt"
+	"strconv"
 
 	corev1 "k8s.io/api/core/v1"
 	apivalidation "k8s.io/apimachinery/pkg/api/validation"
@@ -272,7 +273,7 @@ func ValidateWorkloadUpdate(newObj, oldObj *kueue.Workload) field.ErrorList {
 	allErrs = append(allErrs, ValidateWorkload(newObj)...)
 
 	if workload.HasQuotaReservation(oldObj) {
-		allErrs = append(allErrs, apivalidation.ValidateImmutableField(newObj.Spec.PodSets, oldObj.Spec.PodSets, specPath.Child("podSets"))...)
+		allErrs = append(allErrs, validateImmutablePodSets(newObj.Spec.PodSets, oldObj.Spec.PodSets, specPath.Child("podSets"))...)
 	}
 	if workload.HasQuotaReservation(newObj) && workload.HasQuotaReservation(oldObj) {
 		allErrs = append(allErrs, validateReclaimablePodsUpdate(newObj, oldObj, field.NewPath("status", "reclaimablePods"))...)
@@ -342,3 +343,26 @@ func validateReclaimablePodsUpdate(newObj, oldObj *kueue.Workload, basePath *fie
 	}
 	return ret
 }
+
+// validateImmutablePodSet helper to validate PodSet immutability on all fields but PodSet.Count.
+func validateImmutablePodSet(new, old kueue.PodSet, path *field.Path) field.ErrorList {
+	if features.Enabled(features.ElasticJobsViaWorkloadSlices) && new.Count < old.Count {
+		// Allow scale-down for elastic jobs.
+		new.Count = old.Count
+	}
+	return apivalidation.ValidateImmutableField(new, old, path)
+}
+
+// validateImmutablePodSets helper to validate PodSet lists for immutability on all fields but PodSet.Count.
+func validateImmutablePodSets(new, old []kueue.PodSet, path *field.Path) field.ErrorList {
+	if len(new) != len(old) {
+		return field.ErrorList{field.Invalid(path, new, apivalidation.FieldImmutableErrorMsg)}
+	}
+	allErrs := make(field.ErrorList, 0, len(new))
+	for i := range new {
+		if errs := validateImmutablePodSet(new[i], old[i], path.Child(strconv.Itoa(i))); len(errs) > 0 {
+			allErrs = append(allErrs, errs...)
+		}
+	}
+	return allErrs
+}
diff --git a/pkg/workload/podsetscounts.go b/pkg/workload/podsetscounts.go
new file mode 100644
index 000000000..0c430e12c
--- /dev/null
+++ b/pkg/workload/podsetscounts.go
@@ -0,0 +1,89 @@
+/*
+Copyright The Kubernetes Authors.
+
+Licensed 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 workload
+
+import (
+	"maps"
+
+	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta1"
+	utilslices "sigs.k8s.io/kueue/pkg/util/slices"
+)
+
+// PodSetsCounts represents a mapping between PodSet references and their desired replica counts.
+//
+// Each key in the map corresponds to a unique kueue.PodSetReference, and the associated value
+// is the number of replicas (int32) requested for that PodSet. This type is used to facilitate
+// comparisons, updates, and compatibility checks between different sets of PodSet definitions
+// within jobs and workloads.
+type PodSetsCounts map[kueue.PodSetReference]int32
+
+// HasSamePodSetKeys returns true if both PodSetsCounts have exactly the same PodSet keys,
+// ignoring their replica count values.
+func (c PodSetsCounts) HasSamePodSetKeys(in PodSetsCounts) bool {
+	return maps.EqualFunc(c, in, func(_ int32, _ int32) bool { return true })
+}
+
+// EqualTo returns true if this PodSetsCount is identical (on keys and values) to
+// the incoming PodSetsCount.
+func (c PodSetsCounts) EqualTo(in PodSetsCounts) bool {
+	return maps.Equal(c, in)
+}
+
+// HasFewerReplicasThan returns true if there is at least one pod set in the current set (`c`)
+// whose count is lower than the corresponding entry in the input set (`in`).
+//
+// Note: This function does not consider keys that exist in current set (`c`) but not the other (`in`).
+func (c PodSetsCounts) HasFewerReplicasThan(in PodSetsCounts) bool {
+	for key, cCount := range c {
+		if inCount, ok := in[key]; ok && cCount < inCount {
+			return true
+		}
+	}
+	return false
+}
+
+// ExtractPodSetCounts builds a PodSetsCounts map from a list of PodSets.
+// Each entry maps PodSet name to its replica count.
+func ExtractPodSetCounts(podSets []kueue.PodSet) PodSetsCounts {
+	return utilslices.ToMap(podSets, func(i int) (kueue.PodSetReference, int32) {
+		return podSets[i].Name, podSets[i].Count
+	})
+}
+
+// ExtractPodSetCountsFromWorkload returns a PodSetsCounts map derived from the provided Workload.
+//
+// Important: This function assumes the Workload is not nil. It does not perform a nil check,
+// and calling it with a nil Workload will result in a panic.
+func ExtractPodSetCountsFromWorkload(wl *kueue.Workload) PodSetsCounts {
+	return ExtractPodSetCounts(wl.Spec.PodSets)
+}
+
+// ApplyPodSetCounts updates the count values of a Workload's PodSets based on the provided counts.
+//
+// This function only updates existing PodSets within the Workload. It ignores:
+//   - PodSet names present in the counts map but not found in the Workload.
+//   - PodSets in the Workload that are not present in the counts map.
+//
+// Important: This function assumes the Workload is not nil. It does not perform a nil check,
+// and calling it with a nil Workload will result in a panic.
+func ApplyPodSetCounts(wl *kueue.Workload, counts PodSetsCounts) {
+	for i := range wl.Spec.PodSets {
+		if count, found := counts[wl.Spec.PodSets[i].Name]; found {
+			wl.Spec.PodSets[i].Count = count
+		}
+	}
+}
diff --git a/pkg/workloadslicing/workloadslicing.go b/pkg/workloadslicing/workloadslicing.go
new file mode 100644
index 000000000..3fd30dc18
--- /dev/null
+++ b/pkg/workloadslicing/workloadslicing.go
@@ -0,0 +1,319 @@
+/*
+Copyright The Kubernetes Authors.
+
+Licensed 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 workloadslicing contains definitions to support workload slicing functionality.
+package workloadslicing
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"slices"
+	"sort"
+
+	corev1 "k8s.io/api/core/v1"
+	apimeta "k8s.io/apimachinery/pkg/api/meta"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+
+	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta1"
+	"sigs.k8s.io/kueue/pkg/cache"
+	"sigs.k8s.io/kueue/pkg/controller/core/indexer"
+	"sigs.k8s.io/kueue/pkg/features"
+	"sigs.k8s.io/kueue/pkg/scheduler/preemption"
+	clientutil "sigs.k8s.io/kueue/pkg/util/client"
+	"sigs.k8s.io/kueue/pkg/util/pod"
+	"sigs.k8s.io/kueue/pkg/workload"
+)
+
+// Workload slicing refers to a specialized Kueue feature designed to support workload scaling up.
+// A workload slice represents a logical grouping of two workloads:
+// the "old slice" (preemptible workload) and the "new slice" (preemptor workload).
+
+// Workload slicing can be enabled for a specific integration job instance,
+// provided that the integration supports workload slicing.
+const (
+	// EnabledAnnotationKey refers to the annotation key present on Job's that support
+	// workload slicing.
+	EnabledAnnotationKey = "kueue.x-k8s.io/elastic-job"
+	// EnabledAnnotationValue refers to the annotation value. To enable
+	// workload slicing for a given job, we match both annotation key and value.
+	EnabledAnnotationValue = "true"
+)
+
+// Enabled returns true if a provided job has slicing annotation indicator.
+// Indicator matched on both annotation key and value.
+func Enabled(object metav1.Object) bool {
+	if object == nil {
+		return false
+	}
+	return object.GetAnnotations()[EnabledAnnotationKey] == EnabledAnnotationValue
+}
+
+const (
+	// WorkloadSliceReplacementFor is the annotation key used to capture an "old" workload slice key
+	// that will be preempted by the "new", e.g., this workload slice with annotation.
+	WorkloadSliceReplacementFor = "kueue.x-k8s.io/workload-slice-replacement-for"
+)
+
+// ReplacementForKey returns a value for workload "WorkloadSliceReplacementFor" annotation
+// key if this workload was annotated with such, otherwise, returns an empty string.
+func ReplacementForKey(wl *kueue.Workload) *workload.Reference {
+	annotations := wl.GetAnnotations()
+	if len(annotations) == 0 {
+		return nil
+	}
+	key, found := annotations[WorkloadSliceReplacementFor]
+	if !found {
+		return nil
+	}
+	ref := workload.Reference(key)
+	return &ref
+}
+
+// Finish updates the status of a workload slice by applying the "Finished" condition.
+// The function checks if the "Finished" condition is already applied, and if so, does nothing (NOOP).
+// If the "Finished" condition is not present, it applies the condition with the provided `reason` and `message`.
+//
+// This function performs the following:
+// 1. It checks if the "Finished" condition is already applied. If true, it returns immediately, doing nothing.
+// 2. If the "Finished" condition is not set, it patches the workload slice's status to add the "Finished" condition.
+// 3. If the patch fails, it returns an error.
+func Finish(ctx context.Context, clnt client.Client, workloadSlice *kueue.Workload, reason, message string) error {
+	// NOOP if the workload already has "Finished" condition (irrespective of reason and message values).
+	if apimeta.IsStatusConditionTrue(workloadSlice.Status.Conditions, kueue.WorkloadFinished) {
+		return nil
+	}
+	if err := clientutil.PatchStatus(ctx, clnt, workloadSlice, func() (bool, error) {
+		return apimeta.SetStatusCondition(&workloadSlice.Status.Conditions, metav1.Condition{
+			Type:    kueue.WorkloadFinished,
+			Status:  metav1.ConditionTrue,
+			Reason:  reason,
+			Message: message,
+		}), nil
+	}); err != nil {
+		return fmt.Errorf("failed to patch workload slice status: %w", err)
+	}
+	return nil
+}
+
+// FindNotFinishedWorkloads returns a sorted list of workloads "owned by" the provided job object/gvk combination and
+// without "Finished" condition with status = "True".
+func FindNotFinishedWorkloads(ctx context.Context, clnt client.Client, jobObject client.Object, jobObjectGVK schema.GroupVersionKind) ([]kueue.Workload, error) {
+	list := &kueue.WorkloadList{}
+	if err := clnt.List(ctx, list, client.InNamespace(jobObject.GetNamespace()), client.MatchingFields{fmt.Sprintf(".metadata.ownerReferences[%s.%s]", jobObjectGVK.Group, jobObjectGVK.Kind): jobObject.GetName()}); err != nil {
+		return nil, err
+	}
+
+	// Sort workloads by creation timestamp - oldest first.
+	sort.Slice(list.Items, func(i, j int) bool {
+		return list.Items[i].CreationTimestamp.Before(&list.Items[j].CreationTimestamp)
+	})
+
+	// Filter out workloads with activated "Finished" condition.
+	return slices.DeleteFunc(list.Items, func(w kueue.Workload) bool {
+		return apimeta.IsStatusConditionTrue(w.Status.Conditions, kueue.WorkloadFinished)
+	}), nil
+}
+
+// EnsureWorkloadSlices processes the Job object and returns the appropriate workload slice.
+//
+// Returns:
+// - *Workload, true, nil: when a compatible workload exists or a new slice is needed.
+// - nil, false, nil: when an incompatible workload exists and no update is performed.
+// - error: on failure to fetch, update, or deactivate a workload slice.
+func EnsureWorkloadSlices(ctx context.Context, clnt client.Client, jobPodSets []kueue.PodSet, jobObject client.Object, jobObjectGVK schema.GroupVersionKind) (*kueue.Workload, bool, error) {
+	jobPodSetsCounts := workload.ExtractPodSetCounts(jobPodSets)
+
+	workloads, err := FindNotFinishedWorkloads(ctx, clnt, jobObject, jobObjectGVK)
+	if err != nil {
+		return nil, true, fmt.Errorf("failed to find active workload slices: %w", err)
+	}
+
+	switch len(workloads) {
+	case 0:
+		// No existing slices found — new slice should be created.
+		return nil, true, nil
+
+	case 1:
+		// A single active workload was found.
+		wl := &workloads[0]
+		wlPodSetsCounts := workload.ExtractPodSetCountsFromWorkload(wl)
+
+		// Check if pod sets are structurally compatible (same number and names).
+		if !jobPodSetsCounts.HasSamePodSetKeys(wlPodSetsCounts) {
+			return nil, false, nil
+		}
+
+		// If counts match, return the existing workload slice.
+		if jobPodSetsCounts.EqualTo(wlPodSetsCounts) {
+			return wl, true, nil
+		}
+
+		// Allow updating the existing slice if:
+		// a. It hasn't been admitted (no quota reserved), or
+		// b. It's a scale-down event.
+		if !workload.HasQuotaReservation(wl) || jobPodSetsCounts.HasFewerReplicasThan(wlPodSetsCounts) {
+			workload.ApplyPodSetCounts(wl, jobPodSetsCounts)
+			if err := clnt.Update(ctx, wl); err != nil {
+				return nil, true, fmt.Errorf("failed to update workload pod set counts: %w", err)
+			}
+			return wl, true, nil
+		}
+
+		// Scale-up on admitted workload → create a new slice.
+		return nil, true, nil
+
+	case 2:
+		// Two active workload slices detected.
+		//
+		// This transient state typically occurs when a new slice has been created,
+		// and the old slice is pending deactivation. The system should resolve this
+		// by deactivating the old slice, after which processing will continue under "case #1".
+
+		// There is also an edge-case when the old slice was preempted/evicted by the scheduler
+		// to make the room for other (than new slice) workload, which would result in two
+		// "pending" workloads. In such case - it is safe to deactivate the old slice.
+		oldWorkload := workloads[0]
+		if !workload.HasQuotaReservation(&oldWorkload) {
+			if err := Finish(ctx, clnt, &oldWorkload, kueue.WorkloadFinishedReasonOutOfSync, "The workload slice is out of sync with its parent job"); err != nil {
+				return nil, true, err
+			}
+		}
+
+		// We consider the new workload slice only when evaluating against the incoming job (pod sets).
+		newWorkload := workloads[1]
+		newCounts := workload.ExtractPodSetCountsFromWorkload(&newWorkload)
+
+		// Check if new workload and job's pod sets are compatible (have the same keys and keys count)
+		if !jobPodSetsCounts.HasSamePodSetKeys(newCounts) {
+			return nil, false, nil
+		}
+
+		// Return an error if the new workload has a reserved quota.
+		//
+		// A combination of a new workload with a reserved quota and an active old workload is considered an anomaly.
+		// This condition may indicate a race condition or external interference. Specifically, a new workload slice
+		// should never gain a quota reservation without the prior finalization of the old slice.
+		if workload.HasQuotaReservation(&newWorkload) {
+			return nil, true, errors.New("unexpected combination of old and new workload slices with reserved quota")
+		}
+
+		// If the pod set counts match, return new workload slice.
+		if jobPodSetsCounts.EqualTo(newCounts) {
+			return &newWorkload, true, nil
+		}
+
+		// Update new workload slice.
+		workload.ApplyPodSetCounts(&newWorkload, jobPodSetsCounts)
+		if err := clnt.Update(ctx, &newWorkload); err != nil {
+			return nil, true, fmt.Errorf("failed to update workload pod set counts: %w", err)
+		}
+		return &newWorkload, true, nil
+	default:
+		// More than two matching slices is unexpected and considered an error.
+		return nil, true, fmt.Errorf("unexpected number of matching workload slices: %d", len(workloads))
+	}
+}
+
+// StartWorkloadSlicePods identifies pods associated with the provided parent object
+// that are gated by the ElasticJobSchedulingGate scheduling gate and removes this gate,
+// allowing them to be considered for scheduling.
+//
+// This function performs the following steps:
+// 1. Lists all pods in the same namespace with an OwnerReference UID matching the parent object.
+// 2. For each pod, removes the ElasticJobSchedulingGate scheduling gate if present.
+//
+// Returns:
+// - An error if any of the operations fail; otherwise, nil.
+func StartWorkloadSlicePods(ctx context.Context, clnt client.Client, object client.Object) error {
+	list := &corev1.PodList{}
+	if err := clnt.List(ctx, list, client.InNamespace(object.GetNamespace()), client.MatchingFields{indexer.OwnerReferenceUID: string(object.GetUID())}); err != nil {
+		return fmt.Errorf("failed to list job pods: %w", err)
+	}
+	for i := range list.Items {
+		if err := clientutil.Patch(ctx, clnt, &list.Items[i], true, func() (bool, error) {
+			return pod.Ungate(&list.Items[i], kueue.ElasticJobSchedulingGate), nil
+		}); err != nil {
+			return fmt.Errorf("failed to patch pod: %w", err)
+		}
+	}
+	return nil
+}
+
+// ReplacedWorkloadSlice returns the replacement workload slice for the given workload `wl`
+// if it has been marked as replaced by another slice. This is determined using the
+// `ReplacementForKey` annotation or metadata.
+//
+// It returns a slice containing the corresponding preemption.Target for the replacement,
+// as well as a pointer to the replacement workload.Info object.
+//
+// Returns nil if:
+// - The ElasticJobsViaWorkloadSlices feature gate is not enabled
+// - The input workload or snapshot is nil
+// - The workload has no replacement slice key
+// - The referenced replacement workload is not found in the ClusterQueue snapshot
+func ReplacedWorkloadSlice(wl *workload.Info, snap *cache.Snapshot) ([]*preemption.Target, *workload.Info) {
+	if !features.Enabled(features.ElasticJobsViaWorkloadSlices) || wl == nil || snap == nil {
+		return nil, nil
+	}
+
+	sliceKey := ReplacementForKey(wl.Obj)
+	if sliceKey == nil {
+		return nil, nil
+	}
+
+	// Retrieve cluster queue.
+	queue := snap.ClusterQueue(wl.ClusterQueue)
+	if queue == nil {
+		return nil, nil
+	}
+
+	replaced, found := queue.Workloads[*sliceKey]
+	if !found {
+		return nil, nil
+	}
+
+	return []*preemption.Target{{WorkloadInfo: replaced}}, replaced
+}
+
+// FindReplacedSliceTarget identifies and removes a preempted workload slice target from the given list of targets.
+// The function checks if Elastic Jobs via Workload Slices feature is enabled and if so, attempts to find a matching
+// workload slice in the target list for the provided preemptor. If a matching slice is found, it is removed from the list
+// and returned alongside the original target list.
+//
+// This function performs the following:
+// 1. It checks if the feature `ElasticJobsViaWorkloadSlices` is enabled. If not, it returns the target list unchanged with `nil` as the second return value.
+// 2. It generates a replacement key for the preemptor workload slice. If no replacement key is found, it returns the original target list unchanged with `nil`.
+// 3. It iterates over the list of preemption targets, looking for a target that matches the replacement key.
+// 4. If a matching target is found, it removes it from the list and returns the updated list and the removed target.
+// 5. If no matching target is found, it returns the original target list and `nil`.
+func FindReplacedSliceTarget(preemptor *kueue.Workload, targets []*preemption.Target) ([]*preemption.Target, *preemption.Target) {
+	if !features.Enabled(features.ElasticJobsViaWorkloadSlices) {
+		return targets, nil
+	}
+	sliceKey := ReplacementForKey(preemptor)
+	if sliceKey == nil {
+		return targets, nil
+	}
+	for i := range targets {
+		if *sliceKey == workload.Key(targets[i].WorkloadInfo.Obj) {
+			return append(targets[:i], targets[i+1:]...), targets[i]
+		}
+	}
+	return targets, nil
+}
diff --git a/site/content/zh-CN/docs/reference/kubectl-kueue/commands/kueuectl_version/_index.md b/site/content/zh-CN/docs/reference/kubectl-kueue/commands/kueuectl_version/_index.md
index c9a19a529..5d2bcfbea 100644
--- a/site/content/zh-CN/docs/reference/kubectl-kueue/commands/kueuectl_version/_index.md
+++ b/site/content/zh-CN/docs/reference/kubectl-kueue/commands/kueuectl_version/_index.md
@@ -5,30 +5,24 @@ auto_generated: true
 no_list: true
 ---
 
-<!--
-The file is auto-generated from the Go source code of the component using the
-[generator](https://github.com/kubernetes-sigs/kueue/tree/main/cmd/kueuectl-docs).
--->
+## 概要 {#synopsis}
 
-## Synopsis
-
-
-Prints the client version and the kueue controller manager image, if installed
+打印客户端版本和 Kueue 控制器管理器镜像（如果已安装）
 
 ```
 kueuectl version
 ```
 
 
-## Examples
+## 示例 {#examples}
 
 ```
-  # Prints the client version and the kueue controller manager image, if installed
+  # 打印客户端版本和 Kueue 控制器管理器镜像（如果已安装）
   kueuectl version
 ```
 
 
-## Options
+## 选项 {#options}
 
 
 <table style="width: 100%; table-layout: fixed;">
@@ -43,7 +37,7 @@ kueuectl version
     <tr>
         <td></td>
         <td style="line-height: 130%; word-wrap: break-word;">
-            <p>help for version</p>
+            <p>version 命令的帮助信息</p>
         </td>
     </tr>
     </tbody>
@@ -51,7 +45,7 @@ kueuectl version
 
 
 
-## Options inherited from parent commands
+## 从父命令继承的选项 {#options_inherited_from_parent_commands}
 <table style="width: 100%; table-layout: fixed;">
     <colgroup>
         <col span="1" style="width: 10px;" />
@@ -64,7 +58,7 @@ kueuectl version
     <tr>
         <td></td>
         <td style="line-height: 130%; word-wrap: break-word;">
-            <p>Username to impersonate for the operation. User could be a regular user or a service account in a namespace.</p>
+            <p>为操作模拟的用户名。用户可以是普通用户或命名空间中的服务账户。</p>
         </td>
     </tr>
     <tr>
@@ -73,7 +67,7 @@ kueuectl version
     <tr>
         <td></td>
         <td style="line-height: 130%; word-wrap: break-word;">
-            <p>Group to impersonate for the operation, this flag can be repeated to specify multiple groups.</p>
+            <p>为操作模拟的组，此标志可以重复指定多个组。</p>
         </td>
     </tr>
     <tr>
@@ -82,16 +76,16 @@ kueuectl version
     <tr>
         <td></td>
         <td style="line-height: 130%; word-wrap: break-word;">
-            <p>UID to impersonate for the operation.</p>
+            <p>为操作模拟的 UID。</p>
         </td>
     </tr>
     <tr>
-        <td colspan="2">--cache-dir string&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;Default: &#34;$HOME/.kube/cache&#34;</td>
+        <td colspan="2">--cache-dir string&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;默认值: &#34;$HOME/.kube/cache&#34;</td>
     </tr>
     <tr>
         <td></td>
         <td style="line-height: 130%; word-wrap: break-word;">
-            <p>Default cache directory</p>
+            <p>默认缓存目录</p>
         </td>
     </tr>
     <tr>
@@ -100,7 +94,7 @@ kueuectl version
     <tr>
         <td></td>
         <td style="line-height: 130%; word-wrap: break-word;">
-            <p>Path to a cert file for the certificate authority</p>
+            <p>证书颁发机构证书文件的路径</p>
         </td>
     </tr>
     <tr>
@@ -109,7 +103,7 @@ kueuectl version
     <tr>
         <td></td>
         <td style="line-height: 130%; word-wrap: break-word;">
-            <p>Path to a client certificate file for TLS</p>
+            <p>TLS 客户端证书文件的路径</p>
         </td>
     </tr>
     <tr>
@@ -118,7 +112,7 @@ kueuectl version
     <tr>
         <td></td>
         <td style="line-height: 130%; word-wrap: break-word;">
-            <p>Path to a client key file for TLS</p>
+            <p>TLS 客户端密钥文件的路径</p>
         </td>
     </tr>
     <tr>
@@ -127,7 +121,7 @@ kueuectl version
     <tr>
         <td></td>
         <td style="line-height: 130%; word-wrap: break-word;">
-            <p>The name of the kubeconfig cluster to use</p>
+            <p>要使用的 kubeconfig 集群的名称</p>
         </td>
     </tr>
     <tr>
@@ -136,7 +130,7 @@ kueuectl version
     <tr>
         <td></td>
         <td style="line-height: 130%; word-wrap: break-word;">
-            <p>The name of the kubeconfig context to use</p>
+            <p>要使用的 kubeconfig 上下文的名称</p>
         </td>
     </tr>
     <tr>
@@ -145,7 +139,7 @@ kueuectl version
     <tr>
         <td></td>
         <td style="line-height: 130%; word-wrap: break-word;">
-            <p>If true, opt-out of response compression for all requests to the server</p>
+            <p>如果为 true，则选择退出对所有服务器请求的响应压缩</p>
         </td>
     </tr>
     <tr>
@@ -154,7 +148,7 @@ kueuectl version
     <tr>
         <td></td>
         <td style="line-height: 130%; word-wrap: break-word;">
-            <p>If true, the server&#39;s certificate will not be checked for validity. This will make your HTTPS connections insecure</p>
+            <p>如果为 true，将不会检查服务器证书的有效性。这会使您的 HTTPS 连接不安全</p>
         </td>
     </tr>
     <tr>
@@ -163,7 +157,7 @@ kueuectl version
     <tr>
         <td></td>
         <td style="line-height: 130%; word-wrap: break-word;">
-            <p>Path to the kubeconfig file to use for CLI requests.</p>
+            <p>用于 CLI 请求的 kubeconfig 文件的路径。</p>
         </td>
     </tr>
     <tr>
@@ -172,16 +166,16 @@ kueuectl version
     <tr>
         <td></td>
         <td style="line-height: 130%; word-wrap: break-word;">
-            <p>If present, the namespace scope for this CLI request</p>
+            <p>如果存在，则为此次 CLI 请求的命名空间范围</p>
         </td>
     </tr>
     <tr>
-        <td colspan="2">--request-timeout string&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;Default: &#34;0&#34;</td>
+        <td colspan="2">--request-timeout string&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;默认值: &#34;0&#34;</td>
     </tr>
     <tr>
         <td></td>
         <td style="line-height: 130%; word-wrap: break-word;">
-            <p>The length of time to wait before giving up on a single server request. Non-zero values should contain a corresponding time unit (e.g. 1s, 2m, 3h). A value of zero means don&#39;t timeout requests.</p>
+            <p>在放弃单个服务器请求之前等待的时间长度。非零值应包含相应的时间单位（例如 1s、2m、3h）。零值表示不超时请求。</p>
         </td>
     </tr>
     <tr>
@@ -190,7 +184,7 @@ kueuectl version
     <tr>
         <td></td>
         <td style="line-height: 130%; word-wrap: break-word;">
-            <p>The address and port of the Kubernetes API server</p>
+            <p>Kubernetes API 服务器的地址和端口</p>
         </td>
     </tr>
     <tr>
@@ -199,7 +193,7 @@ kueuectl version
     <tr>
         <td></td>
         <td style="line-height: 130%; word-wrap: break-word;">
-            <p>Server name to use for server certificate validation. If it is not provided, the hostname used to contact the server is used</p>
+            <p>用于服务器证书验证的服务器名称。如果未提供，则使用用于联系服务器的主机名</p>
         </td>
     </tr>
     <tr>
@@ -208,7 +202,7 @@ kueuectl version
     <tr>
         <td></td>
         <td style="line-height: 130%; word-wrap: break-word;">
-            <p>Bearer token for authentication to the API server</p>
+            <p>用于 API 服务器身份验证的持有者令牌</p>
         </td>
     </tr>
     <tr>
@@ -217,7 +211,7 @@ kueuectl version
     <tr>
         <td></td>
         <td style="line-height: 130%; word-wrap: break-word;">
-            <p>The name of the kubeconfig user to use</p>
+            <p>要使用的 kubeconfig 用户的名称</p>
         </td>
     </tr>
     </tbody>
@@ -225,7 +219,7 @@ kueuectl version
 
 
 
-## See Also
+## 另请参阅 {#see_also}
 
-* [kueuectl](../kueuectl/)	 - Controls Kueue queueing manager
+* [kueuectl](../kueuectl/) - 控制 Kueue 队列管理器
 



Test patch:
diff --git a/pkg/controller/jobframework/reconciler_inpackage_test.go b/pkg/controller/jobframework/reconciler_inpackage_test.go
new file mode 100644
index 000000000..4e2aa813d
--- /dev/null
+++ b/pkg/controller/jobframework/reconciler_inpackage_test.go
@@ -0,0 +1,285 @@
+/*
+Copyright The Kubernetes Authors.
+
+Licensed 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 jobframework
+
+import (
+	"context"
+	"testing"
+
+	"github.com/google/go-cmp/cmp"
+	batchv1 "k8s.io/api/batch/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/client/fake"
+
+	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta1"
+	"sigs.k8s.io/kueue/pkg/features"
+	"sigs.k8s.io/kueue/pkg/podset"
+	"sigs.k8s.io/kueue/pkg/workload"
+	"sigs.k8s.io/kueue/pkg/workloadslicing"
+)
+
+var testJobGVK = batchv1.SchemeGroupVersion.WithKind("Job")
+
+type mockJob struct {
+	object       *batchv1.Job
+	suspended    bool
+	podSets      []kueue.PodSet
+	podSetsError error
+	gvk          schema.GroupVersionKind
+}
+
+func (j *mockJob) Object() client.Object {
+	return j.object
+}
+
+func (j *mockJob) IsSuspended() bool {
+	return j.suspended
+}
+
+func (j *mockJob) Suspend() {}
+
+func (j *mockJob) RunWithPodSetsInfo(_ []podset.PodSetInfo) error {
+	panic("implement me")
+}
+
+func (j *mockJob) RestorePodSetsInfo(_ []podset.PodSetInfo) bool {
+	panic("implement me")
+}
+
+func (j *mockJob) Finished() (message string, success, finished bool) {
+	panic("implement me")
+}
+
+func (j *mockJob) PodSets() ([]kueue.PodSet, error) {
+	return j.podSets, j.podSetsError
+}
+
+func (j *mockJob) IsActive() bool {
+	panic("implement me")
+}
+
+func (j *mockJob) PodsReady() bool {
+	panic("implement me")
+}
+
+func (j *mockJob) GVK() schema.GroupVersionKind { return j.gvk }
+
+// testJobObject helper returns a workload-slice-enabled v1.Job instance.
+func testJobObject(optIn bool) *batchv1.Job {
+	job := &batchv1.Job{}
+	if optIn {
+		metav1.SetMetaDataAnnotation(&job.ObjectMeta, workloadslicing.EnabledAnnotationKey, workloadslicing.EnabledAnnotationValue)
+	}
+	return job
+}
+
+// testScheme helper returns runtime.Scheme with registered types needed for unit-tests.
+func testScheme() *runtime.Scheme {
+	sch := runtime.NewScheme()
+	_ = batchv1.AddToScheme(sch)
+	_ = kueue.AddToScheme(sch)
+	return sch
+}
+
+// testClient helper returns a fake.ClientBuilder initialized with expected Scheme and Index.
+func testClient() *fake.ClientBuilder {
+	return fake.NewClientBuilder().
+		WithScheme(testScheme()).
+		WithIndex(&kueue.Workload{}, OwnerReferenceIndexKey(testJobGVK), func(object client.Object) []string {
+			wl, ok := object.(*kueue.Workload)
+			if !ok || len(wl.OwnerReferences) == 0 {
+				return nil
+			}
+			owners := make([]string, 0, len(wl.OwnerReferences))
+			for i := range wl.OwnerReferences {
+				owner := &wl.OwnerReferences[i]
+				if owner.Kind == testJobGVK.Kind && owner.APIVersion == testJobGVK.GroupVersion().String() {
+					owners = append(owners, owner.Name)
+				}
+			}
+			return owners
+		})
+}
+
+// testWorkload helper creates a workload instance for the provided generic job.
+func testWorkload(t *testing.T, job GenericJob) *kueue.Workload {
+	t.Helper()
+	jobObject := job.Object()
+	wl := NewWorkload(GetWorkloadNameForOwnerWithGVKAndGeneration(jobObject.GetName(), jobObject.GetUID(), job.GVK(), jobObject.GetGeneration()), jobObject, nil, nil)
+	if err := ctrl.SetControllerReference(jobObject, wl, testScheme()); err != nil {
+		t.Errorf("unexpected error assigning workload controller reference: %v", err)
+		return nil
+	}
+	return wl
+}
+
+func Test_prepareWorkloadSlice(t *testing.T) {
+	type args struct {
+		ctx  context.Context
+		clnt client.Client
+		job  GenericJob
+		wl   *kueue.Workload
+	}
+	type want struct {
+		err      bool
+		workload *kueue.Workload
+	}
+	testJob := func(generation int64) GenericJob {
+		job := testJobObject(true)
+		job.Generation = generation
+		return &mockJob{
+			object: job,
+			gvk:    testJobGVK,
+		}
+	}
+	tests := map[string]struct {
+		args args
+		want want
+	}{
+		"FailureRetrievingWorkloads": {
+			args: args{
+				ctx: t.Context(),
+				// Intentionally using a scheme that doesn’t register Kueue types to trigger a failure.
+				clnt: fake.NewClientBuilder().Build(),
+				job: &mockJob{
+					object: testJobObject(true),
+					gvk:    testJobGVK,
+				},
+			},
+			want: want{err: true},
+		},
+		"NoExistingWorkloads": {
+			args: args{
+				ctx:  t.Context(),
+				clnt: testClient().Build(),
+				job:  testJob(1),
+				wl:   testWorkload(t, testJob(1)),
+			},
+			want: want{
+				workload: testWorkload(t, testJob(1)),
+			},
+		},
+		"OneExistingWorkload": {
+			args: args{
+				ctx: t.Context(),
+				clnt: testClient().WithLists(&kueue.WorkloadList{
+					Items: []kueue.Workload{
+						*testWorkload(t, testJob(1)),
+					},
+				}).Build(),
+				job: testJob(2),
+				wl:  testWorkload(t, testJob(2)),
+			},
+			want: want{
+				workload: func() *kueue.Workload {
+					wl := testWorkload(t, testJob(2))
+					metav1.SetMetaDataAnnotation(&wl.ObjectMeta, workloadslicing.WorkloadSliceReplacementFor, string(workload.Key(testWorkload(t, testJob(1)))))
+					return wl
+				}(),
+			},
+		},
+		"MoreThanOneExistingWorkload": {
+			args: args{
+				ctx: t.Context(),
+				clnt: testClient().WithLists(&kueue.WorkloadList{
+					Items: []kueue.Workload{
+						*testWorkload(t, testJob(1)),
+						*testWorkload(t, testJob(2)),
+					},
+				}).Build(),
+				job: testJob(3),
+				wl:  testWorkload(t, testJob(3)),
+			},
+			want: want{
+				err:      true,
+				workload: testWorkload(t, testJob(3)),
+			},
+		},
+	}
+	for name, tt := range tests {
+		t.Run(name, func(t *testing.T) {
+			err := prepareWorkloadSlice(tt.args.ctx, tt.args.clnt, tt.args.job, tt.args.wl)
+			if (err != nil) != tt.want.err {
+				t.Errorf("prepareWorkloadSlice() error = %v, wantErr %v", err, tt.want.err)
+				return
+			}
+			if diff := cmp.Diff(tt.args.wl, tt.want.workload); diff != "" {
+				t.Errorf("prepareWorkloadSlice() workload(-),want(+): %s", diff)
+			}
+		})
+	}
+}
+
+func TestWorkloadSliceEnabled(t *testing.T) {
+	type args struct {
+		job GenericJob
+	}
+	tests := map[string]struct {
+		featureEnabled bool
+		args           args
+		want           bool
+	}{
+		"FeatureIsNotEnabled": {
+			args: args{
+				job: &testGenericJob{
+					Job: &batchv1.Job{},
+				},
+			},
+		},
+		"JobIsNil": {
+			featureEnabled: true,
+			args:           args{},
+		},
+		"NotOptIn": {
+			featureEnabled: true,
+			args: args{
+				job: &testGenericJob{
+					Job: &batchv1.Job{},
+				},
+			},
+		},
+		"OptIn": {
+			featureEnabled: true,
+			args: args{
+				job: &testGenericJob{
+					Job: &batchv1.Job{
+						ObjectMeta: metav1.ObjectMeta{
+							Annotations: map[string]string{workloadslicing.EnabledAnnotationKey: workloadslicing.EnabledAnnotationValue},
+						},
+					},
+				},
+			},
+			want: true,
+		},
+	}
+	for name, tt := range tests {
+		t.Run(name, func(t *testing.T) {
+			if tt.featureEnabled {
+				if err := features.SetEnable(features.ElasticJobsViaWorkloadSlices, true); err != nil {
+					t.Errorf("workloadSliceEnabled() unexpected error enbabling features: %v", err)
+				}
+			}
+			if got := workloadSliceEnabled(tt.args.job); got != tt.want {
+				t.Errorf("workloadSliceEnabled() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
diff --git a/pkg/controller/jobs/job/job_webhook_test.go b/pkg/controller/jobs/job/job_webhook_test.go
index 3faf19d42..35dbde191 100644
--- a/pkg/controller/jobs/job/job_webhook_test.go
+++ b/pkg/controller/jobs/job/job_webhook_test.go
@@ -24,6 +24,7 @@ import (
 	"github.com/google/go-cmp/cmp/cmpopts"
 	kfmpi "github.com/kubeflow/mpi-operator/pkg/apis/kubeflow/v2beta1"
 	batchv1 "k8s.io/api/batch/v1"
+	corev1 "k8s.io/api/core/v1"
 	apivalidation "k8s.io/apimachinery/pkg/api/validation"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
@@ -42,6 +43,7 @@ import (
 	utiltesting "sigs.k8s.io/kueue/pkg/util/testing"
 	testingutil "sigs.k8s.io/kueue/pkg/util/testingjobs/job"
 	testingmpijob "sigs.k8s.io/kueue/pkg/util/testingjobs/mpijob"
+	"sigs.k8s.io/kueue/pkg/workloadslicing"
 
 	// without this only the job framework is registered
 	_ "sigs.k8s.io/kueue/pkg/controller/jobs/mpijob"
@@ -858,3 +860,97 @@ func TestDefault(t *testing.T) {
 		})
 	}
 }
+
+func Test_applyWorkloadSliceSchedulingGate(t *testing.T) {
+	type args struct {
+		job *Job
+	}
+	// Test case matrix covers combinations of:
+	//   - WorkloadSlice feature enabled: true or false
+	//   - Job opt-in via annotation: present or absent
+	tests := map[string]struct {
+		featureEnabled bool
+		args           args
+		want           []corev1.PodSchedulingGate
+	}{
+		"FeatureDisabledAndNotOptIn": {
+			args: args{job: &Job{}},
+		},
+		"FeatureDisabledAndOptIn": {
+			args: args{
+				job: &Job{
+					ObjectMeta: metav1.ObjectMeta{
+						Annotations: map[string]string{
+							workloadslicing.EnabledAnnotationKey: workloadslicing.EnabledAnnotationValue,
+						},
+					},
+				},
+			},
+		},
+		"FeatureEnabledButNotOptIn": {
+			featureEnabled: true,
+			args:           args{job: &Job{}},
+		},
+		"FeatureEnabledAndOptIn": {
+			featureEnabled: true,
+			args: args{
+				job: &Job{
+					ObjectMeta: metav1.ObjectMeta{
+						Annotations: map[string]string{
+							workloadslicing.EnabledAnnotationKey: workloadslicing.EnabledAnnotationValue,
+						},
+					},
+					Spec: batchv1.JobSpec{
+						Template: corev1.PodTemplateSpec{
+							Spec: corev1.PodSpec{
+								SchedulingGates: []corev1.PodSchedulingGate{
+									// To assert that other gates are not removed as a side effect.
+									{Name: "SomeOtherGate"},
+								},
+							},
+						},
+					},
+				},
+			},
+			want: []corev1.PodSchedulingGate{
+				{Name: "SomeOtherGate"},
+				{Name: kueue.ElasticJobSchedulingGate},
+			},
+		},
+		"FeatureEnabledAndOptIn_SpecAlreadyContainsWorkloadSliceGate": {
+			featureEnabled: true,
+			args: args{
+				job: &Job{
+					ObjectMeta: metav1.ObjectMeta{
+						Annotations: map[string]string{
+							workloadslicing.EnabledAnnotationKey: workloadslicing.EnabledAnnotationValue,
+						},
+					},
+					Spec: batchv1.JobSpec{
+						Template: corev1.PodTemplateSpec{
+							Spec: corev1.PodSpec{
+								SchedulingGates: []corev1.PodSchedulingGate{
+									{Name: kueue.ElasticJobSchedulingGate},
+								},
+							},
+						},
+					},
+				},
+			},
+			want: []corev1.PodSchedulingGate{
+				{Name: kueue.ElasticJobSchedulingGate},
+			},
+		},
+	}
+	for name, tt := range tests {
+		t.Run(name, func(t *testing.T) {
+			if err := features.SetEnable(features.ElasticJobsViaWorkloadSlices, tt.featureEnabled); err != nil {
+				t.Errorf("applyWorkloadSliceSchedulingGate() unexpcted error enabling feature: %v", err)
+			}
+			applyWorkloadSliceSchedulingGate(tt.args.job)
+			if diff := cmp.Diff(tt.args.job.Spec.Template.Spec.SchedulingGates, tt.want); diff != "" {
+				t.Errorf("applyWorkloadSliceSchedulingGate() got(-),want(+): %s", diff)
+			}
+		})
+	}
+}
diff --git a/pkg/scheduler/flavorassigner/flavorassigner_test.go b/pkg/scheduler/flavorassigner/flavorassigner_test.go
index 97520d844..511b773fd 100644
--- a/pkg/scheduler/flavorassigner/flavorassigner_test.go
+++ b/pkg/scheduler/flavorassigner/flavorassigner_test.go
@@ -77,17 +77,19 @@ func TestAssignFlavors(t *testing.T) {
 	}
 
 	cases := map[string]struct {
-		wlPods                     []kueue.PodSet
-		wlReclaimablePods          []kueue.ReclaimablePod
-		clusterQueue               kueue.ClusterQueue
-		clusterQueueUsage          resources.FlavorResourceQuantities
-		secondaryClusterQueue      *kueue.ClusterQueue
-		secondaryClusterQueueUsage resources.FlavorResourceQuantities
-		wantRepMode                FlavorAssignmentMode
-		wantAssignment             Assignment
-		disableLendingLimit        bool
-		enableFairSharing          bool
-		simulationResult           map[resources.FlavorResource]preemptioncommon.PreemptionPossibility
+		wlPods                              []kueue.PodSet
+		wlReclaimablePods                   []kueue.ReclaimablePod
+		clusterQueue                        kueue.ClusterQueue
+		clusterQueueUsage                   resources.FlavorResourceQuantities
+		secondaryClusterQueue               *kueue.ClusterQueue
+		secondaryClusterQueueUsage          resources.FlavorResourceQuantities
+		wantRepMode                         FlavorAssignmentMode
+		wantAssignment                      Assignment
+		disableLendingLimit                 bool
+		enableFairSharing                   bool
+		simulationResult                    map[resources.FlavorResource]preemptioncommon.PreemptionPossibility
+		elasticJobsViaWorkloadSlicesEnabled bool
+		preemptWorkloadSlice                *workload.Info
 	}{
 		"single flavor, fits": {
 			wlPods: []kueue.PodSet{
@@ -2163,6 +2165,115 @@ func TestAssignFlavors(t *testing.T) {
 				}},
 			},
 		},
+		"workload slice preemption fits in the original workload resource flavor": {
+			elasticJobsViaWorkloadSlicesEnabled: true,
+			wlPods: []kueue.PodSet{
+				*utiltesting.MakePodSet(kueue.DefaultPodSetName, 1).
+					Request(corev1.ResourceCPU, "3").
+					Request(corev1.ResourceMemory, "10Mi").
+					Obj(),
+			},
+			clusterQueue: utiltesting.MakeClusterQueue("test-clusterqueue").
+				ResourceGroup(
+					utiltesting.MakeFlavorQuotas("one").
+						Resource(corev1.ResourceCPU, "2").
+						Resource(corev1.ResourceMemory, "1Gi").
+						FlavorQuotas,
+					utiltesting.MakeFlavorQuotas("two").
+						Resource(corev1.ResourceCPU, "4").
+						Resource(corev1.ResourceMemory, "2Gi").
+						FlavorQuotas,
+				).
+				ClusterQueue,
+			preemptWorkloadSlice: &workload.Info{
+				TotalRequests: []workload.PodSetResources{
+					{
+						Name: "main",
+						Requests: resources.Requests{
+							corev1.ResourceCPU:    2000,
+							corev1.ResourceMemory: 10 * utiltesting.Mi,
+						},
+						Flavors: map[corev1.ResourceName]kueue.ResourceFlavorReference{
+							corev1.ResourceCPU:    "two",
+							corev1.ResourceMemory: "two",
+						},
+					},
+				},
+			},
+			wantRepMode: Fit,
+			wantAssignment: Assignment{
+				PodSets: []PodSetAssignment{{
+					Name: kueue.DefaultPodSetName,
+					Flavors: ResourceAssignment{
+						corev1.ResourceCPU:    {Name: "two", Mode: Fit, TriedFlavorIdx: -1},
+						corev1.ResourceMemory: {Name: "two", Mode: Fit, TriedFlavorIdx: -1},
+					},
+					Requests: corev1.ResourceList{
+						corev1.ResourceCPU:    resource.MustParse("3"),
+						corev1.ResourceMemory: resource.MustParse("10Mi"),
+					},
+					Count: 1,
+				}},
+				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{
+					{Flavor: "two", Resource: corev1.ResourceCPU}:    3_000,
+					{Flavor: "two", Resource: corev1.ResourceMemory}: 10 * utiltesting.Mi,
+				}},
+			},
+		},
+		"workload slice preemption does not fit in the original workload resource flavor": {
+			elasticJobsViaWorkloadSlicesEnabled: true,
+			wlPods: []kueue.PodSet{
+				*utiltesting.MakePodSet(kueue.DefaultPodSetName, 1).
+					Request(corev1.ResourceCPU, "3").
+					Request(corev1.ResourceMemory, "10Mi").
+					Obj(),
+			},
+			clusterQueue: utiltesting.MakeClusterQueue("test-clusterqueue").
+				ResourceGroup(
+					utiltesting.MakeFlavorQuotas("one").
+						Resource(corev1.ResourceCPU, "500m"). // <-- does not fit after scale-up.
+						Resource(corev1.ResourceMemory, "1Gi").
+						FlavorQuotas,
+					utiltesting.MakeFlavorQuotas("two").
+						Resource(corev1.ResourceCPU, "4").
+						Resource(corev1.ResourceMemory, "2Gi").
+						FlavorQuotas,
+				).
+				ClusterQueue,
+			preemptWorkloadSlice: &workload.Info{
+				TotalRequests: []workload.PodSetResources{
+					{
+						Name: "main",
+						Requests: resources.Requests{
+							corev1.ResourceCPU:    2000,
+							corev1.ResourceMemory: 10 * utiltesting.Mi,
+						},
+						Flavors: map[corev1.ResourceName]kueue.ResourceFlavorReference{
+							corev1.ResourceCPU:    "one",
+							corev1.ResourceMemory: "one",
+						},
+					},
+				},
+			},
+			wantRepMode: NoFit,
+			wantAssignment: Assignment{
+				PodSets: []PodSetAssignment{{
+					Name: kueue.DefaultPodSetName,
+					Requests: corev1.ResourceList{
+						corev1.ResourceCPU:    resource.MustParse("3"),
+						corev1.ResourceMemory: resource.MustParse("10Mi"),
+					},
+					Count: 1,
+					Status: &Status{
+						reasons: []string{
+							"insufficient quota for cpu in flavor one, request > maximum capacity (1 > 500m)",
+							"could not assign two flavor since the original workload is assigned: one",
+						},
+					},
+				}},
+				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{}},
+			},
+		},
 	}
 	for name, tc := range cases {
 		t.Run(name, func(t *testing.T) {
@@ -2220,7 +2331,11 @@ func TestAssignFlavors(t *testing.T) {
 				secondaryClusterQueue.AddUsage(workload.Usage{Quota: tc.secondaryClusterQueueUsage})
 			}
 
-			flvAssigner := New(wlInfo, clusterQueue, resourceFlavors, tc.enableFairSharing, &testOracle{simulationResult: tc.simulationResult})
+			if tc.elasticJobsViaWorkloadSlicesEnabled {
+				features.SetFeatureGateDuringTest(t, features.ElasticJobsViaWorkloadSlices, true)
+			}
+
+			flvAssigner := New(wlInfo, clusterQueue, resourceFlavors, tc.enableFairSharing, &testOracle{simulationResult: tc.simulationResult}, tc.preemptWorkloadSlice)
 			assignment := flvAssigner.Assign(log, nil)
 			if repMode := assignment.RepresentativeMode(); repMode != tc.wantRepMode {
 				t.Errorf("e.assignFlavors(_).RepresentativeMode()=%s, want %s", repMode, tc.wantRepMode)
@@ -2383,7 +2498,7 @@ func TestReclaimBeforePriorityPreemption(t *testing.T) {
 			testClusterQueue := snapshot.ClusterQueue("test-clusterqueue")
 			testClusterQueue.AddUsage(workload.Usage{Quota: tc.testClusterQueueUsage})
 
-			flvAssigner := New(wlInfo, testClusterQueue, resourceFlavors, false, &testOracle{tc.simulationResult})
+			flvAssigner := New(wlInfo, testClusterQueue, resourceFlavors, false, &testOracle{tc.simulationResult}, nil)
 			assignment := flvAssigner.Assign(log, nil)
 			if gotRepMode := assignment.RepresentativeMode(); gotRepMode != tc.wantMode {
 				t.Errorf("Unexpected RepresentativeMode. got %s, want %s", gotRepMode, tc.wantMode)
@@ -2513,7 +2628,7 @@ func TestDeletedFlavors(t *testing.T) {
 			cache.DeleteResourceFlavor(log, flavorMap["deleted-flavor"])
 			delete(flavorMap, "deleted-flavor")
 
-			flvAssigner := New(wlInfo, clusterQueue, flavorMap, false, &testOracle{})
+			flvAssigner := New(wlInfo, clusterQueue, flavorMap, false, &testOracle{}, nil)
 
 			assignment := flvAssigner.Assign(log, nil)
 			if repMode := assignment.RepresentativeMode(); repMode != tc.wantRepMode {
@@ -2690,7 +2805,7 @@ func TestHierarchical(t *testing.T) {
 			testClusterQueue := snapshot.ClusterQueue("test-clusterqueue")
 			testClusterQueue.AddUsage(workload.Usage{Quota: tc.testClusterQueueUsage})
 
-			flvAssigner := New(wlInfo, testClusterQueue, resourceFlavors, false, &testOracle{})
+			flvAssigner := New(wlInfo, testClusterQueue, resourceFlavors, false, &testOracle{}, nil)
 			assignment := flvAssigner.Assign(log, nil)
 			if gotRepMode := assignment.RepresentativeMode(); gotRepMode != tc.wantMode {
 				t.Errorf("Unexpected RepresentativeMode. got %s, want %s", gotRepMode, tc.wantMode)
diff --git a/pkg/scheduler/scheduler_test.go b/pkg/scheduler/scheduler_test.go
index b96da3faf..e204008ed 100644
--- a/pkg/scheduler/scheduler_test.go
+++ b/pkg/scheduler/scheduler_test.go
@@ -20,7 +20,10 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"maps"
 	"reflect"
+	goslices "slices"
+	"strconv"
 	"sync"
 	"testing"
 	"time"
@@ -61,6 +64,7 @@ import (
 	testingnode "sigs.k8s.io/kueue/pkg/util/testingjobs/node"
 	testingpod "sigs.k8s.io/kueue/pkg/util/testingjobs/pod"
 	"sigs.k8s.io/kueue/pkg/workload"
+	"sigs.k8s.io/kueue/pkg/workloadslicing"
 )
 
 const (
@@ -243,9 +247,10 @@ func TestSchedule(t *testing.T) {
 	}
 	cases := map[string]struct {
 		// Features
-		disableLendingLimit     bool
-		disablePartialAdmission bool
-		enableFairSharing       bool
+		disableLendingLimit               bool
+		disablePartialAdmission           bool
+		enableFairSharing                 bool
+		enableElasticJobsViaWorkloadSlice bool
 
 		workloads      []kueue.Workload
 		objects        []client.Object
@@ -264,7 +269,8 @@ func TestSchedule(t *testing.T) {
 		// workloadCmpOpts are the cmp options to compare workloads.
 		workloadCmpOpts cmp.Options
 		// wantWorkloads is the subset of workloads that got admitted in this cycle.
-		wantWorkloads []kueue.Workload
+		wantWorkloads             []kueue.Workload
+		wantWorkloadsWithStatuses []kueue.Workload
 		// wantLeft is the workload keys that are left in the queues after this cycle.
 		wantLeft map[kueue.ClusterQueueReference][]workload.Reference
 		// wantInadmissibleLeft is the workload keys that are left in the inadmissible state after this cycle.
@@ -277,6 +283,10 @@ func TestSchedule(t *testing.T) {
 		eventCmpOpts cmp.Options
 
 		wantSkippedPreemptions map[string]int
+
+		// trackPatchWorkloads flag to collect workload patches and use patched workloads
+		// for "wantWorkloads" assertion.
+		trackPatchedWorkloads bool
 	}{
 		"use second flavor when the first has no preemption candidates; WhenCanPreempt: Preempt": {
 			additionalClusterQueues: []kueue.ClusterQueue{
@@ -3352,6 +3362,144 @@ func TestSchedule(t *testing.T) {
 				"eng-alpha/admitted": *utiltesting.MakeAdmission("cq2").Assignment(corev1.ResourceCPU, "on-demand", "1").Obj(),
 			},
 		},
+		// Workload-slice scheduling test case.
+		"workload-slice fits in single clusterQueue": {
+			enableElasticJobsViaWorkloadSlice: true,
+			trackPatchedWorkloads:             true,
+
+			// workloads that will be returned by the fake.client.
+			workloads: []kueue.Workload{
+				*utiltesting.MakeWorkload("foo-1", "sales").
+					ResourceVersion("1").
+					Queue("main").
+					PodSets(*utiltesting.MakePodSet("one", 10).
+						Request(corev1.ResourceCPU, "1").
+						Obj()).
+					Generation(1).
+					ReserveQuota(utiltesting.MakeAdmission("sales", "one").Assignment(corev1.ResourceCPU, "default", "10000m").AssignmentPodCount(10).Obj()).
+					Condition(metav1.Condition{
+						Type:               kueue.WorkloadQuotaReserved,
+						Status:             metav1.ConditionTrue,
+						Reason:             kueue.WorkloadQuotaReserved,
+						Message:            "Quota reserved in ClusterQueue sales",
+						ObservedGeneration: 1,
+					}).
+					Condition(metav1.Condition{
+						Type:               kueue.WorkloadAdmitted,
+						Status:             metav1.ConditionTrue,
+						Reason:             kueue.WorkloadAdmitted,
+						Message:            "The workload is admitted",
+						ObservedGeneration: 1,
+					}).
+					Obj(),
+				*utiltesting.MakeWorkload("foo-2", "sales").
+					ResourceVersion("1").
+					Annotation(workloadslicing.WorkloadSliceReplacementFor, "sales/foo-1").
+					Queue("main").
+					PodSets(*utiltesting.MakePodSet("one", 15).
+						Request(corev1.ResourceCPU, "1").
+						Obj()).
+					Generation(1).
+					Obj(),
+			},
+			// wantAssignments is a map of workload name to the status assignments expected to be in the cache after the scheduling cycle.
+			wantAssignments: map[workload.Reference]kueue.Admission{
+				"sales/foo-1": *utiltesting.MakeAdmission("sales", "one").Assignment(corev1.ResourceCPU, "default", "10").AssignmentPodCount(10).Obj(),
+				"sales/foo-2": *utiltesting.MakeAdmission("sales", "one").Assignment(corev1.ResourceCPU, "default", "15").AssignmentPodCount(15).Obj(),
+			},
+			// wantScheduled is a list of workloads admission status expected to be added to the cache after the scheduling cycle.
+			wantScheduled: []workload.Reference{"sales/foo-2"},
+			// wantWorkloads an authoritative list of workloads expected in K8s API after the scheduling cycle.
+			// This may not be the same as previous "want*" values due to the stabbed apply status invocations in the test.
+			wantWorkloads: []kueue.Workload{
+				*utiltesting.MakeWorkload("foo-1", "sales").
+					ResourceVersion("1").
+					Queue("main").
+					PodSets(*utiltesting.MakePodSet("one", 10).
+						Request(corev1.ResourceCPU, "1").
+						Obj()).
+					Admission(
+						utiltesting.MakeAdmission("sales", "one").
+							Assignment(corev1.ResourceCPU, "default", "10000m").
+							AssignmentPodCount(10).
+							Obj(),
+					).
+					Generation(1).
+					Condition(metav1.Condition{
+						Type:               kueue.WorkloadQuotaReserved,
+						Status:             metav1.ConditionTrue,
+						Reason:             kueue.WorkloadQuotaReserved,
+						Message:            "Quota reserved in ClusterQueue sales",
+						ObservedGeneration: 1,
+					}).
+					Condition(metav1.Condition{
+						Type:               kueue.WorkloadAdmitted,
+						Status:             metav1.ConditionTrue,
+						Reason:             kueue.WorkloadAdmitted,
+						Message:            "The workload is admitted",
+						ObservedGeneration: 1,
+					}).
+					Condition(metav1.Condition{
+						Type:    kueue.WorkloadFinished,
+						Status:  metav1.ConditionTrue,
+						Reason:  kueue.WorkloadSliceReplaced,
+						Message: "Replaced to accommodate a workload (UID: , JobUID: ) due to workload slice aggregation",
+					}).
+					Obj(),
+				*utiltesting.MakeWorkload("foo-2", "sales").
+					Annotation(workloadslicing.WorkloadSliceReplacementFor, "sales/foo-1").
+					ResourceVersion("1").
+					Queue("main").
+					PodSets(*utiltesting.MakePodSet("one", 15).
+						Request(corev1.ResourceCPU, "1").
+						Obj()).
+					Admission(
+						utiltesting.MakeAdmission("sales", "one").
+							Assignment(corev1.ResourceCPU, "default", "15000m").
+							AssignmentPodCount(15).
+							Obj(),
+					).
+					Generation(1).
+					Condition(metav1.Condition{
+						Type:               kueue.WorkloadQuotaReserved,
+						Status:             metav1.ConditionTrue,
+						Reason:             kueue.WorkloadQuotaReserved,
+						Message:            "Quota reserved in ClusterQueue sales",
+						ObservedGeneration: 1,
+					}).
+					Condition(metav1.Condition{
+						Type:               kueue.WorkloadAdmitted,
+						Status:             metav1.ConditionTrue,
+						Reason:             kueue.WorkloadAdmitted,
+						Message:            "The workload is admitted",
+						ObservedGeneration: 1,
+					}).
+					Obj(),
+			},
+			workloadCmpOpts: cmp.Options{
+				cmpopts.EquateEmpty(),
+				cmpopts.IgnoreFields(metav1.Condition{}, "LastTransitionTime"),
+				cmpopts.SortSlices(func(a, b kueue.Workload) bool { return workload.Key(&a) < workload.Key(&b) }),
+			},
+			eventCmpOpts: ignoreEventMessageCmpOpts,
+			wantEvents: []utiltesting.EventRecord{
+				{
+					Key:       types.NamespacedName{Namespace: "sales", Name: "foo-1"},
+					Reason:    kueue.WorkloadSliceReplaced,
+					EventType: corev1.EventTypeNormal,
+				},
+				{
+					Key:       types.NamespacedName{Namespace: "sales", Name: "foo-2"},
+					Reason:    "QuotaReserved",
+					EventType: corev1.EventTypeNormal,
+				},
+				{
+					Key:       types.NamespacedName{Namespace: "sales", Name: "foo-2"},
+					Reason:    "Admitted",
+					EventType: corev1.EventTypeNormal,
+				},
+			},
+		},
 	}
 	for name, tc := range cases {
 		t.Run(name, func(t *testing.T) {
@@ -3362,6 +3510,8 @@ func TestSchedule(t *testing.T) {
 			if tc.disablePartialAdmission {
 				features.SetFeatureGateDuringTest(t, features.PartialAdmission, false)
 			}
+			features.SetFeatureGateDuringTest(t, features.ElasticJobsViaWorkloadSlices, tc.enableElasticJobsViaWorkloadSlice)
+
 			ctx, log := utiltesting.ContextWithLog(t)
 
 			allQueues := append(queues, tc.additionalLocalQueues...)
@@ -3378,6 +3528,32 @@ func TestSchedule(t *testing.T) {
 						utiltesting.MakeNamespaceWrapper("lend").Label("dep", "lend").Obj(),
 					}, tc.objects...,
 				)...)
+			for i := range tc.workloads {
+				clientBuilder.WithStatusSubresource(&tc.workloads[i])
+			}
+
+			// Patch workloads tracking by leveraging fake.Client sub-resource interceptor.
+			patchedWorkloads := sets.New[*kueue.Workload]()
+			if tc.trackPatchedWorkloads {
+				clientBuilder.WithInterceptorFuncs(interceptor.Funcs{
+					SubResourcePatch: func(ctx context.Context, clnt client.Client, subResourceName string, obj client.Object, patch client.Patch, opts ...client.SubResourcePatchOption) error {
+						if subResourceName != "status" {
+							// There should be only "status" subresource patches in the workload scheduler context.
+							return fmt.Errorf("unexpected subresource in workload status patch: %s", subResourceName)
+						}
+						// There should be only *kueue.Workloads status patches.
+						wl, ok := obj.(*kueue.Workload)
+						if !ok {
+							return fmt.Errorf("unexpected object in workload status patch: %T", obj)
+						}
+						// Bump up resource version.
+						rv, _ := strconv.Atoi(wl.GetResourceVersion())
+						wl.SetResourceVersion(strconv.Itoa(rv))
+						patchedWorkloads.Insert(wl)
+						return nil
+					},
+				})
+			}
 			cl := clientBuilder.Build()
 			recorder := &utiltesting.EventRecorder{}
 			cqCache := cache.New(cl)
@@ -3410,6 +3586,12 @@ func TestSchedule(t *testing.T) {
 			}
 
 			scheduler := New(qManager, cqCache, cl, recorder, WithFairSharing(&config.FairSharing{Enable: tc.enableFairSharing}), WithClock(t, fakeClock))
+			wg := sync.WaitGroup{}
+			scheduler.setAdmissionRoutineWrapper(routine.NewWrapper(
+				func() { wg.Add(1) },
+				func() { wg.Done() },
+			))
+
 			gotScheduled := make(map[workload.Reference]kueue.Admission)
 			var mu sync.Mutex
 			scheduler.applyAdmission = func(ctx context.Context, w *kueue.Workload) error {
@@ -3421,11 +3603,6 @@ func TestSchedule(t *testing.T) {
 				mu.Unlock()
 				return nil
 			}
-			wg := sync.WaitGroup{}
-			scheduler.setAdmissionRoutineWrapper(routine.NewWrapper(
-				func() { wg.Add(1) },
-				func() { wg.Done() },
-			))
 			gotPreempted := sets.New[workload.Reference]()
 			scheduler.preemptor.OverrideApply(func(_ context.Context, w *kueue.Workload, _, _ string) error {
 				mu.Lock()
@@ -3455,14 +3632,14 @@ func TestSchedule(t *testing.T) {
 
 			// Verify assignments in cache.
 			gotAssignments := make(map[workload.Reference]kueue.Admission)
-			var gotWorkloads []kueue.Workload
+			gotWorkloads := make(map[workload.Reference]kueue.Workload)
 			snapshot, err := cqCache.Snapshot(ctx)
 			if err != nil {
 				t.Fatalf("unexpected error while building snapshot: %v", err)
 			}
 			for cqName, c := range snapshot.ClusterQueues() {
 				for name, w := range c.Workloads {
-					gotWorkloads = append(gotWorkloads, *w.Obj)
+					gotWorkloads[name] = *w.Obj
 					switch {
 					case !workload.HasQuotaReservation(w.Obj):
 						t.Errorf("Workload %s is not admitted by a clusterQueue, but it is found as member of clusterQueue %s in the cache", name, cqName)
@@ -3473,9 +3650,15 @@ func TestSchedule(t *testing.T) {
 					}
 				}
 			}
+			// If we track patched workloads, replace workload defined in queue with the tracked patch instance,
+			// since workloads in the queue are "stale" in respect of status updates.
+			// Note: if we don't track workload status patching - patchWorkloads will be empty, e.g., "noop".
+			for wl := range patchedWorkloads {
+				gotWorkloads[workload.Key(wl)] = *wl
+			}
 
 			if tc.wantWorkloads != nil {
-				if diff := cmp.Diff(tc.wantWorkloads, gotWorkloads, tc.workloadCmpOpts...); diff != "" {
+				if diff := cmp.Diff(tc.wantWorkloads, goslices.Collect(maps.Values(gotWorkloads)), tc.workloadCmpOpts...); diff != "" {
 					t.Errorf("Unexpected workloads in cache (-want,+got):\n%s", diff)
 				}
 			}
diff --git a/pkg/webhooks/workload_webhook_test.go b/pkg/webhooks/workload_webhook_test.go
index 310335153..4c7c46aa1 100644
--- a/pkg/webhooks/workload_webhook_test.go
+++ b/pkg/webhooks/workload_webhook_test.go
@@ -28,6 +28,7 @@ import (
 	"k8s.io/utils/ptr"
 
 	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta1"
+	"sigs.k8s.io/kueue/pkg/features"
 	testingutil "sigs.k8s.io/kueue/pkg/util/testing"
 )
 
@@ -230,6 +231,7 @@ func TestValidateWorkload(t *testing.T) {
 func TestValidateWorkloadUpdate(t *testing.T) {
 	testCases := map[string]struct {
 		enableTopologyAwareScheduling bool
+		enableElasticJobsFeature      bool
 
 		before, after *kueue.Workload
 		wantErr       field.ErrorList
@@ -575,9 +577,47 @@ func TestValidateWorkloadUpdate(t *testing.T) {
 				field.Invalid(field.NewPath("status", "admission"), nil, ""),
 			},
 		},
+
+		"workload.podSets[].count is immutable": {
+			before: testingutil.MakeWorkload(testWorkloadName, testWorkloadNamespace).
+				PodSets(
+					*testingutil.MakePodSet("ps1", 3).Obj(),
+					*testingutil.MakePodSet("ps2", 4).Obj()).
+				ReserveQuota(testingutil.MakeAdmission("cluster-queue").PodSets(
+					kueue.PodSetAssignment{Name: "ps1"},
+					kueue.PodSetAssignment{Name: "ps2"}).Obj()).Obj(),
+			after: testingutil.MakeWorkload(testWorkloadName, testWorkloadNamespace).
+				PodSets(
+					*testingutil.MakePodSet("ps1", 3).Obj(),
+					*testingutil.MakePodSet("ps2", 3).Obj()).
+				ReserveQuota(testingutil.MakeAdmission("cluster-queue").PodSets(
+					kueue.PodSetAssignment{Name: "ps1"},
+					kueue.PodSetAssignment{Name: "ps2"}).Obj()).Obj(),
+			wantErr: field.ErrorList{
+				field.Invalid(field.NewPath("spec", "podSets", "1"), nil, ""),
+			},
+		},
+		"workload.podSets[].count is mutable with ElasticJobs feature gate": {
+			enableElasticJobsFeature: true,
+			before: testingutil.MakeWorkload(testWorkloadName, testWorkloadNamespace).
+				PodSets(
+					*testingutil.MakePodSet("ps1", 3).Obj(),
+					*testingutil.MakePodSet("ps2", 4).Obj()).
+				ReserveQuota(testingutil.MakeAdmission("cluster-queue").PodSets(
+					kueue.PodSetAssignment{Name: "ps1"},
+					kueue.PodSetAssignment{Name: "ps2"}).Obj()).Obj(),
+			after: testingutil.MakeWorkload(testWorkloadName, testWorkloadNamespace).
+				PodSets(
+					*testingutil.MakePodSet("ps1", 3).Obj(),
+					*testingutil.MakePodSet("ps2", 3).Obj()).
+				ReserveQuota(testingutil.MakeAdmission("cluster-queue").PodSets(
+					kueue.PodSetAssignment{Name: "ps1"},
+					kueue.PodSetAssignment{Name: "ps2"}).Obj()).Obj(),
+		},
 	}
 	for name, tc := range testCases {
 		t.Run(name, func(t *testing.T) {
+			features.SetFeatureGateDuringTest(t, features.ElasticJobsViaWorkloadSlices, tc.enableElasticJobsFeature)
 			errList := ValidateWorkloadUpdate(tc.after, tc.before)
 			if diff := cmp.Diff(tc.wantErr, errList, cmpopts.IgnoreFields(field.Error{}, "Detail", "BadValue")); diff != "" {
 				t.Errorf("ValidateWorkloadUpdate() mismatch (-want +got):\n%s", diff)
diff --git a/pkg/workload/podsetscounts_test.go b/pkg/workload/podsetscounts_test.go
new file mode 100644
index 000000000..ec7dcb6ee
--- /dev/null
+++ b/pkg/workload/podsetscounts_test.go
@@ -0,0 +1,237 @@
+/*
+Copyright The Kubernetes Authors.
+
+Licensed 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 workload
+
+import (
+	"testing"
+
+	"github.com/google/go-cmp/cmp"
+
+	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta1"
+)
+
+func testPodSet(name string, count int32) kueue.PodSet {
+	return kueue.PodSet{
+		Name:  kueue.PodSetReference(name),
+		Count: count,
+	}
+}
+
+func TestExtractPodSetCounts(t *testing.T) {
+	type args struct {
+		podSets []kueue.PodSet
+	}
+
+	tests := map[string]struct {
+		args args
+		want PodSetsCounts
+	}{
+		"EmptyPodSets": {},
+		"SinglePodSet": {
+			args: args{
+				podSets: []kueue.PodSet{testPodSet("test", 10)},
+			},
+			want: PodSetsCounts{
+				"test": 10,
+			},
+		},
+		"MultiplePodSets": {
+			args: args{
+				podSets: []kueue.PodSet{testPodSet("test-a", 10), testPodSet("test-b", 11)},
+			},
+			want: PodSetsCounts{
+				"test-a": 10,
+				"test-b": 11,
+			},
+		},
+	}
+	for name, tt := range tests {
+		t.Run(name, func(t *testing.T) {
+			if diff := cmp.Diff(ExtractPodSetCounts(tt.args.podSets), tt.want); diff != "" {
+				t.Errorf("ExtractPodSetCounts() got(-),want(+): %s", diff)
+			}
+		})
+	}
+}
+
+func TestExtractPodSetCountsFromWorkload(t *testing.T) {
+	type args struct {
+		workload *kueue.Workload
+	}
+	tests := map[string]struct {
+		args args
+		want PodSetsCounts
+	}{
+		"EmptyPodSets": {
+			args: args{
+				workload: &kueue.Workload{},
+			},
+		},
+		"SinglePodSet": {
+			args: args{
+				workload: &kueue.Workload{
+					Spec: kueue.WorkloadSpec{PodSets: []kueue.PodSet{testPodSet("test", 10)}},
+				},
+			},
+			want: PodSetsCounts{
+				"test": 10,
+			},
+		},
+		"MultiplePodSets": {
+			args: args{
+				workload: &kueue.Workload{
+					Spec: kueue.WorkloadSpec{PodSets: []kueue.PodSet{
+						testPodSet("test-a", 10),
+						testPodSet("test-b", 11),
+					}},
+				},
+			},
+			want: PodSetsCounts{
+				"test-a": 10,
+				"test-b": 11,
+			},
+		},
+	}
+	for name, tt := range tests {
+		t.Run(name, func(t *testing.T) {
+			if diff := cmp.Diff(ExtractPodSetCountsFromWorkload(tt.args.workload), tt.want); diff != "" {
+				t.Errorf("ExtractPodSetCountsFromWorkload() got(-),want(+): %s", diff)
+			}
+		})
+	}
+}
+
+func TestApplyPodSetCounts(t *testing.T) {
+	type args struct {
+		wl     *kueue.Workload
+		counts PodSetsCounts
+	}
+	tests := map[string]struct {
+		args args
+		want *kueue.Workload
+	}{
+		"EdgeCase_EmptyWorkloadPodsSets": {
+			args: args{
+				wl: &kueue.Workload{},
+				counts: PodSetsCounts{
+					"test": 10,
+				},
+			},
+			want: &kueue.Workload{},
+		},
+		"EdgeCase_EmptyPodsSetsCounts": {
+			args: args{
+				wl: &kueue.Workload{
+					Spec: kueue.WorkloadSpec{
+						PodSets: []kueue.PodSet{
+							{
+								Name:  "test-a",
+								Count: 10,
+							},
+						},
+					},
+				},
+			},
+			want: &kueue.Workload{
+				Spec: kueue.WorkloadSpec{
+					PodSets: []kueue.PodSet{
+						{
+							Name:  "test-a",
+							Count: 10,
+						},
+					},
+				},
+			},
+		},
+		"ApplyCounts": {
+			args: args{
+				wl: &kueue.Workload{
+					Spec: kueue.WorkloadSpec{
+						PodSets: []kueue.PodSet{
+							{Name: "test-a", Count: 10},
+							{Name: "test-b", Count: 11},
+						},
+					},
+				},
+				counts: PodSetsCounts{
+					"test-a": 20,
+					"test-b": 2,
+				},
+			},
+			want: &kueue.Workload{
+				Spec: kueue.WorkloadSpec{
+					PodSets: []kueue.PodSet{
+						{Name: "test-a", Count: 20},
+						{Name: "test-b", Count: 2},
+					},
+				},
+			},
+		},
+	}
+	for name, tt := range tests {
+		t.Run(name, func(t *testing.T) {
+			ApplyPodSetCounts(tt.args.wl, tt.args.counts)
+			if diff := cmp.Diff(tt.args.wl, tt.want); diff != "" {
+				t.Errorf("ApplyPodSetCounts() got(-),want(+): %s", diff)
+			}
+		})
+	}
+}
+
+func TestPodSetsCounts_HasFewerReplicasThan(t *testing.T) {
+	type args struct {
+		in PodSetsCounts
+	}
+	tests := map[string]struct {
+		c    PodSetsCounts
+		args args
+		want bool
+	}{
+		"HasFewer": {
+			c: PodSetsCounts{
+				"test-a": 11,
+				"test-b": 2,
+			},
+			args: args{
+				in: PodSetsCounts{
+					"test-a": 10,
+					"test-b": 20,
+				},
+			},
+			want: true,
+		},
+		"DoesNotHaveFewer": {
+			c: PodSetsCounts{
+				"test-a": 11,
+				"test-b": 2,
+			},
+			args: args{
+				in: PodSetsCounts{
+					"test-a": 10,
+					"test-b": 2,
+				},
+			},
+		},
+	}
+	for name, tt := range tests {
+		t.Run(name, func(t *testing.T) {
+			if got := tt.c.HasFewerReplicasThan(tt.args.in); got != tt.want {
+				t.Errorf("HasFewerReplicasThan() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
diff --git a/pkg/workloadslicing/workloadslicing_test.go b/pkg/workloadslicing/workloadslicing_test.go
new file mode 100644
index 000000000..45f276053
--- /dev/null
+++ b/pkg/workloadslicing/workloadslicing_test.go
@@ -0,0 +1,1108 @@
+/*
+Copyright The Kubernetes Authors.
+
+Licensed 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 workloadslicing
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"strconv"
+	"testing"
+	"time"
+
+	"github.com/google/go-cmp/cmp"
+	"github.com/google/go-cmp/cmp/cmpopts"
+	batchv1 "k8s.io/api/batch/v1"
+	corev1 "k8s.io/api/core/v1"
+	apimeta "k8s.io/apimachinery/pkg/api/meta"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/uuid"
+	"k8s.io/client-go/kubernetes/scheme"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/client/fake"
+	"sigs.k8s.io/controller-runtime/pkg/client/interceptor"
+
+	kueuealpha "sigs.k8s.io/kueue/apis/kueue/v1alpha1"
+	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta1"
+	"sigs.k8s.io/kueue/pkg/cache"
+	"sigs.k8s.io/kueue/pkg/controller/core/indexer"
+	"sigs.k8s.io/kueue/pkg/features"
+	"sigs.k8s.io/kueue/pkg/hierarchy"
+	"sigs.k8s.io/kueue/pkg/scheduler/preemption"
+	utiltesting "sigs.k8s.io/kueue/pkg/util/testing"
+	"sigs.k8s.io/kueue/pkg/workload"
+)
+
+func TestEnabled(t *testing.T) {
+	type args struct {
+		object metav1.Object
+	}
+	tests := map[string]struct {
+		args args
+		want bool
+	}{
+		"NilObject": {},
+		"NilAnnotation": {
+			args: args{
+				object: &batchv1.Job{},
+			},
+		},
+		"EmptyAnnotation": {
+			args: args{
+				object: &batchv1.Job{
+					ObjectMeta: metav1.ObjectMeta{Annotations: map[string]string{}},
+				},
+			},
+		},
+		"Enabled": {
+			args: args{
+				object: &batchv1.Job{
+					ObjectMeta: metav1.ObjectMeta{Annotations: map[string]string{
+						EnabledAnnotationKey: EnabledAnnotationValue,
+					}},
+				},
+			},
+			want: true,
+		},
+		"NotEnabled": {
+			args: args{
+				object: &batchv1.Job{
+					ObjectMeta: metav1.ObjectMeta{Annotations: map[string]string{
+						EnabledAnnotationKey: "True", // <-- value is case sensitive.
+					}},
+				},
+			},
+		},
+	}
+	for name, tt := range tests {
+		t.Run(name, func(t *testing.T) {
+			if got := Enabled(tt.args.object); got != tt.want {
+				t.Errorf("Enabled() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
+func TestPreemptibleSliceKey(t *testing.T) {
+	type args struct {
+		wl *kueue.Workload
+	}
+	testReference := workload.NewReference("test", "test")
+	tests := map[string]struct {
+		args args
+		want *workload.Reference
+	}{
+		"NilAnnotations": {
+			args: args{
+				wl: &kueue.Workload{},
+			},
+		},
+		"EmptyAnnotations": {
+			args: args{
+				wl: &kueue.Workload{
+					ObjectMeta: metav1.ObjectMeta{Annotations: make(map[string]string)},
+				},
+			},
+		},
+		"Found": {
+			args: args{
+				wl: &kueue.Workload{
+					ObjectMeta: metav1.ObjectMeta{Annotations: map[string]string{WorkloadSliceReplacementFor: string(testReference)}},
+				},
+			},
+			want: &testReference,
+		},
+	}
+	for name, tt := range tests {
+		t.Run(name, func(t *testing.T) {
+			if diff := cmp.Diff(ReplacementForKey(tt.args.wl), tt.want); diff != "" {
+				t.Errorf("ReplacementForKey() (-want,+got)\n:%s", diff)
+			}
+		})
+	}
+}
+
+var (
+	testJobGVK = batchv1.SchemeGroupVersion.WithKind("Job")
+
+	testJobObject = &batchv1.Job{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "test",
+			UID:  uuid.NewUUID(),
+		},
+	}
+)
+
+func testWorkloadClientBuilder() *fake.ClientBuilder {
+	testSchema := runtime.NewScheme()
+	_ = kueue.AddToScheme(testSchema)
+	return fake.NewClientBuilder().
+		WithScheme(testSchema).
+		WithIndex(&kueue.Workload{}, fmt.Sprintf(".metadata.ownerReferences[%s.%s]", testJobGVK.Group, testJobGVK.Kind), func(object client.Object) []string {
+			wl, ok := object.(*kueue.Workload)
+			if !ok || len(wl.OwnerReferences) == 0 {
+				return nil
+			}
+			owners := make([]string, 0, len(wl.OwnerReferences))
+			for i := range wl.OwnerReferences {
+				owner := &wl.OwnerReferences[i]
+				if owner.Kind == testJobGVK.Kind && owner.APIVersion == testJobGVK.GroupVersion().String() {
+					owners = append(owners, owner.Name)
+				}
+			}
+			return owners
+		})
+}
+
+func testWorkload(name, jobName string, jobUID types.UID, created time.Time) *utiltesting.WorkloadWrapper {
+	return utiltesting.MakeWorkload(name, "default").
+		OwnerReference(testJobGVK, jobName, string(jobUID)).
+		Creation(created).
+		ResourceVersion("1").
+		Request(corev1.ResourceCPU, "100m")
+}
+
+func TestFindNotFinishedWorkloads(t *testing.T) {
+	type args struct {
+		ctx          context.Context
+		clnt         client.Client
+		jobObject    client.Object
+		jobObjectGVK schema.GroupVersionKind
+	}
+
+	// test "constants".
+	now := time.Now()
+
+	// test cases.
+	tests := map[string]struct {
+		args    args
+		want    []kueue.Workload
+		wantErr bool
+	}{
+		"ListFailure": {
+			args: args{
+				ctx:          t.Context(),
+				clnt:         fake.NewFakeClient(),
+				jobObject:    testJobObject,
+				jobObjectGVK: testJobGVK,
+			},
+			wantErr: true,
+		},
+		"EmptyList": {
+			args: args{
+				ctx:          t.Context(),
+				clnt:         testWorkloadClientBuilder().Build(),
+				jobObject:    testJobObject,
+				jobObjectGVK: testJobGVK,
+			},
+			want: []kueue.Workload{},
+		},
+		"SortedAndFiltered": {
+			args: args{
+				ctx: t.Context(),
+				clnt: testWorkloadClientBuilder().WithLists(&kueue.WorkloadList{
+					Items: []kueue.Workload{
+						*testWorkload("test-2", testJobObject.Name, testJobObject.UID, now).Obj(),
+						*testWorkload("test-1", testJobObject.Name, testJobObject.UID, now.Add(-time.Minute)).Obj(),
+						*testWorkload("test-0", testJobObject.Name, testJobObject.UID, now.Add(-time.Hour)).Finished().Obj(),
+						*testWorkload("test-4", "some-other-job", uuid.NewUUID(), now).Obj(),
+					},
+				}).Build(),
+				jobObject:    testJobObject,
+				jobObjectGVK: testJobGVK,
+			},
+			want: []kueue.Workload{
+				*testWorkload("test-1", testJobObject.Name, testJobObject.UID, now.Add(-time.Minute)).Obj(),
+				*testWorkload("test-2", testJobObject.Name, testJobObject.UID, now).Obj(),
+			},
+		},
+	}
+	for name, tt := range tests {
+		t.Run(name, func(t *testing.T) {
+			got, err := FindNotFinishedWorkloads(tt.args.ctx, tt.args.clnt, tt.args.jobObject, tt.args.jobObjectGVK)
+			if (err != nil) != tt.wantErr {
+				t.Errorf("FindActiveSlices() error = %v, wantErr %v", err, tt.wantErr)
+				return
+			}
+			if diff := cmp.Diff(got, tt.want, cmpopts.EquateApproxTime(time.Second)); diff != "" {
+				t.Errorf("FindActiveSlices() got(-),want(+): %s", diff)
+			}
+		})
+	}
+}
+
+func TestFinish(t *testing.T) {
+	type args struct {
+		ctx           context.Context
+		clnt          client.Client
+		workloadSlice *kueue.Workload
+		reason        string
+		message       string
+	}
+	type want struct {
+		err bool
+		// side effect.
+		workload *kueue.Workload
+	}
+	now := time.Now()
+	tests := map[string]struct {
+		args args
+		want want
+	}{
+		"FailureToApplyConditions": {
+			args: args{
+				ctx: t.Context(),
+				clnt: testWorkloadClientBuilder().
+					WithInterceptorFuncs(interceptor.Funcs{
+						SubResourcePatch: func(ctx context.Context, client client.Client, subResourceName string, obj client.Object, patch client.Patch, opts ...client.SubResourcePatchOption) error {
+							return errors.New("test-patch-status-error")
+						},
+					}).
+					Build(),
+				workloadSlice: testWorkload("test", "test-job", "job-uid", now).Obj(),
+				reason:        "TestReason",
+				message:       "Test Message.",
+			},
+			want: want{err: true},
+		},
+		"AlreadyFinishedWorkload": {
+			args: args{
+				ctx: t.Context(),
+				clnt: testWorkloadClientBuilder().
+					WithObjects(testWorkload("test", "test-job", "job-uid", now).Finished().Obj()).Build(),
+				workloadSlice: testWorkload("test", "test-job", "job-uid", now).Finished().Obj(),
+				reason:        "TestReason",
+				message:       "Test Message.",
+			},
+			want: want{
+				workload: testWorkload("test", "test-job", "job-uid", now).Finished().Obj(),
+			},
+		},
+		"NotFinished": {
+			args: args{
+				ctx: t.Context(),
+				clnt: testWorkloadClientBuilder().
+					WithObjects(testWorkload("test", "test-job", "job-uid", now).Obj()).
+					WithStatusSubresource(testWorkload("test", "test-job", "job-uid", now).Obj()).Build(),
+				workloadSlice: testWorkload("test", "test-job", "job-uid", now).Obj(),
+				reason:        "TestReason",
+				message:       "Test Message.",
+			},
+			want: want{
+				workload: testWorkload("test", "test-job", "job-uid", now).
+					ResourceVersion("2").
+					Condition(metav1.Condition{
+						Type:    kueue.WorkloadFinished,
+						Status:  metav1.ConditionTrue,
+						Reason:  "TestReason",
+						Message: "Test Message.",
+					}).Obj(),
+			},
+		},
+	}
+	for name, tt := range tests {
+		t.Run(name, func(t *testing.T) {
+			if err := Finish(tt.args.ctx, tt.args.clnt, tt.args.workloadSlice, tt.args.reason, tt.args.message); (err != nil) != tt.want.err {
+				t.Errorf("Finish() error = %v, wantErr %v", err, tt.want.err)
+			}
+			if tt.want.workload != nil {
+				if err := tt.args.clnt.Get(tt.args.ctx, client.ObjectKeyFromObject(tt.args.workloadSlice), tt.args.workloadSlice); err != nil {
+					t.Errorf("unexpected error retrieving workload: %v", err)
+				}
+				if diff := cmp.Diff(tt.want.workload, tt.args.workloadSlice, cmpopts.SortSlices(func(a, b metav1.Condition) bool {
+					return a.Type < b.Type
+				}), cmpopts.EquateApproxTime(time.Second)); diff != "" {
+					t.Errorf("Deactivated() (-want,+got):\n%s", diff)
+				}
+			}
+		})
+	}
+}
+
+func TestEnsureWorkloadSlices(t *testing.T) {
+	type args struct {
+		ctx          context.Context
+		clnt         client.Client
+		jobPodSets   []kueue.PodSet
+		jobObject    client.Object
+		jobObjectGVK schema.GroupVersionKind
+	}
+	type want struct {
+		workload   *kueue.Workload
+		compatible bool
+		error      bool
+	}
+	now := time.Now()
+	fiveMinutesAgo := now.Add(-5 * time.Minute)
+	testPodSets := func(count int32) []kueue.PodSet {
+		return []kueue.PodSet{
+			{
+				Name:  kueue.DefaultPodSetName,
+				Count: count,
+			},
+		}
+	}
+	testResourceVersion := int64(100)
+	// testWorkload helper constructs a workload object with the provided name, resourceVersion/generation and podSets.
+	// The workload's creation time is offset by 5 minutes in the past + N milliseconds, where N is "derived"
+	// from the resource version, to adjust for later creation time.
+	// For example: resource version 100 - will result in creation timestamp 5 min ago + 100 millis.
+	testWorkload := func(name string, resourceVersion int64, podSets []kueue.PodSet) *kueue.Workload {
+		return &kueue.Workload{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:              name,
+				Namespace:         testJobObject.Namespace,
+				ResourceVersion:   strconv.FormatInt(resourceVersion, 10),
+				CreationTimestamp: metav1.NewTime(fiveMinutesAgo.Add(time.Duration(resourceVersion) * time.Millisecond)),
+				OwnerReferences: []metav1.OwnerReference{
+					{
+						APIVersion: testJobGVK.GroupVersion().String(),
+						Kind:       testJobGVK.Kind,
+						Name:       testJobObject.Name,
+					},
+				},
+			},
+			Spec: kueue.WorkloadSpec{
+				PodSets: podSets,
+			},
+		}
+	}
+
+	assertStatusConditionPatch := func(t *testing.T, subResourceName string, obj client.Object, wantWorkloadName string, activeConditionType, activeConditionReason string) error {
+		// Assert side effect: old slice is aggregated and marked as "finished".
+		if subResourceName != "status" {
+			t.Errorf("unexpected workload patch subresource: %s", subResourceName)
+		}
+		wl, ok := obj.(*kueue.Workload)
+		if !ok {
+			t.Errorf("unexpected workload patch object type: %T", obj)
+		}
+		if wl.Name != wantWorkloadName {
+			t.Errorf("unexpected workload name: %s", wl.Name)
+		}
+		condition := apimeta.FindStatusCondition(wl.Status.Conditions, activeConditionType)
+		if condition == nil {
+			t.Fatalf("patched condition: %s is not found", activeConditionType)
+		}
+		if condition.Status != metav1.ConditionTrue {
+			t.Errorf("patched condition: %s is not active", activeConditionType)
+		}
+		if condition.Reason != activeConditionReason {
+			t.Errorf("patched condition: %s reseason - want: %s, got: %s", activeConditionType, activeConditionReason, condition.Reason)
+		}
+		return nil
+	}
+
+	tests := map[string]struct {
+		args args
+		want want
+	}{
+		"FailedListWorkloads": {
+			args: args{
+				ctx: t.Context(),
+				clnt: testWorkloadClientBuilder().
+					WithInterceptorFuncs(interceptor.Funcs{
+						List: func(_ context.Context, _ client.WithWatch, _ client.ObjectList, _ ...client.ListOption) error {
+							return errors.New("test-list-error")
+						},
+					}).
+					Build(),
+				jobObject:    testJobObject,
+				jobObjectGVK: testJobGVK,
+			},
+			want: want{
+				error:      true,
+				compatible: true,
+			},
+		},
+		// No workloads.
+		"NoWorkloadSlices": {
+			args: args{
+				ctx:          t.Context(),
+				clnt:         testWorkloadClientBuilder().Build(),
+				jobObject:    testJobObject,
+				jobObjectGVK: testJobGVK,
+			},
+			want: want{
+				compatible: true,
+			},
+		},
+		// One workload.
+		"OneWorkloadSlice_IncompatibleWithJob": {
+			args: args{
+				ctx: t.Context(),
+				clnt: testWorkloadClientBuilder().WithObjects(
+					utiltesting.MakeWorkload(testJobObject.Name+"-1", testJobObject.Namespace).
+						OwnerReference(testJobGVK, testJobObject.Name, string(testJobObject.UID)).
+						ResourceVersion("1").
+						PodSets(*utiltesting.MakePodSet(kueue.DefaultPodSetName, 1).Request(corev1.ResourceCPU, "1").Obj()).
+						Obj()).Build(),
+				jobPodSets:   []kueue.PodSet{*utiltesting.MakePodSet("different-name", 1).Request(corev1.ResourceCPU, "1").Obj()},
+				jobObject:    testJobObject,
+				jobObjectGVK: testJobGVK,
+			},
+		},
+		"OneWorkloadSlice_CurrentWorkload": {
+			args: args{
+				ctx: t.Context(),
+				clnt: testWorkloadClientBuilder().WithObjects(
+					utiltesting.MakeWorkload(testJobObject.Name+"-1", testJobObject.Namespace).
+						OwnerReference(testJobGVK, testJobObject.Name, string(testJobObject.UID)).
+						ResourceVersion("1").
+						PodSets(*utiltesting.MakePodSet(kueue.DefaultPodSetName, 1).Request(corev1.ResourceCPU, "1").Obj()).
+						Obj()).Build(),
+				jobPodSets:   []kueue.PodSet{*utiltesting.MakePodSet(kueue.DefaultPodSetName, 1).Request(corev1.ResourceCPU, "1").Obj()},
+				jobObject:    testJobObject,
+				jobObjectGVK: testJobGVK,
+			},
+			want: want{
+				workload: utiltesting.MakeWorkload(testJobObject.Name+"-1", testJobObject.Namespace).
+					OwnerReference(testJobGVK, testJobObject.Name, string(testJobObject.UID)).
+					ResourceVersion("1").
+					PodSets(*utiltesting.MakePodSet(kueue.DefaultPodSetName, 1).Request(corev1.ResourceCPU, "1").Obj()).
+					Obj(),
+				compatible: true,
+			},
+		},
+		"OneWorkloadSlice_ReservedQuota_ScaleUp": {
+			args: args{
+				ctx: t.Context(),
+				clnt: testWorkloadClientBuilder().WithObjects(
+					utiltesting.MakeWorkload(testJobObject.Name+"-1", testJobObject.Namespace).
+						OwnerReference(testJobGVK, testJobObject.Name, string(testJobObject.UID)).
+						ResourceVersion("1").
+						PodSets(*utiltesting.MakePodSet(kueue.DefaultPodSetName, 1).Request(corev1.ResourceCPU, "1").Obj()).
+						ReserveQuota(utiltesting.MakeAdmission("default", kueue.DefaultPodSetName).Assignment(corev1.ResourceCPU, "default", "1").AssignmentPodCount(1).Obj()).
+						Obj()).Build(),
+				jobPodSets:   []kueue.PodSet{*utiltesting.MakePodSet(kueue.DefaultPodSetName, 3).Request(corev1.ResourceCPU, "1").Obj()},
+				jobObject:    testJobObject,
+				jobObjectGVK: testJobGVK,
+			},
+			want: want{
+				compatible: true,
+			},
+		},
+		"OneWorkloadSlice_ReservedQuota_ScaleDown": {
+			args: args{
+				ctx: t.Context(),
+				clnt: testWorkloadClientBuilder().WithObjects(
+					utiltesting.MakeWorkload(testJobObject.Name+"-1", testJobObject.Namespace).
+						OwnerReference(testJobGVK, testJobObject.Name, string(testJobObject.UID)).
+						ResourceVersion("1").
+						PodSets(*utiltesting.MakePodSet(kueue.DefaultPodSetName, 3).Request(corev1.ResourceCPU, "1").Obj()).
+						ReserveQuota(utiltesting.MakeAdmission("default", kueue.DefaultPodSetName).Assignment(corev1.ResourceCPU, "default", "1").AssignmentPodCount(3).Obj()).
+						Obj()).Build(),
+				jobPodSets:   []kueue.PodSet{*utiltesting.MakePodSet(kueue.DefaultPodSetName, 1).Request(corev1.ResourceCPU, "1").Obj()},
+				jobObject:    testJobObject,
+				jobObjectGVK: testJobGVK,
+			},
+			want: want{
+				compatible: true,
+				workload: utiltesting.MakeWorkload(testJobObject.Name+"-1", testJobObject.Namespace).
+					OwnerReference(testJobGVK, testJobObject.Name, string(testJobObject.UID)).
+					ResourceVersion("2").
+					PodSets(*utiltesting.MakePodSet(kueue.DefaultPodSetName, 1).Request(corev1.ResourceCPU, "1").Obj()).
+					ReserveQuota(utiltesting.MakeAdmission("default", kueue.DefaultPodSetName).Assignment(corev1.ResourceCPU, "default", "1").AssignmentPodCount(3).Obj()).
+					Obj(),
+			},
+		},
+		"OneWorkloadSlice_UnreservedQuota_ScaleUp": {
+			args: args{
+				ctx: t.Context(),
+				clnt: testWorkloadClientBuilder().WithObjects(
+					utiltesting.MakeWorkload(testJobObject.Name+"-1", testJobObject.Namespace).
+						OwnerReference(testJobGVK, testJobObject.Name, string(testJobObject.UID)).
+						ResourceVersion("1").
+						PodSets(*utiltesting.MakePodSet(kueue.DefaultPodSetName, 1).Request(corev1.ResourceCPU, "1").Obj()).
+						Obj()).Build(),
+				jobPodSets:   []kueue.PodSet{*utiltesting.MakePodSet(kueue.DefaultPodSetName, 3).Request(corev1.ResourceCPU, "1").Obj()},
+				jobObject:    testJobObject,
+				jobObjectGVK: testJobGVK,
+			},
+			want: want{
+				compatible: true,
+				workload: utiltesting.MakeWorkload(testJobObject.Name+"-1", testJobObject.Namespace).
+					OwnerReference(testJobGVK, testJobObject.Name, string(testJobObject.UID)).
+					ResourceVersion("2").
+					PodSets(*utiltesting.MakePodSet(kueue.DefaultPodSetName, 3).Request(corev1.ResourceCPU, "1").Obj()).
+					Obj(),
+			},
+		},
+		"OneWorkloadSlice_UnreservedQuota_ScaleDown": {
+			args: args{
+				ctx: t.Context(),
+				clnt: testWorkloadClientBuilder().WithObjects(
+					utiltesting.MakeWorkload(testJobObject.Name+"-1", testJobObject.Namespace).
+						OwnerReference(testJobGVK, testJobObject.Name, string(testJobObject.UID)).
+						ResourceVersion("1").
+						PodSets(*utiltesting.MakePodSet(kueue.DefaultPodSetName, 3).Request(corev1.ResourceCPU, "1").Obj()).
+						Obj()).Build(),
+				jobPodSets:   []kueue.PodSet{*utiltesting.MakePodSet(kueue.DefaultPodSetName, 1).Request(corev1.ResourceCPU, "1").Obj()},
+				jobObject:    testJobObject,
+				jobObjectGVK: testJobGVK,
+			},
+			want: want{
+				compatible: true,
+				workload: utiltesting.MakeWorkload(testJobObject.Name+"-1", testJobObject.Namespace).
+					OwnerReference(testJobGVK, testJobObject.Name, string(testJobObject.UID)).
+					ResourceVersion("2").
+					PodSets(*utiltesting.MakePodSet(kueue.DefaultPodSetName, 1).Request(corev1.ResourceCPU, "1").Obj()).
+					Obj(),
+			},
+		},
+		"OneWorkloadSlice_UpdateFailure": {
+			args: args{
+				ctx: t.Context(),
+				clnt: testWorkloadClientBuilder().WithObjects(
+					utiltesting.MakeWorkload(testJobObject.Name+"-1", testJobObject.Namespace).
+						OwnerReference(testJobGVK, testJobObject.Name, string(testJobObject.UID)).
+						ResourceVersion("1").
+						PodSets(*utiltesting.MakePodSet(kueue.DefaultPodSetName, 3).Request(corev1.ResourceCPU, "1").Obj()).
+						Obj()).WithInterceptorFuncs(interceptor.Funcs{
+					Update: func(ctx context.Context, client client.WithWatch, obj client.Object, opts ...client.UpdateOption) error {
+						return errors.New("test-update-error")
+					}}).Build(),
+				jobPodSets:   []kueue.PodSet{*utiltesting.MakePodSet(kueue.DefaultPodSetName, 1).Request(corev1.ResourceCPU, "1").Obj()},
+				jobObject:    testJobObject,
+				jobObjectGVK: testJobGVK,
+			},
+			want: want{
+				error:      true,
+				compatible: true,
+			},
+		},
+		//
+		"TwoWorkloads_BothUnreserved_NewIsCurrent": {
+			args: args{
+				ctx: t.Context(),
+				clnt: testWorkloadClientBuilder().WithObjects(
+					utiltesting.MakeWorkload(testJobObject.Name+"-1", testJobObject.Namespace).
+						OwnerReference(testJobGVK, testJobObject.Name, string(testJobObject.UID)).
+						ResourceVersion("1").
+						Creation(fiveMinutesAgo).
+						PodSets(*utiltesting.MakePodSet(kueue.DefaultPodSetName, 1).Request(corev1.ResourceCPU, "1").Obj()).
+						Obj(),
+					utiltesting.MakeWorkload(testJobObject.Name+"-2", testJobObject.Namespace).
+						OwnerReference(testJobGVK, testJobObject.Name, string(testJobObject.UID)).
+						ResourceVersion("1").
+						Creation(now).
+						PodSets(*utiltesting.MakePodSet(kueue.DefaultPodSetName, 3).Request(corev1.ResourceCPU, "1").Obj()).
+						Obj()).
+					WithInterceptorFuncs(interceptor.Funcs{
+						SubResourcePatch: func(ctx context.Context, client client.Client, subResourceName string, obj client.Object, patch client.Patch, opts ...client.SubResourcePatchOption) error {
+							return assertStatusConditionPatch(t, subResourceName, obj, testJobObject.Name+"-1", kueue.WorkloadFinished, kueue.WorkloadFinishedReasonOutOfSync)
+						},
+					}).
+					Build(),
+				jobPodSets:   []kueue.PodSet{*utiltesting.MakePodSet(kueue.DefaultPodSetName, 3).Request(corev1.ResourceCPU, "1").Obj()},
+				jobObject:    testJobObject,
+				jobObjectGVK: testJobGVK,
+			},
+			want: want{
+				compatible: true,
+				workload: utiltesting.MakeWorkload(testJobObject.Name+"-2", testJobObject.Namespace).
+					OwnerReference(testJobGVK, testJobObject.Name, string(testJobObject.UID)).
+					ResourceVersion("1").
+					Creation(now).
+					PodSets(*utiltesting.MakePodSet(kueue.DefaultPodSetName, 3).Request(corev1.ResourceCPU, "1").Obj()).
+					Obj(),
+			},
+		},
+		"TwoWorkloads_BothUnreserved_NewIsCurrent_FailureToPatchOldSliceStatus": {
+			args: args{
+				ctx: t.Context(),
+				clnt: testWorkloadClientBuilder().WithObjects(
+					utiltesting.MakeWorkload(testJobObject.Name+"-1", testJobObject.Namespace).
+						OwnerReference(testJobGVK, testJobObject.Name, string(testJobObject.UID)).
+						ResourceVersion("1").
+						Creation(fiveMinutesAgo).
+						PodSets(*utiltesting.MakePodSet(kueue.DefaultPodSetName, 1).Request(corev1.ResourceCPU, "1").Obj()).
+						Obj(),
+					utiltesting.MakeWorkload(testJobObject.Name+"-2", testJobObject.Namespace).
+						OwnerReference(testJobGVK, testJobObject.Name, string(testJobObject.UID)).
+						ResourceVersion("1").
+						Creation(now).
+						PodSets(*utiltesting.MakePodSet(kueue.DefaultPodSetName, 3).Request(corev1.ResourceCPU, "1").Obj()).
+						Obj()).
+					WithInterceptorFuncs(interceptor.Funcs{
+						SubResourcePatch: func(_ context.Context, client client.Client, subResourceName string, obj client.Object, patch client.Patch, opts ...client.SubResourcePatchOption) error {
+							return errors.New("test-patch-failure")
+						},
+					}).
+					Build(),
+				jobPodSets:   []kueue.PodSet{*utiltesting.MakePodSet(kueue.DefaultPodSetName, 3).Request(corev1.ResourceCPU, "1").Obj()},
+				jobObject:    testJobObject,
+				jobObjectGVK: testJobGVK,
+			},
+			want: want{
+				error:      true,
+				compatible: true,
+			},
+		},
+		"TwoWorkloadSlices_NewIsIncompatible": {
+			args: args{
+				ctx: t.Context(),
+				clnt: testWorkloadClientBuilder().WithObjects(
+					utiltesting.MakeWorkload(testJobObject.Name+"-1", testJobObject.Namespace).
+						OwnerReference(testJobGVK, testJobObject.Name, string(testJobObject.UID)).
+						ResourceVersion("1").
+						Creation(fiveMinutesAgo).
+						PodSets(*utiltesting.MakePodSet(kueue.DefaultPodSetName, 1).Request(corev1.ResourceCPU, "1").Obj()).
+						ReserveQuota(utiltesting.MakeAdmission("default", kueue.DefaultPodSetName).Assignment(corev1.ResourceCPU, "default", "1").AssignmentPodCount(1).Obj()).
+						Obj(),
+					utiltesting.MakeWorkload(testJobObject.Name+"-2", testJobObject.Namespace).
+						OwnerReference(testJobGVK, testJobObject.Name, string(testJobObject.UID)).
+						ResourceVersion("1").
+						Creation(now).
+						PodSets(*utiltesting.MakePodSet(kueue.DefaultPodSetName, 3).Request(corev1.ResourceCPU, "1").Obj()).
+						Obj()).
+					Build(),
+				jobPodSets:   []kueue.PodSet{*utiltesting.MakePodSet("different-key", 1).Request(corev1.ResourceCPU, "1").Obj()},
+				jobObject:    testJobObject,
+				jobObjectGVK: testJobGVK,
+			},
+		},
+		"TwoWorkloads_BothWithReservedQuota": {
+			args: args{
+				ctx: t.Context(),
+				clnt: testWorkloadClientBuilder().WithObjects(
+					utiltesting.MakeWorkload(testJobObject.Name+"-1", testJobObject.Namespace).
+						OwnerReference(testJobGVK, testJobObject.Name, string(testJobObject.UID)).
+						ResourceVersion("1").
+						Creation(fiveMinutesAgo).
+						PodSets(*utiltesting.MakePodSet(kueue.DefaultPodSetName, 1).Request(corev1.ResourceCPU, "1").Obj()).
+						ReserveQuota(utiltesting.MakeAdmission("default", kueue.DefaultPodSetName).Assignment(corev1.ResourceCPU, "default", "1").AssignmentPodCount(1).Obj()).
+						Obj(),
+					utiltesting.MakeWorkload(testJobObject.Name+"-2", testJobObject.Namespace).
+						OwnerReference(testJobGVK, testJobObject.Name, string(testJobObject.UID)).
+						ResourceVersion("1").
+						Creation(now).
+						PodSets(*utiltesting.MakePodSet(kueue.DefaultPodSetName, 3).Request(corev1.ResourceCPU, "1").Obj()).
+						ReserveQuota(utiltesting.MakeAdmission("default", kueue.DefaultPodSetName).Assignment(corev1.ResourceCPU, "default", "1").AssignmentPodCount(3).Obj()).
+						Obj()).
+					Build(),
+				jobPodSets:   []kueue.PodSet{*utiltesting.MakePodSet(kueue.DefaultPodSetName, 5).Request(corev1.ResourceCPU, "1").Obj()},
+				jobObject:    testJobObject,
+				jobObjectGVK: testJobGVK,
+			},
+			want: want{
+				error:      true,
+				compatible: true,
+			},
+		},
+		"TwoWorkloadSlices_NewIsUnreservedAndCurrent": {
+			args: args{
+				ctx: t.Context(),
+				clnt: testWorkloadClientBuilder().WithObjects(
+					utiltesting.MakeWorkload(testJobObject.Name+"-1", testJobObject.Namespace).
+						OwnerReference(testJobGVK, testJobObject.Name, string(testJobObject.UID)).
+						ResourceVersion("1").
+						Creation(fiveMinutesAgo).
+						PodSets(*utiltesting.MakePodSet(kueue.DefaultPodSetName, 1).Request(corev1.ResourceCPU, "1").Obj()).
+						ReserveQuota(utiltesting.MakeAdmission("default", kueue.DefaultPodSetName).Assignment(corev1.ResourceCPU, "default", "1").AssignmentPodCount(1).Obj()).
+						Obj(),
+					utiltesting.MakeWorkload(testJobObject.Name+"-2", testJobObject.Namespace).
+						OwnerReference(testJobGVK, testJobObject.Name, string(testJobObject.UID)).
+						ResourceVersion("1").
+						Creation(now).
+						PodSets(*utiltesting.MakePodSet(kueue.DefaultPodSetName, 3).Request(corev1.ResourceCPU, "1").Obj()).
+						Obj()).Build(),
+				jobPodSets:   []kueue.PodSet{*utiltesting.MakePodSet(kueue.DefaultPodSetName, 3).Request(corev1.ResourceCPU, "1").Obj()},
+				jobObject:    testJobObject,
+				jobObjectGVK: testJobGVK,
+			},
+			want: want{
+				compatible: true,
+				workload: utiltesting.MakeWorkload(testJobObject.Name+"-2", testJobObject.Namespace).
+					OwnerReference(testJobGVK, testJobObject.Name, string(testJobObject.UID)).
+					ResourceVersion("1").
+					Creation(now).
+					PodSets(*utiltesting.MakePodSet(kueue.DefaultPodSetName, 3).Request(corev1.ResourceCPU, "1").Obj()).
+					Obj(),
+			},
+		},
+		"TwoWorkloadSlices_NewIsUnreservedAndOutOfSync_ScaleUp": {
+			args: args{
+				ctx: t.Context(),
+				clnt: testWorkloadClientBuilder().WithObjects(
+					utiltesting.MakeWorkload(testJobObject.Name+"-1", testJobObject.Namespace).
+						OwnerReference(testJobGVK, testJobObject.Name, string(testJobObject.UID)).
+						ResourceVersion("1").
+						Creation(fiveMinutesAgo).
+						PodSets(*utiltesting.MakePodSet(kueue.DefaultPodSetName, 1).Request(corev1.ResourceCPU, "1").Obj()).
+						ReserveQuota(utiltesting.MakeAdmission("default", kueue.DefaultPodSetName).Assignment(corev1.ResourceCPU, "default", "1").AssignmentPodCount(1).Obj()).
+						Obj(),
+					utiltesting.MakeWorkload(testJobObject.Name+"-2", testJobObject.Namespace).
+						OwnerReference(testJobGVK, testJobObject.Name, string(testJobObject.UID)).
+						ResourceVersion("1").
+						Creation(now).
+						PodSets(*utiltesting.MakePodSet(kueue.DefaultPodSetName, 3).Request(corev1.ResourceCPU, "1").Obj()).
+						Obj()).
+					Build(),
+				jobPodSets:   []kueue.PodSet{*utiltesting.MakePodSet(kueue.DefaultPodSetName, 5).Request(corev1.ResourceCPU, "1").Obj()},
+				jobObject:    testJobObject,
+				jobObjectGVK: testJobGVK,
+			},
+			want: want{
+				compatible: true,
+				workload: utiltesting.MakeWorkload(testJobObject.Name+"-2", testJobObject.Namespace).
+					OwnerReference(testJobGVK, testJobObject.Name, string(testJobObject.UID)).
+					ResourceVersion("2").
+					Creation(now).
+					PodSets(*utiltesting.MakePodSet(kueue.DefaultPodSetName, 5).Request(corev1.ResourceCPU, "1").Obj()).
+					Obj(),
+			},
+		},
+		"TwoWorkloadSlices_NewIsUnreservedAndOutOfSync_ScaleDown": {
+			args: args{
+				ctx: t.Context(),
+				clnt: testWorkloadClientBuilder().WithObjects(
+					utiltesting.MakeWorkload(testJobObject.Name+"-1", testJobObject.Namespace).
+						OwnerReference(testJobGVK, testJobObject.Name, string(testJobObject.UID)).
+						ResourceVersion("1").
+						Creation(fiveMinutesAgo).
+						PodSets(*utiltesting.MakePodSet(kueue.DefaultPodSetName, 1).Request(corev1.ResourceCPU, "1").Obj()).
+						ReserveQuota(utiltesting.MakeAdmission("default", kueue.DefaultPodSetName).Assignment(corev1.ResourceCPU, "default", "1").AssignmentPodCount(1).Obj()).
+						Obj(),
+					utiltesting.MakeWorkload(testJobObject.Name+"-2", testJobObject.Namespace).
+						OwnerReference(testJobGVK, testJobObject.Name, string(testJobObject.UID)).
+						ResourceVersion("1").
+						Creation(now).
+						PodSets(*utiltesting.MakePodSet(kueue.DefaultPodSetName, 3).Request(corev1.ResourceCPU, "1").Obj()).
+						Obj()).
+					Build(),
+				jobPodSets:   []kueue.PodSet{*utiltesting.MakePodSet(kueue.DefaultPodSetName, 2).Request(corev1.ResourceCPU, "1").Obj()},
+				jobObject:    testJobObject,
+				jobObjectGVK: testJobGVK,
+			},
+			want: want{
+				compatible: true,
+				workload: utiltesting.MakeWorkload(testJobObject.Name+"-2", testJobObject.Namespace).
+					OwnerReference(testJobGVK, testJobObject.Name, string(testJobObject.UID)).
+					ResourceVersion("2").
+					Creation(now).
+					PodSets(*utiltesting.MakePodSet(kueue.DefaultPodSetName, 2).Request(corev1.ResourceCPU, "1").Obj()).
+					Obj(),
+			},
+		},
+		"TwoWorkloadSlices_NewIsUnreservedAndOutOfSync_UpdateFailure": {
+			args: args{
+				ctx: t.Context(),
+				clnt: testWorkloadClientBuilder().WithObjects(
+					utiltesting.MakeWorkload(testJobObject.Name+"-1", testJobObject.Namespace).
+						OwnerReference(testJobGVK, testJobObject.Name, string(testJobObject.UID)).
+						ResourceVersion("1").
+						Creation(fiveMinutesAgo).
+						PodSets(*utiltesting.MakePodSet(kueue.DefaultPodSetName, 1).Request(corev1.ResourceCPU, "1").Obj()).
+						ReserveQuota(utiltesting.MakeAdmission("default", kueue.DefaultPodSetName).Assignment(corev1.ResourceCPU, "default", "1").AssignmentPodCount(1).Obj()).
+						Obj(),
+					utiltesting.MakeWorkload(testJobObject.Name+"-2", testJobObject.Namespace).
+						OwnerReference(testJobGVK, testJobObject.Name, string(testJobObject.UID)).
+						ResourceVersion("1").
+						Creation(now).
+						PodSets(*utiltesting.MakePodSet(kueue.DefaultPodSetName, 3).Request(corev1.ResourceCPU, "1").Obj()).
+						Obj()).
+					WithInterceptorFuncs(interceptor.Funcs{
+						Update: func(_ context.Context, _ client.WithWatch, obj client.Object, _ ...client.UpdateOption) error {
+							// Assert that we are updating correct workload slice.
+							if obj.GetName() != testJobObject.Name+"-2" {
+								t.Errorf("unexptected workload update: %v", obj)
+							}
+							return errors.New("test-update-error")
+						},
+					}).
+					Build(),
+				jobPodSets:   []kueue.PodSet{*utiltesting.MakePodSet(kueue.DefaultPodSetName, 2).Request(corev1.ResourceCPU, "1").Obj()},
+				jobObject:    testJobObject,
+				jobObjectGVK: testJobGVK,
+			},
+			want: want{
+				error:      true,
+				compatible: true,
+			},
+		},
+		//
+		"MoreThanTwoWorkloadSlices": {
+			args: args{
+				ctx: t.Context(),
+				clnt: testWorkloadClientBuilder().WithObjects(
+					testWorkload(testJobObject.Name+"-1", testResourceVersion, testPodSets(1)),
+					testWorkload(testJobObject.Name+"-2", testResourceVersion+1, testPodSets(2)),
+					testWorkload(testJobObject.Name+"-3", testResourceVersion+1, testPodSets(3))).
+					Build(),
+				jobPodSets:   testPodSets(3),
+				jobObject:    testJobObject,
+				jobObjectGVK: testJobGVK,
+			},
+			want: want{
+				error:      true,
+				compatible: true,
+			},
+		},
+	}
+	for name, tt := range tests {
+		t.Run(name, func(t *testing.T) {
+			gotWorkload, gotCompatible, gotError := EnsureWorkloadSlices(tt.args.ctx, tt.args.clnt, tt.args.jobPodSets, tt.args.jobObject, tt.args.jobObjectGVK)
+			if (gotError != nil) != tt.want.error {
+				t.Errorf("EnsureWorkloadSlices() error = %v, wantErr %v", gotError, tt.want.error)
+				return
+			}
+			if diff := cmp.Diff(tt.want.workload, gotWorkload, cmpopts.EquateApproxTime(time.Second)); diff != "" {
+				t.Errorf("EnsureWorkloadSlices() (-want,+got):\n%s", diff)
+			}
+			if gotCompatible != tt.want.compatible {
+				t.Errorf("EnsureWorkloadSlices() compatible = %v, want %v", gotCompatible, tt.want.compatible)
+			}
+		})
+	}
+}
+
+func Test_StartWorkloadSlicePods(t *testing.T) {
+	clientBuilder := func() *fake.ClientBuilder {
+		return fake.NewClientBuilder().WithScheme(scheme.Scheme).
+			WithIndex(&corev1.Pod{}, indexer.OwnerReferenceUID, indexer.IndexOwnerUID)
+	}
+	testPod := func(name, resourceVersion string, owner client.Object, schedulingGates ...corev1.PodSchedulingGate) corev1.Pod {
+		return corev1.Pod{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: name,
+				OwnerReferences: []metav1.OwnerReference{
+					{
+						APIVersion: owner.GetObjectKind().GroupVersionKind().GroupVersion().String(),
+						Kind:       owner.GetObjectKind().GroupVersionKind().Kind,
+						Name:       owner.GetName(),
+						UID:        owner.GetUID(),
+					},
+				},
+				ResourceVersion: resourceVersion,
+			},
+			Spec: corev1.PodSpec{
+				SchedulingGates: schedulingGates,
+			},
+		}
+	}
+
+	type args struct {
+		ctx    context.Context
+		clnt   client.Client
+		object client.Object
+	}
+	tests := map[string]struct {
+		args     args
+		wantErr  bool
+		wantPods *corev1.PodList
+	}{
+		"FailureToListPods": {
+			args: args{
+				ctx: t.Context(),
+				clnt: clientBuilder().WithInterceptorFuncs(interceptor.Funcs{
+					List: func(_ context.Context, _ client.WithWatch, _ client.ObjectList, _ ...client.ListOption) error {
+						return errors.New("test-list-pods-error")
+					},
+				}).Build(),
+				object: testJobObject,
+			},
+			wantErr: true,
+		},
+		"NoPods": {
+			args: args{
+				ctx:    t.Context(),
+				clnt:   clientBuilder().Build(),
+				object: testJobObject,
+			},
+		},
+		"ProcessPods": {
+			args: args{
+				ctx: t.Context(),
+				clnt: clientBuilder().WithLists(&corev1.PodList{
+					Items: []corev1.Pod{
+						// Un-gated pod should remain un-gated, i.e., no change.
+						testPod("test-one", "100", testJobObject),
+						// Gated pod - gate should be removed.
+						testPod("test-two", "200", testJobObject, corev1.PodSchedulingGate{Name: kueue.ElasticJobSchedulingGate}),
+						// Gated with some other gate -
+						testPod("test-three", "300", testJobObject, corev1.PodSchedulingGate{Name: kueue.ElasticJobSchedulingGate}, corev1.PodSchedulingGate{Name: kueuealpha.TopologySchedulingGate}),
+						// Other gated pod (not for this job)
+						testPod("other-pod", "400", &batchv1.Job{
+							ObjectMeta: metav1.ObjectMeta{
+								Name: "other-job",
+							},
+						}, corev1.PodSchedulingGate{Name: kueue.ElasticJobSchedulingGate}),
+					},
+				}).Build(),
+				object: testJobObject,
+			},
+			wantPods: &corev1.PodList{
+				Items: []corev1.Pod{
+					testPod("test-one", "100", testJobObject),
+					// Gated pod - gate removed (resource version increase).
+					testPod("test-two", "201", testJobObject),
+					// Gated with some other gate - other gate remains (resource version increased).
+					testPod("test-three", "301", testJobObject, corev1.PodSchedulingGate{Name: kueuealpha.TopologySchedulingGate}),
+					// Other gated pod (not for this job) - no change.
+					testPod("other-pod", "400", &batchv1.Job{
+						ObjectMeta: metav1.ObjectMeta{
+							Name: "other-job",
+						},
+					}, corev1.PodSchedulingGate{Name: kueue.ElasticJobSchedulingGate}),
+				},
+			},
+		},
+		"FailureUpdatingPod": {
+			args: args{
+				ctx: t.Context(),
+				clnt: clientBuilder().WithLists(&corev1.PodList{
+					Items: []corev1.Pod{
+						testPod("test", "100", testJobObject, corev1.PodSchedulingGate{Name: kueue.ElasticJobSchedulingGate}),
+					},
+				}).WithInterceptorFuncs(interceptor.Funcs{
+					Patch: func(_ context.Context, _ client.WithWatch, _ client.Object, _ client.Patch, _ ...client.PatchOption) error {
+						return errors.New("test-pod-patch-error")
+					},
+				}).
+					Build(),
+				object: testJobObject,
+			},
+			wantErr: true,
+		},
+	}
+	for name, tt := range tests {
+		t.Run(name, func(t *testing.T) {
+			if err := StartWorkloadSlicePods(tt.args.ctx, tt.args.clnt, tt.args.object); (err != nil) != tt.wantErr {
+				t.Errorf("StartWorkloadSlicePods() error = %v, wantErr %v", err, tt.wantErr)
+			}
+			if tt.wantPods == nil {
+				return
+			}
+
+			pods := &corev1.PodList{}
+			if err := tt.args.clnt.List(tt.args.ctx, pods); err != nil {
+				t.Errorf("unexpected list error: %v", err)
+			}
+			if diff := cmp.Diff(pods.Items, tt.wantPods.Items, cmpopts.SortSlices(func(a, b corev1.Pod) bool {
+				return a.Name < b.Name
+			})); diff != "" {
+				t.Errorf("StartWorkloadSlicePods() pod-validaion: got(-),want(+): %s", diff)
+			}
+		})
+	}
+}
+
+func TestReplacedWorkloadSlice(t *testing.T) {
+	type args struct {
+		wl   *workload.Info
+		snap *cache.Snapshot
+	}
+	type want struct {
+		wl      *workload.Info
+		targets []*preemption.Target
+	}
+
+	tests := map[string]struct {
+		featureEnabled bool
+		args           args
+		want           want
+	}{
+		"FeatureNotEnabled": {},
+		"EdgeCase_WorkloadIsNil": {
+			featureEnabled: true,
+		},
+		"EdgeCase_SnapshotIsNil": {
+			featureEnabled: true,
+			args: args{
+				wl: workload.NewInfo(utiltesting.MakeWorkload("test", "default").Obj()),
+			},
+		},
+		"WorkloadWithoutReplacementAnnotation": {
+			featureEnabled: true,
+			args: args{
+				wl:   workload.NewInfo(utiltesting.MakeWorkload("test", "default").Obj()),
+				snap: &cache.Snapshot{},
+			},
+		},
+		"ReplacedWorkloadIsNotFound_MissingClusterQueue": {
+			featureEnabled: true,
+			args: args{
+				wl: workload.NewInfo(utiltesting.MakeWorkload("test-new", "default").
+					Annotation(WorkloadSliceReplacementFor, "test-old").
+					Obj()),
+				snap: &cache.Snapshot{
+					Manager: hierarchy.NewManagerForTest(
+						map[kueue.CohortReference]*cache.CohortSnapshot{},
+						map[kueue.ClusterQueueReference]*cache.ClusterQueueSnapshot{}),
+				},
+			},
+		},
+		"EdgeCase_ReplacedWorkloadIsNotFound_NotInClusterQueue": {
+			featureEnabled: true,
+			args: args{
+				wl: workload.NewInfo(utiltesting.MakeWorkload("test-new", "default").
+					Annotation(WorkloadSliceReplacementFor, "test-old").
+					Admission(utiltesting.MakeAdmission("default").Obj()).
+					Obj()),
+				snap: &cache.Snapshot{
+					Manager: hierarchy.NewManagerForTest(
+						map[kueue.CohortReference]*cache.CohortSnapshot{},
+						map[kueue.ClusterQueueReference]*cache.ClusterQueueSnapshot{
+							"default": {},
+						}),
+				},
+			},
+		},
+		"ReplacedWorkloadIsFound": {
+			featureEnabled: true,
+			args: args{
+				wl: workload.NewInfo(utiltesting.MakeWorkload("test-new", "default").
+					Annotation(WorkloadSliceReplacementFor, "test-old").
+					Admission(utiltesting.MakeAdmission("default").Obj()).
+					Obj()),
+				snap: &cache.Snapshot{
+					Manager: hierarchy.NewManagerForTest(
+						map[kueue.CohortReference]*cache.CohortSnapshot{},
+						map[kueue.ClusterQueueReference]*cache.ClusterQueueSnapshot{
+							"default": {
+								Workloads: map[workload.Reference]*workload.Info{
+									"test-old": workload.NewInfo(utiltesting.MakeWorkload("test-old", "default").Obj()),
+								},
+							},
+						}),
+				},
+			},
+			want: want{
+				wl: workload.NewInfo(utiltesting.MakeWorkload("test-old", "default").Obj()),
+				targets: []*preemption.Target{
+					{WorkloadInfo: workload.NewInfo(utiltesting.MakeWorkload("test-old", "default").Obj())},
+				},
+			},
+		},
+	}
+	for name, tt := range tests {
+		t.Run(name, func(t *testing.T) {
+			features.SetFeatureGateDuringTest(t, features.ElasticJobsViaWorkloadSlices, tt.featureEnabled)
+			targets, wl := ReplacedWorkloadSlice(tt.args.wl, tt.args.snap)
+			if diff := cmp.Diff(tt.want.targets, targets); diff != "" {
+				t.Errorf("ReplacedWorkloadSlice() targets (+want,-got):\n%s", diff)
+			}
+			if diff := cmp.Diff(tt.want.wl, wl); diff != "" {
+				t.Errorf("ReplacedWorkloadSlice() workload (+want,-got):\n%s", diff)
+			}
+		})
+	}
+}
diff --git a/test/integration/singlecluster/controller/jobs/job/job_controller_test.go b/test/integration/singlecluster/controller/jobs/job/job_controller_test.go
index 01926d193..56c5970eb 100644
--- a/test/integration/singlecluster/controller/jobs/job/job_controller_test.go
+++ b/test/integration/singlecluster/controller/jobs/job/job_controller_test.go
@@ -31,6 +31,7 @@ import (
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
+	utilfeature "k8s.io/apiserver/pkg/util/feature"
 	"k8s.io/utils/clock"
 	"k8s.io/utils/ptr"
 	ctrl "sigs.k8s.io/controller-runtime"
@@ -47,6 +48,7 @@ import (
 	testingjob "sigs.k8s.io/kueue/pkg/util/testingjobs/job"
 	testingnode "sigs.k8s.io/kueue/pkg/util/testingjobs/node"
 	"sigs.k8s.io/kueue/pkg/workload"
+	"sigs.k8s.io/kueue/pkg/workloadslicing"
 	"sigs.k8s.io/kueue/test/integration/framework"
 	"sigs.k8s.io/kueue/test/util"
 )
@@ -3085,3 +3087,145 @@ var _ = ginkgo.Describe("Job controller with ObjectRetentionPolicies", ginkgo.Or
 		})
 	})
 })
+
+var _ = ginkgo.Describe("Job with elastic jobs via workload-slices support", ginkgo.Ordered, ginkgo.ContinueOnFailure, func() {
+	var (
+		ns             *corev1.Namespace
+		resourceFlavor *kueue.ResourceFlavor
+		clusterQueue   *kueue.ClusterQueue
+		localQueue     *kueue.LocalQueue
+	)
+
+	ginkgo.BeforeAll(func() {
+		gomega.Expect(utilfeature.DefaultMutableFeatureGate.SetFromMap(map[string]bool{string(features.ElasticJobsViaWorkloadSlices): true})).Should(gomega.Succeed())
+		fwk.StartManager(ctx, cfg, managerAndControllersSetup(false, true, nil))
+	})
+	ginkgo.AfterAll(func() {
+		fwk.StopManager(ctx)
+	})
+
+	ginkgo.BeforeEach(func() {
+		ns = util.CreateNamespaceFromPrefixWithLog(ctx, k8sClient, "core-")
+
+		resourceFlavor = testing.MakeResourceFlavor("default").Obj()
+		util.MustCreate(ctx, k8sClient, resourceFlavor)
+
+		clusterQueue = testing.MakeClusterQueue("default").
+			ResourceGroup(*testing.MakeFlavorQuotas("default").Resource(corev1.ResourceCPU, "5").Obj()).
+			Obj()
+		util.MustCreate(ctx, k8sClient, clusterQueue)
+
+		localQueue = testing.MakeLocalQueue("default", ns.Name).ClusterQueue(clusterQueue.Name).Obj()
+		util.MustCreate(ctx, k8sClient, localQueue)
+	})
+
+	ginkgo.AfterEach(func() {
+		gomega.Expect(util.DeleteNamespace(ctx, k8sClient, ns)).To(gomega.Succeed())
+		util.ExpectObjectToBeDeleted(ctx, k8sClient, clusterQueue, true)
+		util.ExpectObjectToBeDeleted(ctx, k8sClient, localQueue, true)
+		util.ExpectObjectToBeDeleted(ctx, k8sClient, resourceFlavor, true)
+	})
+
+	ginkgo.When("ElasticJobsViaWorkloadSlices is enabled", func() {
+		ginkgo.When("Job is opted in", func() {
+			var (
+				testJob         *batchv1.Job
+				testJobWorkload *kueue.Workload
+			)
+
+			ginkgo.It("Should support job scale-down and scale-up", func() {
+				testJob = testingjob.MakeJob("job1", ns.Name).
+					SetAnnotation(workloadslicing.EnabledAnnotationKey, workloadslicing.EnabledAnnotationValue).
+					Queue(kueue.LocalQueueName(localQueue.Name)).
+					Request(corev1.ResourceCPU, "100m").
+					Parallelism(3).
+					Completions(3).
+					Obj()
+
+				ginkgo.By("creating a job")
+				util.MustCreate(ctx, k8sClient, testJob)
+
+				ginkgo.By("admitting the job's workload")
+				gomega.Eventually(func(g gomega.Gomega) {
+					workloads := &kueue.WorkloadList{}
+					g.Expect(k8sClient.List(ctx, workloads, client.InNamespace(testJob.Namespace))).Should(gomega.Succeed())
+					g.Expect(workloads.Items).Should(gomega.HaveLen(1))
+					testJobWorkload = &workloads.Items[0]
+					util.ExpectWorkloadsToBeAdmitted(ctx, k8sClient, testJobWorkload)
+				}, util.Timeout, util.Interval).Should(gomega.Succeed())
+
+				ginkgo.By("the job is unsuspended")
+				gomega.Eventually(func(g gomega.Gomega) {
+					g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(testJob), testJob)).Should(gomega.Succeed())
+					g.Expect(ptr.Deref(testJob.Spec.Suspend, false)).Should(gomega.BeFalse())
+				}, util.Timeout, util.Interval).Should(gomega.Succeed())
+
+				ginkgo.By("resource flavor utilization is correctly recorded")
+				gomega.Eventually(func(g gomega.Gomega) {
+					cq := &kueue.ClusterQueue{}
+					g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(clusterQueue), cq)).Should(gomega.Succeed())
+					g.Expect(len(cq.Status.FlavorsUsage)).Should(gomega.BeEquivalentTo(1))
+					g.Expect(len(cq.Status.FlavorsUsage[0].Resources)).Should(gomega.BeEquivalentTo(1))
+					g.Expect(cq.Status.FlavorsUsage[0].Resources[0].Total).Should(gomega.BeEquivalentTo(resource.MustParse("300m")))
+				}).Should(gomega.Succeed())
+
+				ginkgo.By("reducing the job's parallelism to emulate scale-down operation")
+				gomega.Eventually(func(g gomega.Gomega) {
+					testJob.Spec.Parallelism = ptr.To(int32(1))
+					g.Expect(k8sClient.Update(ctx, testJob)).Should(gomega.Succeed())
+				}).Should(gomega.Succeed())
+
+				ginkgo.By("resource flavor utilization is correctly updated")
+				gomega.Eventually(func(g gomega.Gomega) {
+					cq := &kueue.ClusterQueue{}
+					g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(clusterQueue), cq)).Should(gomega.Succeed())
+					g.Expect(len(cq.Status.FlavorsUsage)).Should(gomega.BeEquivalentTo(1))
+					g.Expect(len(cq.Status.FlavorsUsage[0].Resources)).Should(gomega.BeEquivalentTo(1))
+					g.Expect(cq.Status.FlavorsUsage[0].Resources[0].Total).Should(gomega.BeEquivalentTo(resource.MustParse("100m")))
+				}, util.Timeout, util.Interval).Should(gomega.Succeed())
+
+				ginkgo.By("assert the job's workload is updated and still admitted")
+				gomega.Eventually(func(g gomega.Gomega) {
+					workloads := &kueue.WorkloadList{}
+					g.Expect(k8sClient.List(ctx, workloads, client.InNamespace(testJob.Namespace))).Should(gomega.Succeed())
+					g.Expect(workloads.Items).Should(gomega.HaveLen(1))
+					util.ExpectWorkloadsToBeAdmitted(ctx, k8sClient, testJobWorkload)
+					g.Expect(workloads.Items[0].Spec.PodSets[0].Count).Should(gomega.BeEquivalentTo(int32(1)))
+					g.Expect(workloads.Items[0].UID).Should(gomega.BeEquivalentTo(testJobWorkload.UID))
+					testJobWorkload = &workloads.Items[0]
+				}).Should(gomega.Succeed())
+
+				ginkgo.By("increasing the job's parallelism to emulate scale-up operation")
+				gomega.Eventually(func(g gomega.Gomega) {
+					g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(testJob), testJob)).Should(gomega.Succeed())
+					testJob.Spec.Parallelism = ptr.To(int32(2))
+					g.Expect(k8sClient.Update(ctx, testJob)).Should(gomega.Succeed())
+				}).Should(gomega.Succeed())
+
+				ginkgo.By("resource flavor utilization is correctly updated")
+				gomega.Eventually(func(g gomega.Gomega) {
+					cq := &kueue.ClusterQueue{}
+					g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(clusterQueue), cq)).Should(gomega.Succeed())
+					g.Expect(len(cq.Status.FlavorsUsage)).Should(gomega.BeEquivalentTo(1))
+					g.Expect(len(cq.Status.FlavorsUsage[0].Resources)).Should(gomega.BeEquivalentTo(1))
+					g.Expect(cq.Status.FlavorsUsage[0].Resources[0].Total).Should(gomega.BeEquivalentTo(resource.MustParse("200m")))
+				}, util.Timeout, util.Interval).Should(gomega.Succeed())
+
+				ginkgo.By("old workload is finished and new workload is admitted")
+				gomega.Eventually(func(g gomega.Gomega) {
+					workloads := &kueue.WorkloadList{}
+					g.Expect(k8sClient.List(ctx, workloads, client.InNamespace(testJob.Namespace))).Should(gomega.Succeed())
+					g.Expect(workloads.Items).Should(gomega.HaveLen(2))
+					for i := range workloads.Items {
+						if workloads.Items[i].Name == testJobWorkload.Name {
+							g.Expect(workload.IsFinished(&workloads.Items[i])).Should(gomega.BeTrue())
+							continue
+						}
+						testJobWorkload = &workloads.Items[i]
+						util.ExpectWorkloadsToBeAdmitted(ctx, k8sClient, testJobWorkload)
+					}
+				}, util.Timeout, util.Interval).Should(gomega.Succeed())
+			})
+		})
+	})
+})
diff --git a/test/integration/singlecluster/webhook/core/workload_test.go b/test/integration/singlecluster/webhook/core/workload_test.go
index 1c0ad3952..80d941a88 100644
--- a/test/integration/singlecluster/webhook/core/workload_test.go
+++ b/test/integration/singlecluster/webhook/core/workload_test.go
@@ -32,6 +32,7 @@ import (
 
 	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta1"
 	"sigs.k8s.io/kueue/pkg/constants"
+	"sigs.k8s.io/kueue/pkg/features"
 	"sigs.k8s.io/kueue/pkg/util/testing"
 	"sigs.k8s.io/kueue/pkg/workload"
 	"sigs.k8s.io/kueue/test/util"
@@ -444,7 +445,6 @@ var _ = ginkgo.Describe("Workload validating webhook", func() {
 					gomega.Expect(util.SetQuotaReservation(ctx, k8sClient, workload, testing.MakeAdmission("cq").Obj())).Should(gomega.Succeed())
 					util.SyncAdmittedConditionForWorkloads(ctx, k8sClient, workload)
 				}
-
 				gomega.Eventually(func(g gomega.Gomega) {
 					var newWL kueue.Workload
 					g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(workload), &newWL)).To(gomega.Succeed())
@@ -966,5 +966,68 @@ var _ = ginkgo.Describe("Workload validating webhook", func() {
 				g.Expect(k8sClient.Status().Update(ctx, wl)).Should(testing.BeForbiddenError())
 			}, util.Timeout, util.Interval).Should(gomega.Succeed())
 		})
+
+		ginkgo.It("Should not allow workload podSets count increase even when ElasticJobsViaWorkloadSlices feature gate is enabled", func() {
+			features.SetFeatureGateDuringTest(ginkgo.GinkgoTB(), features.ElasticJobsViaWorkloadSlices, true)
+
+			ginkgo.By("Creating a new Workload")
+			wl := testing.MakeWorkload(workloadName, ns.Name).
+				PodSets(*testing.MakePodSet(kueue.DefaultPodSetName, 1).
+					Request(corev1.ResourceCPU, "1").
+					Obj()).
+				Obj()
+			util.MustCreate(ctx, k8sClient, wl)
+
+			ginkgo.By("Admitting the Workload")
+			gomega.Eventually(func(g gomega.Gomega) {
+				g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(wl), wl)).To(gomega.Succeed())
+				admission := testing.MakeAdmission("default").
+					Assignment(corev1.ResourceCPU, "default", "1").
+					AssignmentPodCount(1).
+					Obj()
+				workload.SetQuotaReservation(wl, admission, realClock)
+				wl.Status.Admission = admission
+
+				g.Expect(k8sClient.Status().Update(ctx, wl)).Should(gomega.Succeed())
+			}, util.Timeout, util.Interval).Should(gomega.Succeed())
+
+			ginkgo.By("Updating PodSets count")
+			gomega.Eventually(func(g gomega.Gomega) {
+				g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(wl), wl)).To(gomega.Succeed())
+				wl.Spec.PodSets[0].Count++ // Increase from 1 -> 2.
+				g.Expect(k8sClient.Update(ctx, wl)).Should(testing.BeForbiddenError())
+			}, util.Timeout, util.Interval).Should(gomega.Succeed())
+		})
+		ginkgo.It("Should allow workload podSets count decrease when ElasticJobsViaWorkloadSlices feature gate is enabled", func() {
+			features.SetFeatureGateDuringTest(ginkgo.GinkgoTB(), features.ElasticJobsViaWorkloadSlices, true)
+
+			ginkgo.By("Creating a new Workload")
+			wl := testing.MakeWorkload(workloadName, ns.Name).
+				PodSets(*testing.MakePodSet(kueue.DefaultPodSetName, 10).
+					Request(corev1.ResourceCPU, "1").
+					Obj()).
+				Obj()
+			util.MustCreate(ctx, k8sClient, wl)
+
+			ginkgo.By("Admitting the Workload")
+			gomega.Eventually(func(g gomega.Gomega) {
+				g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(wl), wl)).To(gomega.Succeed())
+				admission := testing.MakeAdmission("default").
+					Assignment(corev1.ResourceCPU, "default", "1").
+					AssignmentPodCount(10).
+					Obj()
+				workload.SetQuotaReservation(wl, admission, realClock)
+				wl.Status.Admission = admission
+
+				g.Expect(k8sClient.Status().Update(ctx, wl)).Should(gomega.Succeed())
+			}, util.Timeout, util.Interval).Should(gomega.Succeed())
+
+			ginkgo.By("Updating PodSets count")
+			gomega.Eventually(func(g gomega.Gomega) {
+				g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(wl), wl)).To(gomega.Succeed())
+				wl.Spec.PodSets[0].Count = 5 // Decrease from 10 -> 5.
+				g.Expect(k8sClient.Update(ctx, wl)).Should(gomega.Succeed())
+			}, util.Timeout, util.Interval).Should(gomega.Succeed())
+		})
 	})
 })