diff --git a/subiquity/common/apidef.py b/subiquity/common/apidef.py
index b89c9e44..b2285cd8 100644
--- a/subiquity/common/apidef.py
+++ b/subiquity/common/apidef.py
@@ -32,7 +32,9 @@ from subiquity.common.types import (
     Disk,
     ErrorReportRef,
     GuidedChoice,
+    GuidedChoiceV2,
     GuidedStorageResponse,
+    GuidedStorageResponseV2,
     KeyboardSetting,
     KeyboardSetup,
     IdentityData,
@@ -253,12 +255,18 @@ class API:
             def GET() -> List[Disk]: ...
 
         class v2:
+            class deprecated:
+                class guided:
+                    def POST(data: Payload[GuidedChoice]) \
+                        -> StorageResponseV2: ...
+
             def GET() -> StorageResponseV2: ...
             def POST() -> StorageResponseV2: ...
 
             class guided:
-                def POST(data: Payload[GuidedChoice]) \
-                    -> StorageResponseV2: ...
+                def GET() -> GuidedStorageResponseV2: ...
+                def POST(data: Payload[GuidedChoiceV2]) \
+                    -> GuidedStorageResponseV2: ...
 
             class reset:
                 def POST() -> StorageResponseV2: ...
diff --git a/subiquity/common/filesystem/boot.py b/subiquity/common/filesystem/boot.py
index 5cb2bcf8..69db4413 100644
--- a/subiquity/common/filesystem/boot.py
+++ b/subiquity/common/filesystem/boot.py
@@ -21,6 +21,7 @@ import attr
 
 from subiquity.common.filesystem import gaps, sizes
 from subiquity.models.filesystem import (
+    align_up,
     Disk,
     Raid,
     Bootloader,
@@ -84,19 +85,22 @@ class CreatePartPlan(MakeBootDevicePlan):
             self.gap.device, self.gap, self.spec, **self.args)
 
 
-def _no_preserve_part(inst, field, part):
-    assert not part.preserve
+def _can_resize_part(inst, field, part):
+    assert not part.preserve or inst.allow_resize_preserved
 
 
 @attr.s(auto_attribs=True)
 class ResizePlan(MakeBootDevicePlan):
     """Resize a partition."""
 
-    part: object = attr.ib(validator=_no_preserve_part)
+    part: object = attr.ib(validator=_can_resize_part)
     size_delta: int = 0
+    allow_resize_preserved: bool = False
 
     def apply(self, manipulator):
         self.part.size += self.size_delta
+        if self.part.preserve:
+            self.part.resize = True
 
 
 def _no_preserve_parts(inst, field, parts):
@@ -138,6 +142,14 @@ class MountBootEfiPlan(MakeBootDevicePlan):
         manipulator._mount_esp(self.part)
 
 
+@attr.s(auto_attribs=True)
+class NoOpBootPlan(MakeBootDevicePlan):
+    """Do nothing, successfully"""
+
+    def apply(self, manipulator):
+        pass
+
+
 @attr.s(auto_attribs=True)
 class MultiStepPlan(MakeBootDevicePlan):
     """Execute several MakeBootDevicePlans in sequence."""
@@ -203,7 +215,7 @@ def get_boot_device_plan_bios(device) -> Optional[MakeBootDevicePlan]:
         ])
 
 
-def get_add_part_plan(device, *, spec, args):
+def get_add_part_plan(device, *, spec, args, resize_partition=None):
     size = spec['size']
     partitions = device.partitions()
 
@@ -212,6 +224,21 @@ def get_add_part_plan(device, *, spec, args):
     if gaps.largest_gap_size(device) >= size:
         create_part_plan.gap = gaps.largest_gap(device).split(size)[0]
         return create_part_plan
+    elif resize_partition is not None:
+        if size > resize_partition.size - resize_partition.estimated_min_size:
+            return None
+
+        offset = resize_partition.offset + resize_partition.size - size
+        create_part_plan.gap = gaps.Gap(
+                device=device, offset=offset, size=size)
+        return MultiStepPlan(plans=[
+            ResizePlan(
+                part=resize_partition,
+                size_delta=-size,
+                allow_resize_preserved=True,
+                ),
+            create_part_plan,
+            ])
     else:
         new_parts = [p for p in partitions if not p.preserve]
         if not new_parts:
@@ -232,7 +259,7 @@ def get_add_part_plan(device, *, spec, args):
             ])
 
 
-def get_boot_device_plan_uefi(device):
+def get_boot_device_plan_uefi(device, resize_partition):
     for part in device.partitions():
         if is_esp(part):
             plans = [SetAttrPlan(part, 'grub_device', True)]
@@ -240,16 +267,18 @@ def get_boot_device_plan_uefi(device):
                 plans.append(MountBootEfiPlan(part))
             return MultiStepPlan(plans=plans)
 
-    size = sizes.get_efi_size(device.size)
+    part_align = device.alignment_data().part_align
+    size = align_up(sizes.get_efi_size(device.size), part_align)
     spec = dict(size=size, fstype='fat32', mount=None)
     if device._m._mount_for_path("/boot/efi") is None:
         spec['mount'] = '/boot/efi'
 
     return get_add_part_plan(
-        device, spec=spec, args=dict(flag='boot', grub_device=True))
+        device, spec=spec, args=dict(flag='boot', grub_device=True),
+        resize_partition=resize_partition)
 
 
-def get_boot_device_plan_prep(device):
+def get_boot_device_plan_prep(device, resize_partition):
     for part in device.partitions():
         if part.flag == "prep":
             return MultiStepPlan(plans=[
@@ -260,22 +289,29 @@ def get_boot_device_plan_prep(device):
     return get_add_part_plan(
         device,
         spec=dict(size=sizes.PREP_GRUB_SIZE_BYTES, fstype=None, mount=None),
-        args=dict(flag='prep', grub_device=True, wipe='zero'))
+        args=dict(flag='prep', grub_device=True, wipe='zero'),
+        resize_partition=resize_partition)
 
 
-def get_boot_device_plan(device):
+def get_boot_device_plan(device, resize_partition=None):
     bl = device._m.bootloader
     if bl == Bootloader.BIOS:
+        # we don't attempt resize_partition with BIOS,
+        # a move might help but a resize alone won't
+        # and we don't move preserved partitions.
         return get_boot_device_plan_bios(device)
     if bl == Bootloader.UEFI:
-        return get_boot_device_plan_uefi(device)
+        return get_boot_device_plan_uefi(device, resize_partition)
     if bl == Bootloader.PREP:
-        return get_boot_device_plan_prep(device)
+        return get_boot_device_plan_prep(device, resize_partition)
+    if bl == Bootloader.NONE:
+        return NoOpBootPlan()
     raise Exception(f'unexpected bootloader {bl} here')
 
 
 @functools.singledispatch
-def can_be_boot_device(device, *, with_reformatting=False):
+def can_be_boot_device(device, *,
+                       resize_partition=None, with_reformatting=False):
     """Can `device` be made into a boot device?
 
     If with_reformatting=True, return true if the device can be made
@@ -285,14 +321,17 @@ def can_be_boot_device(device, *, with_reformatting=False):
 
 
 @can_be_boot_device.register(Disk)
-def _can_be_boot_device_disk(disk, *, with_reformatting=False):
+def _can_be_boot_device_disk(disk, *,
+                             resize_partition=None, with_reformatting=False):
     if with_reformatting:
         return True
-    return get_boot_device_plan(disk) is not None
+    plan = get_boot_device_plan(disk, resize_partition=resize_partition)
+    return plan is not None
 
 
 @can_be_boot_device.register(Raid)
-def _can_be_boot_device_raid(raid, *, with_reformatting=False):
+def _can_be_boot_device_raid(raid, *,
+                             resize_partition=None, with_reformatting=False):
     bl = raid._m.bootloader
     if bl != Bootloader.UEFI:
         return False
@@ -300,7 +339,8 @@ def _can_be_boot_device_raid(raid, *, with_reformatting=False):
         return False
     if with_reformatting:
         return True
-    return get_boot_device_plan_uefi(raid) is not None
+    plan = get_boot_device_plan_uefi(raid, resize_partition=resize_partition)
+    return plan is not None
 
 
 @functools.singledispatch
diff --git a/subiquity/common/filesystem/gaps.py b/subiquity/common/filesystem/gaps.py
index 911a1b98..f569e71a 100644
--- a/subiquity/common/filesystem/gaps.py
+++ b/subiquity/common/filesystem/gaps.py
@@ -230,3 +230,14 @@ def at_offset(device, offset):
             if pg.offset == offset:
                 return pg
     return None
+
+
+def within(device, gap):
+    """Find the first gap that is contained wholly inside the supplied gap."""
+    gap_end = gap.offset + gap.size
+    for pg in parts_and_gaps(device):
+        if isinstance(pg, Gap):
+            pg_end = pg.offset + pg.size
+            if pg.offset >= gap.offset and pg_end <= gap_end:
+                return pg
+    return None
diff --git a/subiquity/common/filesystem/sizes.py b/subiquity/common/filesystem/sizes.py
index b63c9dbf..851f0a70 100644
--- a/subiquity/common/filesystem/sizes.py
+++ b/subiquity/common/filesystem/sizes.py
@@ -115,6 +115,7 @@ def calculate_guided_resize(part_min: int, part_size: int, install_min: int,
     raw_recommended = math.ceil(resize_window * ratio) + other_min
     recommended = align_up(raw_recommended, part_align)
     return GuidedResizeValues(
+            install_max=plausible_free_space,
             minimum=other_min, recommended=recommended, maximum=other_max)
 
 
diff --git a/subiquity/common/filesystem/tests/test_gaps.py b/subiquity/common/filesystem/tests/test_gaps.py
index 95c502e9..08c5677e 100644
--- a/subiquity/common/filesystem/tests/test_gaps.py
+++ b/subiquity/common/filesystem/tests/test_gaps.py
@@ -82,6 +82,59 @@ class TestAtOffset(unittest.TestCase):
         self.assertEqual(g2, gaps.at_offset(d, 60 << 20))
 
 
+class TestWithin(unittest.TestCase):
+    def test_identity(self):
+        d = make_disk()
+        [gap] = gaps.parts_and_gaps(d)
+        self.assertEqual(gap, gaps.within(d, gap))
+
+    def test_front_used(self):
+        m, d = make_model_and_disk(size=200 << 20)
+        m.storage_version = 2
+        make_partition(m, d, offset=100 << 20, size=1 << 20)
+        [orig_g1, p1, orig_g2] = gaps.parts_and_gaps(d)
+        make_partition(m, d, offset=0, size=20 << 20)
+        [p1, g1, p2, g2] = gaps.parts_and_gaps(d)
+        self.assertEqual(g1, gaps.within(d, orig_g1))
+
+    def test_back_used(self):
+        m, d = make_model_and_disk(size=200 << 20)
+        m.storage_version = 2
+        make_partition(m, d, offset=100 << 20, size=1 << 20)
+        [orig_g1, p1, orig_g2] = gaps.parts_and_gaps(d)
+        make_partition(m, d, offset=80 << 20, size=20 << 20)
+        [g1, p1, p2, g2] = gaps.parts_and_gaps(d)
+        self.assertEqual(g1, gaps.within(d, orig_g1))
+
+    def test_front_and_back_used(self):
+        m, d = make_model_and_disk(size=200 << 20)
+        m.storage_version = 2
+        make_partition(m, d, offset=100 << 20, size=1 << 20)
+        [orig_g1, p1, orig_g2] = gaps.parts_and_gaps(d)
+        make_partition(m, d, offset=0, size=20 << 20)
+        make_partition(m, d, offset=80 << 20, size=20 << 20)
+        [p1, g1, p2, p3, g2] = gaps.parts_and_gaps(d)
+        self.assertEqual(g1, gaps.within(d, orig_g1))
+
+    def test_multi_gap(self):
+        m, d = make_model_and_disk(size=200 << 20)
+        m.storage_version = 2
+        make_partition(m, d, offset=100 << 20, size=1 << 20)
+        [orig_g1, p1, orig_g2] = gaps.parts_and_gaps(d)
+        make_partition(m, d, offset=20 << 20, size=20 << 20)
+        [g1, p1, g2, p2, g3] = gaps.parts_and_gaps(d)
+        self.assertEqual(g1, gaps.within(d, orig_g1))
+
+    def test_later_part_of_disk(self):
+        m, d = make_model_and_disk(size=200 << 20)
+        m.storage_version = 2
+        make_partition(m, d, offset=100 << 20, size=1 << 20)
+        [orig_g1, p1, orig_g2] = gaps.parts_and_gaps(d)
+        make_partition(m, d, offset=120 << 20, size=20 << 20)
+        [g1, p1, g2, p2, g3] = gaps.parts_and_gaps(d)
+        self.assertEqual(g2, gaps.within(d, orig_g2))
+
+
 class TestDiskGaps(unittest.TestCase):
 
     def test_no_partition_gpt(self):
diff --git a/subiquity/common/filesystem/tests/test_manipulator.py b/subiquity/common/filesystem/tests/test_manipulator.py
index 79554ba9..eb29483b 100644
--- a/subiquity/common/filesystem/tests/test_manipulator.py
+++ b/subiquity/common/filesystem/tests/test_manipulator.py
@@ -677,7 +677,6 @@ class TestReformat(unittest.TestCase):
 class TestCanResize(unittest.TestCase):
     def setUp(self):
         self.manipulator = make_manipulator()
-        self.manipulator.model._probe_data = {}
 
     def test_resize_unpreserved(self):
         disk = make_disk(self.manipulator.model, ptable=None)
diff --git a/subiquity/common/filesystem/tests/test_sizes.py b/subiquity/common/filesystem/tests/test_sizes.py
index 048f06cd..bd233b03 100644
--- a/subiquity/common/filesystem/tests/test_sizes.py
+++ b/subiquity/common/filesystem/tests/test_sizes.py
@@ -97,6 +97,7 @@ class TestCalculateGuidedResize(unittest.TestCase):
         actual = calculate_guided_resize(
                 part_min=8 << 30, part_size=100 << 30, install_min=size)
         expected = GuidedResizeValues(
+                install_max=(100 << 30) - size,
                 minimum=size, recommended=50 << 30, maximum=(100 << 30) - size)
         self.assertEqual(expected, actual)
 
@@ -104,6 +105,7 @@ class TestCalculateGuidedResize(unittest.TestCase):
         actual = calculate_guided_resize(
                 part_min=40 << 30, part_size=240 << 30, install_min=10 << 30)
         expected = GuidedResizeValues(
+                install_max=190 << 30,
                 minimum=50 << 30, recommended=200 << 30, maximum=230 << 30)
         self.assertEqual(expected, actual)
 
diff --git a/subiquity/common/types.py b/subiquity/common/types.py
index 556c54f0..5b5c904b 100644
--- a/subiquity/common/types.py
+++ b/subiquity/common/types.py
@@ -333,11 +333,70 @@ class StorageResponseV2:
 
 @attr.s(auto_attribs=True)
 class GuidedResizeValues:
+    install_max: int
     minimum: int
     recommended: int
     maximum: int
 
 
+@attr.s(auto_attribs=True)
+class GuidedStorageTargetReformat:
+    disk_id: str
+
+
+@attr.s(auto_attribs=True)
+class GuidedStorageTargetResize:
+    disk_id: str
+    partition_number: int
+    new_size: int
+    minimum: Optional[int]
+    recommended: Optional[int]
+    maximum: Optional[int]
+
+    @staticmethod
+    def from_recommendations(part, resize_vals):
+        return GuidedStorageTargetResize(
+                disk_id=part.device.id,
+                partition_number=part.number,
+                new_size=resize_vals.recommended,
+                minimum=resize_vals.minimum,
+                recommended=resize_vals.recommended,
+                maximum=resize_vals.maximum,
+                )
+
+
+@attr.s(auto_attribs=True)
+class GuidedStorageTargetUseGap:
+    disk_id: str
+    gap: Gap
+
+
+GuidedStorageTarget = Union[GuidedStorageTargetReformat,
+                            GuidedStorageTargetResize,
+                            GuidedStorageTargetUseGap]
+
+
+@attr.s(auto_attribs=True)
+class GuidedChoiceV2:
+    target: GuidedStorageTarget
+    use_lvm: bool = False
+    password: Optional[str] = attr.ib(default=None, repr=False)
+
+    @staticmethod
+    def from_guided_choice(choice: GuidedChoice):
+        return GuidedChoiceV2(
+                target=GuidedStorageTargetReformat(disk_id=choice.disk_id),
+                use_lvm=choice.use_lvm,
+                password=choice.password,
+                )
+
+
+@attr.s(auto_attribs=True)
+class GuidedStorageResponseV2:
+    configured: Optional[GuidedChoiceV2] = None
+    possible: List[GuidedStorageTarget] = attr.Factory(list)
+
+
 @attr.s(auto_attribs=True)
 class AddPartitionV2:
     disk_id: str
diff --git a/subiquity/models/filesystem.py b/subiquity/models/filesystem.py
index db873186..9becc0c4 100644
--- a/subiquity/models/filesystem.py
+++ b/subiquity/models/filesystem.py
@@ -1039,6 +1039,7 @@ class FilesystemModel(object):
             self._actions = []
         self.swap = None
         self.grub = None
+        self.guided_configuration = None
 
     def load_server_data(self, status):
         log.debug('load_server_data %s', status)
diff --git a/subiquity/models/tests/test_filesystem.py b/subiquity/models/tests/test_filesystem.py
index 519c57ed..631b0b8c 100644
--- a/subiquity/models/tests/test_filesystem.py
+++ b/subiquity/models/tests/test_filesystem.py
@@ -140,6 +140,7 @@ def make_model(bootloader=None):
     model = FilesystemModel()
     if bootloader is not None:
         model.bootloader = bootloader
+    model._probe_data = {}
     return model
 
 
diff --git a/subiquity/server/controllers/filesystem.py b/subiquity/server/controllers/filesystem.py
index 8e38ea2a..eb7f37b3 100644
--- a/subiquity/server/controllers/filesystem.py
+++ b/subiquity/server/controllers/filesystem.py
@@ -50,7 +50,12 @@ from subiquity.common.types import (
     Bootloader,
     Disk,
     GuidedChoice,
+    GuidedChoiceV2,
     GuidedStorageResponse,
+    GuidedStorageResponseV2,
+    GuidedStorageTargetReformat,
+    GuidedStorageTargetResize,
+    GuidedStorageTargetUseGap,
     ModifyPartitionV2,
     ProbeStatus,
     ReformatDisk,
@@ -58,6 +63,7 @@ from subiquity.common.types import (
     StorageResponseV2,
     )
 from subiquity.models.filesystem import (
+    align_up,
     align_down,
     LVM_CHUNK_SIZE,
     Raid,
@@ -132,20 +138,33 @@ class FilesystemController(SubiquityController, FilesystemManipulator):
                 "autoinstall config did not create needed bootloader "
                 "partition")
 
-    def setup_disk_for_guided(self, disk, mode):
+    def setup_disk_for_guided(self, target, mode):
+        if isinstance(target, gaps.Gap):
+            disk = target.device
+            gap = target
+        else:
+            disk = target
+            gap = None
         if mode is None or mode == 'reformat_disk':
             self.reformat(disk, wipe='superblock-recursive')
         if DeviceAction.TOGGLE_BOOT in DeviceAction.supported(disk):
             self.add_boot_disk(disk)
-        return gaps.largest_gap(disk)
+        if gap is None:
+            return disk, gaps.largest_gap(disk)
+        else:
+            # find what's left of the gap after adding boot
+            gap = gaps.within(disk, gap)
+            if gap is None:
+                raise Exception(f'failed to locate gap after adding boot')
+            return disk, gap
 
-    def guided_direct(self, disk, mode=None):
-        gap = self.setup_disk_for_guided(disk, mode)
+    def guided_direct(self, target, mode=None):
+        disk, gap = self.setup_disk_for_guided(target, mode)
         spec = dict(fstype="ext4", mount="/")
         self.create_partition(device=disk, gap=gap, spec=spec)
 
-    def guided_lvm(self, disk, mode=None, lvm_options=None):
-        gap = self.setup_disk_for_guided(disk, mode)
+    def guided_lvm(self, target, mode=None, lvm_options=None):
+        disk, gap = self.setup_disk_for_guided(target, mode)
         gap_boot, gap_rest = gap.split(sizes.get_bootfs_size(gap.size))
         spec = dict(fstype="ext4", mount='/boot')
         self.create_partition(device=disk, gap=gap_boot, spec=spec)
@@ -187,8 +206,31 @@ class FilesystemController(SubiquityController, FilesystemManipulator):
                 mount="/",
                 ))
 
-    def guided(self, choice):
-        disk = self.model._one(id=choice.disk_id)
+    def guided(self, choice: GuidedChoiceV2):
+        self.model.guided_configuration = choice
+
+        disk = self.model._one(id=choice.target.disk_id)
+        if isinstance(choice.target, GuidedStorageTargetReformat):
+            mode = 'reformat_disk'
+            target = disk
+        elif isinstance(choice.target, GuidedStorageTargetUseGap):
+            mode = 'use_gap'
+            target = gaps.at_offset(disk, choice.target.gap.offset)
+        elif isinstance(choice.target, GuidedStorageTargetResize):
+            partition = self.get_partition(
+                    disk, choice.target.partition_number)
+            part_align = disk.alignment_data().part_align
+            new_size = align_up(choice.target.new_size, part_align)
+            if new_size > partition.size:
+                raise Exception(f'Aligned requested size {new_size} too large')
+            gap_offset = partition.offset + new_size
+            partition.size = new_size
+            partition.resize = True
+            mode = 'use_gap'
+            target = gaps.at_offset(disk, gap_offset)
+        else:
+            raise Exception(f'Unknown guided target {choice.target}')
+
         if choice.use_lvm:
             lvm_options = None
             if choice.password is not None:
@@ -198,9 +240,9 @@ class FilesystemController(SubiquityController, FilesystemManipulator):
                         'password': choice.password,
                         },
                     }
-            self.guided_lvm(disk, lvm_options=lvm_options)
+            self.guided_lvm(target, mode=mode, lvm_options=lvm_options)
         else:
-            self.guided_direct(disk)
+            self.guided_direct(target, mode=mode)
 
     async def _probe_response(self, wait, resp_cls):
         if self._probe_task.task is None or not self._probe_task.task.done():
@@ -244,6 +286,29 @@ class FilesystemController(SubiquityController, FilesystemManipulator):
             config, self.model._probe_data['blockdev'], is_probe_data=False)
         await self.configured()
 
+    def get_guided_disks(self, check_boot=True, with_reformatting=False):
+        disks = []
+        for raid in self.model._all(type='raid'):
+            if check_boot and not boot.can_be_boot_device(
+                    raid, with_reformatting=with_reformatting):
+                continue
+            disks.append(raid)
+        for disk in self.model._all(type='disk'):
+            if check_boot and not boot.can_be_boot_device(
+                    disk, with_reformatting=with_reformatting):
+                continue
+            cd = disk.constructed_device()
+            if isinstance(cd, Raid):
+                can_be_boot = False
+                for v in cd._subvolumes:
+                    if check_boot and boot.can_be_boot_device(
+                            v, with_reformatting=with_reformatting):
+                        can_be_boot = True
+                if can_be_boot:
+                    continue
+            disks.append(disk)
+        return disks
+
     async def guided_GET(self, wait: bool = False) -> GuidedStorageResponse:
         probe_resp = await self._probe_response(wait, GuidedStorageResponse)
         if probe_resp is not None:
@@ -253,23 +318,7 @@ class FilesystemController(SubiquityController, FilesystemManipulator):
         # source catalog should directly specify the minimum suitable
         # size?)
         min_size = 2*self.app.base_model.source.current.size + (1 << 30)
-        disks = []
-        for raid in self.model._all(type='raid'):
-            if not boot.can_be_boot_device(raid, with_reformatting=True):
-                continue
-            disks.append(raid)
-        for disk in self.model._all(type='disk'):
-            if not boot.can_be_boot_device(disk, with_reformatting=True):
-                continue
-            cd = disk.constructed_device()
-            if isinstance(cd, Raid):
-                can_be_boot = False
-                for v in cd._subvolumes:
-                    if boot.can_be_boot_device(v, with_reformatting=True):
-                        can_be_boot = True
-                if can_be_boot:
-                    continue
-            disks.append(disk)
+        disks = self.get_guided_disks(with_reformatting=True)
         return GuidedStorageResponse(
             status=ProbeStatus.DONE,
             error_report=self.full_probe_error(),
@@ -277,7 +326,7 @@ class FilesystemController(SubiquityController, FilesystemManipulator):
 
     async def guided_POST(self, data: GuidedChoice) -> StorageResponse:
         log.debug(data)
-        self.guided(data)
+        self.guided(GuidedChoiceV2.from_guided_choice(data))
         return self._done_response()
 
     async def reset_POST(self, context, request) -> StorageResponse:
@@ -315,7 +364,9 @@ class FilesystemController(SubiquityController, FilesystemManipulator):
 
     def calculate_suggested_install_min(self):
         source_min = self.app.base_model.source.current.size
-        return sizes.calculate_suggested_install_min(source_min)
+        align = max((pa.part_align
+                     for pa in self.model._partition_alignment_data.values()))
+        return sizes.calculate_suggested_install_min(source_min, align)
 
     async def v2_GET(self) -> StorageResponseV2:
         disks = self.model._all(type='disk')
@@ -335,10 +386,63 @@ class FilesystemController(SubiquityController, FilesystemManipulator):
         self.model.reset()
         return await self.v2_GET()
 
-    async def v2_guided_POST(self, data: GuidedChoice) -> StorageResponseV2:
+    async def v2_deprecated_guided_POST(self, data: GuidedChoice) \
+            -> StorageResponseV2:
+        log.debug(data)
+        self.guided(GuidedChoiceV2.from_guided_choice(data))
+        return await self.v2_GET()
+
+    async def v2_guided_GET(self) -> GuidedStorageResponseV2:
+        """Acquire a list of possible guided storage configuration scenarios.
+        Results are sorted by the size of the space potentially available to
+        the install."""
+
+        scenarios = []
+        install_min = self.calculate_suggested_install_min()
+
+        for disk in self.get_guided_disks(with_reformatting=True):
+            reformat = GuidedStorageTargetReformat(disk_id=disk.id)
+            scenarios.append((disk.size, reformat))
+
+        for disk in self.get_guided_disks(with_reformatting=False):
+            if len(disk.partitions()) < 1:
+                # On an empty disk, don't bother to offer it with UseGap, as
+                # it's basically the same as the Reformat case.
+                continue
+            gap = gaps.largest_gap(disk)
+            if gap is not None and gap.size >= install_min:
+                api_gap = labels.for_client(gap)
+                use_gap = GuidedStorageTargetUseGap(
+                        disk_id=disk.id,
+                        gap=api_gap)
+                scenarios.append((gap.size, use_gap))
+
+        for disk in self.get_guided_disks(check_boot=False):
+            part_align = disk.alignment_data().part_align
+            for partition in disk.partitions():
+                vals = sizes.calculate_guided_resize(
+                        partition.estimated_min_size, partition.size,
+                        install_min, part_align=part_align)
+                if vals is None:
+                    continue
+                if not boot.can_be_boot_device(
+                        disk, resize_partition=partition,
+                        with_reformatting=False):
+                    continue
+                resize = GuidedStorageTargetResize.from_recommendations(
+                        partition, vals)
+                scenarios.append((vals.install_max, resize))
+
+        scenarios.sort(reverse=True, key=lambda x: x[0])
+        return GuidedStorageResponseV2(
+                configured=self.model.guided_configuration,
+                possible=[s[1] for s in scenarios])
+
+    async def v2_guided_POST(self, data: GuidedChoiceV2) \
+            -> GuidedStorageResponseV2:
         log.debug(data)
         self.guided(data)
-        return await self.v2_GET()
+        return await self.v2_guided_GET()
 
     async def v2_reformat_disk_POST(self, data: ReformatDisk) \
             -> StorageResponseV2:
@@ -472,8 +576,8 @@ class FilesystemController(SubiquityController, FilesystemManipulator):
 
         if mode == 'reformat_disk':
             match = layout.get("match", {'size': 'largest'})
-            disk = self.model.disk_for_match(self.model.all_disks(), match)
-            if not disk:
+            target = self.model.disk_for_match(self.model.all_disks(), match)
+            if not target:
                 raise Exception("autoinstall cannot configure storage "
                                 "- no disk found large enough for install")
         elif mode == 'use_gap':
@@ -485,10 +589,10 @@ class FilesystemController(SubiquityController, FilesystemManipulator):
                                 "- no gap found large enough for install")
             # This is not necessarily the exact gap to be used, as the gap size
             # may change once add_boot_disk has sorted things out.
-            disk = gap.device
+            target = gap
         log.info(f'autoinstall: running guided {name} install in mode {mode} '
-                 f'using {disk}')
-        guided_method(disk=disk, mode=mode)
+                 f'using {target}')
+        guided_method(target=target, mode=mode)
 
     def validate_layout_mode(self, mode):
         if mode not in ('reformat_disk', 'use_gap'):
diff --git a/subiquity/server/controllers/tests/test_filesystem.py b/subiquity/server/controllers/tests/test_filesystem.py
index a117975b..d5e7e265 100644
--- a/subiquity/server/controllers/tests/test_filesystem.py
+++ b/subiquity/server/controllers/tests/test_filesystem.py
@@ -13,6 +13,7 @@
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
+import copy
 from unittest import mock, TestCase, IsolatedAsyncioTestCase
 
 from parameterized import parameterized
@@ -20,10 +21,18 @@ from parameterized import parameterized
 from subiquity.server.controllers.filesystem import FilesystemController
 
 from subiquitycore.tests.mocks import make_app
-from subiquity.common.types import Bootloader
+from subiquity.common.filesystem import gaps
+from subiquity.common.types import (
+    Bootloader,
+    GuidedChoiceV2,
+    GuidedStorageTargetReformat,
+    GuidedStorageTargetResize,
+    GuidedStorageTargetUseGap,
+    )
 from subiquity.models.tests.test_filesystem import (
     make_disk,
     make_model,
+    make_partition,
     )
 
 
@@ -127,3 +136,194 @@ class TestLayout(TestCase):
     def test_bad_modes(self, mode):
         with self.assertRaises(ValueError):
             self.fsc.validate_layout_mode(mode)
+
+
+bootloaders_and_ptables = [(bl, pt)
+                           for bl in list(Bootloader)
+                           for pt in ('gpt', 'msdos', 'vtoc')]
+
+
+class TestGuidedV2(IsolatedAsyncioTestCase):
+    def _setup(self, bootloader, ptable, fix_bios=True, **kw):
+        self.app = make_app()
+        self.app.opts.bootloader = bootloader.value
+        self.fsc = FilesystemController(app=self.app)
+        self.fsc.calculate_suggested_install_min = mock.Mock()
+        self.fsc.calculate_suggested_install_min.return_value = 10 << 30
+        self.fsc.model = self.model = make_model(bootloader)
+        self.disk = make_disk(self.model, ptable=ptable, **kw)
+        self.model.storage_version = 2
+        self.fs_probe = {}
+        self.fsc.model._probe_data = {
+            'blockdev': {},
+            'filesystem': self.fs_probe,
+            }
+        if bootloader == Bootloader.BIOS and ptable != 'msdos' and fix_bios:
+            make_partition(self.model, self.disk, preserve=True,
+                           flag='bios_grub', size=1 << 20, offset=1 << 20)
+
+    @parameterized.expand(bootloaders_and_ptables)
+    async def test_blank_disk(self, bootloader, ptable):
+        # blank disks should not report a UseGap case
+        self._setup(bootloader, ptable, fix_bios=False)
+        expected = [GuidedStorageTargetReformat(disk_id=self.disk.id)]
+        resp = await self.fsc.v2_guided_GET()
+        self.assertEqual(expected, resp.possible)
+
+    @parameterized.expand(bootloaders_and_ptables)
+    async def test_used_half_disk(self, bootloader, ptable):
+        self._setup(bootloader, ptable, size=100 << 30)
+        p = make_partition(self.model, self.disk, preserve=True, size=50 << 30)
+        gap_offset = p.size + p.offset
+        self.fs_probe[p._path()] = {'ESTIMATED_MIN_SIZE': 1 << 20}
+        resp = await self.fsc.v2_guided_GET()
+
+        reformat = resp.possible.pop(0)
+        self.assertEqual(GuidedStorageTargetReformat(disk_id=self.disk.id),
+                         reformat)
+
+        use_gap = resp.possible.pop(0)
+        self.assertEqual(self.disk.id, use_gap.disk_id)
+        if gap_offset != use_gap.gap.offset:
+            breakpoint()
+        self.assertEqual(gap_offset, use_gap.gap.offset)
+
+        resize = resp.possible.pop(0)
+        self.assertEqual(self.disk.id, resize.disk_id)
+        self.assertEqual(p.number, resize.partition_number)
+        self.assertTrue(isinstance(resize, GuidedStorageTargetResize))
+        self.assertEqual(0, len(resp.possible))
+
+    @parameterized.expand(bootloaders_and_ptables)
+    async def test_used_full_disk(self, bootloader, ptable):
+        self._setup(bootloader, ptable)
+        p = make_partition(self.model, self.disk, preserve=True,
+                           size=gaps.largest_gap_size(self.disk))
+        self.fs_probe[p._path()] = {'ESTIMATED_MIN_SIZE': 1 << 20}
+        resp = await self.fsc.v2_guided_GET()
+        reformat = resp.possible.pop(0)
+        self.assertEqual(GuidedStorageTargetReformat(disk_id=self.disk.id),
+                         reformat)
+
+        resize = resp.possible.pop(0)
+        self.assertEqual(self.disk.id, resize.disk_id)
+        self.assertEqual(p.number, resize.partition_number)
+        self.assertTrue(isinstance(resize, GuidedStorageTargetResize))
+        self.assertEqual(0, len(resp.possible))
+
+    @parameterized.expand(bootloaders_and_ptables)
+    async def test_weighted_split(self, bootloader, ptable):
+        self._setup(bootloader, ptable, size=250 << 30)
+        # add an extra, filler, partition so that there is no use_gap result
+        make_partition(self.model, self.disk, preserve=True, size=9 << 30)
+        p = make_partition(self.model, self.disk, preserve=True,
+                           size=240 << 30)
+        self.fs_probe[p._path()] = {'ESTIMATED_MIN_SIZE': 40 << 30}
+        self.fsc.calculate_suggested_install_min.return_value = 10 << 30
+        resp = await self.fsc.v2_guided_GET()
+        reformat = resp.possible.pop(0)
+        self.assertEqual(GuidedStorageTargetReformat(disk_id=self.disk.id),
+                         reformat)
+
+        resize = resp.possible.pop(0)
+        expected = GuidedStorageTargetResize(
+                disk_id=self.disk.id,
+                partition_number=p.number,
+                new_size=200 << 30,
+                minimum=50 << 30,
+                recommended=200 << 30,
+                maximum=230 << 30)
+        self.assertEqual(expected, resize)
+        self.assertEqual(0, len(resp.possible))
+
+    @parameterized.expand(bootloaders_and_ptables)
+    async def test_half_disk_reformat(self, bootloader, ptable):
+        self._setup(bootloader, ptable, size=100 << 30)
+        p = make_partition(self.model, self.disk, preserve=True, size=50 << 30)
+        self.fs_probe[p._path()] = {'ESTIMATED_MIN_SIZE': 1 << 20}
+
+        guided_get_resp = await self.fsc.v2_guided_GET()
+        reformat = guided_get_resp.possible.pop(0)
+        self.assertTrue(isinstance(reformat, GuidedStorageTargetReformat))
+
+        use_gap = guided_get_resp.possible.pop(0)
+        self.assertTrue(isinstance(use_gap, GuidedStorageTargetUseGap))
+
+        resize = guided_get_resp.possible.pop(0)
+        self.assertTrue(isinstance(resize, GuidedStorageTargetResize))
+
+        data = GuidedChoiceV2(target=reformat)
+        expected_config = copy.copy(data)
+        resp = await self.fsc.v2_guided_POST(data=data)
+        self.assertEqual(expected_config, resp.configured)
+
+        resp = await self.fsc.v2_GET()
+        self.assertFalse(resp.need_root)
+        self.assertFalse(resp.need_boot)
+        self.assertEqual(0, len(guided_get_resp.possible))
+
+    @parameterized.expand(bootloaders_and_ptables)
+    async def test_half_disk_use_gap(self, bootloader, ptable):
+        self._setup(bootloader, ptable, size=100 << 30)
+        p = make_partition(self.model, self.disk, preserve=True, size=50 << 30)
+        self.fs_probe[p._path()] = {'ESTIMATED_MIN_SIZE': 1 << 20}
+
+        resp = await self.fsc.v2_GET()
+        [orig_p, g] = resp.disks[0].partitions[-2:]
+
+        guided_get_resp = await self.fsc.v2_guided_GET()
+        reformat = guided_get_resp.possible.pop(0)
+        self.assertTrue(isinstance(reformat, GuidedStorageTargetReformat))
+
+        use_gap = guided_get_resp.possible.pop(0)
+        self.assertTrue(isinstance(use_gap, GuidedStorageTargetUseGap))
+        self.assertEqual(g, use_gap.gap)
+        data = GuidedChoiceV2(target=use_gap)
+        expected_config = copy.copy(data)
+        resp = await self.fsc.v2_guided_POST(data=data)
+        self.assertEqual(expected_config, resp.configured)
+
+        resize = guided_get_resp.possible.pop(0)
+        self.assertTrue(isinstance(resize, GuidedStorageTargetResize))
+
+        resp = await self.fsc.v2_GET()
+        existing_part = [p for p in resp.disks[0].partitions
+                         if getattr(p, 'number', None) == orig_p.number][0]
+        self.assertEqual(orig_p, existing_part)
+        self.assertFalse(resp.need_root)
+        self.assertFalse(resp.need_boot)
+        self.assertEqual(0, len(guided_get_resp.possible))
+
+    @parameterized.expand(bootloaders_and_ptables)
+    async def test_half_disk_resize(self, bootloader, ptable):
+        self._setup(bootloader, ptable, size=100 << 30)
+        p = make_partition(self.model, self.disk, preserve=True, size=50 << 30)
+        self.fs_probe[p._path()] = {'ESTIMATED_MIN_SIZE': 1 << 20}
+
+        resp = await self.fsc.v2_GET()
+        [orig_p, g] = resp.disks[0].partitions[-2:]
+
+        guided_get_resp = await self.fsc.v2_guided_GET()
+        reformat = guided_get_resp.possible.pop(0)
+        self.assertTrue(isinstance(reformat, GuidedStorageTargetReformat))
+
+        use_gap = guided_get_resp.possible.pop(0)
+        self.assertTrue(isinstance(use_gap, GuidedStorageTargetUseGap))
+
+        resize = guided_get_resp.possible.pop(0)
+        self.assertTrue(isinstance(resize, GuidedStorageTargetResize))
+        p_expected = copy.copy(orig_p)
+        p_expected.size = resize.new_size = 20 << 30
+        p_expected.resize = True
+        data = GuidedChoiceV2(target=resize)
+        expected_config = copy.copy(data)
+        resp = await self.fsc.v2_guided_POST(data=data)
+        self.assertEqual(expected_config, resp.configured)
+
+        resp = await self.fsc.v2_GET()
+        existing_part = [p for p in resp.disks[0].partitions
+                         if getattr(p, 'number', None) == orig_p.number][0]
+        self.assertEqual(p_expected, existing_part)
+        self.assertFalse(resp.need_root)
+        self.assertFalse(resp.need_boot)
+        self.assertEqual(0, len(guided_get_resp.possible))
diff --git a/subiquity/tests/api/test_api.py b/subiquity/tests/api/test_api.py
index b495aa3d..efe96ffb 100755
--- a/subiquity/tests/api/test_api.py
+++ b/subiquity/tests/api/test_api.py
@@ -1,3 +1,18 @@
+# Copyright 2021 Canonical, Ltd.
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as
+# published by the Free Software Foundation, either version 3 of the
+# License, or (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
 import aiohttp
 from aiohttp.client_exceptions import ClientResponseError
 import async_timeout
@@ -26,9 +41,12 @@ def first(items, key, value):
     return next(find(items, key, value))
 
 
-def match(items, **kwargs):
+def match(items, **kw):
+    typename = kw.pop('_type', None)
+    if typename is not None:
+        kw['$type'] = typename
     return [item for item in items
-            if all(item.get(k) == v for k, v in kwargs.items())]
+            if all(item.get(k) == v for k, v in kw.items())]
 
 
 def timeout(multiplier=1):
@@ -247,7 +265,7 @@ class TestFlow(TestAPI):
             resp = await inst.get('/storage/guided')
             disk_id = resp['disks'][0]['id']
             choice = {"disk_id": disk_id}
-            await inst.post('/storage/v2/guided', choice)
+            await inst.post('/storage/v2/deprecated/guided', choice)
             await inst.post('/storage/v2')
             await inst.get('/meta/status', cur='WAITING')
             await inst.post('/meta/confirm', tty='/dev/tty1')
@@ -331,7 +349,8 @@ class TestFlow(TestAPI):
             self.assertEqual(orig_resp, reset_resp)
 
             choice = {'disk_id': disk_id}
-            guided_resp = await inst.post('/storage/v2/guided', choice)
+            guided_resp = await inst.post('/storage/v2/deprecated/guided',
+                                          choice)
             post_resp = await inst.post('/storage/v2')
             # posting to the endpoint shouldn't change the answer
             self.assertEqual(guided_resp, post_resp)
@@ -339,13 +358,127 @@ class TestFlow(TestAPI):
 
 class TestGuided(TestAPI):
     @timeout()
-    async def test_guided_v2(self):
+    async def test_deprecated_guided_v2(self):
         async with start_server('examples/simple.json') as inst:
             choice = {'disk_id': 'disk-sda'}
-            resp = await inst.post('/storage/v2/guided', choice)
+            resp = await inst.post('/storage/v2/deprecated/guided', choice)
             self.assertEqual(1, len(resp['disks']))
             self.assertEqual('disk-sda', resp['disks'][0]['id'])
 
+    @timeout()
+    async def test_guided_v2_reformat(self):
+        cfg = 'examples/win10-along-ubuntu.json'
+        extra = ['--storage-version', '2']
+        async with start_server(cfg, extra_args=extra) as inst:
+            resp = await inst.get('/storage/v2/guided')
+            [reformat] = match(resp['possible'],
+                               _type='GuidedStorageTargetReformat')
+            data = {'target': reformat}
+            resp = await inst.post('/storage/v2/guided', data)
+            self.assertEqual(reformat, resp['configured']['target'])
+            resp = await inst.get('/storage/v2')
+            [p1, p2] = resp['disks'][0]['partitions']
+            expected_p1 = {
+                '$type': 'Partition',
+                'boot': True,
+                'format': 'fat32',
+                'grub_device': True,
+                'mount': '/boot/efi',
+                'number': 1,
+                'wipe': 'superblock',
+            }
+            self.assertDictSubset(expected_p1, p1)
+            expected_p2 = {
+                'number': 2,
+                'mount': '/',
+                'format': 'ext4',
+                'wipe': 'superblock',
+            }
+            self.assertDictSubset(expected_p2, p2)
+
+    @timeout()
+    async def test_guided_v2_resize(self):
+        cfg = 'examples/win10-along-ubuntu.json'
+        extra = ['--storage-version', '2']
+        async with start_server(cfg, extra_args=extra) as inst:
+            orig_resp = await inst.get('/storage/v2')
+            [orig_p1, orig_p2, orig_p3, orig_p4, orig_p5] = \
+                orig_resp['disks'][0]['partitions']
+            resp = await inst.get('/storage/v2/guided')
+            [resize_ntfs, resize_ext4] = match(
+                    resp['possible'], _type='GuidedStorageTargetResize')
+            resize_ntfs['new_size'] = 30 << 30
+            data = {'target': resize_ntfs}
+            resp = await inst.post('/storage/v2/guided', data)
+            self.assertEqual(resize_ntfs, resp['configured']['target'])
+            resp = await inst.get('/storage/v2')
+            [p1, p2, p3, p6, p4, p5] = resp['disks'][0]['partitions']
+            expected_p1 = {
+                '$type': 'Partition',
+                'boot': True,
+                'format': 'vfat',
+                'grub_device': True,
+                'mount': '/boot/efi',
+                'number': 1,
+                'size': orig_p1['size'],
+                'resize': None,
+                'wipe': None,
+            }
+            self.assertDictSubset(expected_p1, p1)
+            self.assertEqual(orig_p2, p2)
+            self.assertEqual(orig_p4, p4)
+            self.assertEqual(orig_p5, p5)
+            expected_p6 = {
+                'number': 6,
+                'mount': '/',
+                'format': 'ext4',
+            }
+            self.assertDictSubset(expected_p6, p6)
+
+    @timeout()
+    async def test_guided_v2_use_gap(self):
+        cfg = self.machineConfig('examples/win10-along-ubuntu.json')
+        with cfg.edit() as data:
+            pt = data['storage']['blockdev']['/dev/sda']['partitiontable']
+            [node] = match(pt['partitions'], node='/dev/sda5')
+            pt['partitions'].remove(node)
+            del data['storage']['blockdev']['/dev/sda5']
+            del data['storage']['filesystem']['/dev/sda5']
+        extra = ['--storage-version', '2']
+        async with start_server(cfg, extra_args=extra) as inst:
+            orig_resp = await inst.get('/storage/v2')
+            [orig_p1, orig_p2, orig_p3, orig_p4, gap] = \
+                orig_resp['disks'][0]['partitions']
+            resp = await inst.get('/storage/v2/guided')
+            [use_gap] = match(resp['possible'],
+                              _type='GuidedStorageTargetUseGap')
+            data = {'target': use_gap}
+            resp = await inst.post('/storage/v2/guided', data)
+            self.assertEqual(use_gap, resp['configured']['target'])
+            resp = await inst.get('/storage/v2')
+            [p1, p2, p3, p4, p5] = resp['disks'][0]['partitions']
+            expected_p1 = {
+                '$type': 'Partition',
+                'boot': True,
+                'format': 'vfat',
+                'grub_device': True,
+                'mount': '/boot/efi',
+                'number': 1,
+                'size': orig_p1['size'],
+                'resize': None,
+                'wipe': None,
+            }
+            self.assertDictSubset(expected_p1, p1)
+            self.assertEqual(orig_p2, p2)
+            self.assertEqual(orig_p3, p3)
+            self.assertEqual(orig_p4, p4)
+            expected_p5 = {
+                'number': 5,
+                'mount': '/',
+                'format': 'ext4',
+            }
+            self.assertDictSubset(expected_p5, p5)
+
 
 class TestAdd(TestAPI):
     @timeout()
@@ -772,7 +905,7 @@ class TestTodos(TestAPI):  # server indicators of required client actions
             self.assertTrue(resp['need_boot'])
 
             choice = {'disk_id': disk_id}
-            resp = await inst.post('/storage/v2/guided', choice)
+            resp = await inst.post('/storage/v2/deprecated/guided', choice)
             self.assertFalse(resp['need_root'])
             self.assertFalse(resp['need_boot'])