diff --git a/src/aks-preview/HISTORY.rst b/src/aks-preview/HISTORY.rst
index fc91075c6a6..c43392f637c 100644
--- a/src/aks-preview/HISTORY.rst
+++ b/src/aks-preview/HISTORY.rst
@@ -11,6 +11,7 @@ To release a new version, please select a new version number (usually plus 1 to
 
 Pending
 +++++++
+* `az aks enable-addons`, `az aks disable-addons` and `az aks update`: Keep Container Insights monitoring and container network log settings consistent with the legacy monitoring addon when updating clusters.
 * `az aks machine add`: Add preview `--capacity-reservation-group` support to associate a machine with a Capacity Reservation Group.
 * Add `az aks alert-config` commands to manage AKS-managed alert configurations.
 * `az aks create`: Honor `--enable-osdisk-full-caching` for the default agent pool.
diff --git a/src/aks-preview/azext_aks_preview/custom.py b/src/aks-preview/azext_aks_preview/custom.py
index e14eace7eb1..0566a831259 100644
--- a/src/aks-preview/azext_aks_preview/custom.py
+++ b/src/aks-preview/azext_aks_preview/custom.py
@@ -4085,6 +4085,19 @@ def _update_addons(cmd,  # pylint: disable=too-many-branches,too-many-statements
                     raise CLIError(f"The addon {addon} is not installed.")
             addon_profiles[addon].config = None
         addon_profiles[addon].enabled = enable
+        if addon == CONST_MONITORING_ADDON_NAME:
+            monitor_profile = getattr(instance, "azure_monitor_profile", None)
+            if getattr(monitor_profile, "container_insights", None) is not None:
+                # Reset canonical monitoring values along with the legacy addon config.
+                ContainerInsights = cmd.get_models(
+                    "ManagedClusterAzureMonitorProfileContainerInsights",
+                    resource_type=CUSTOM_MGMT_AKS_PREVIEW,
+                    operation_group="managed_clusters",
+                )
+                monitor_profile.container_insights = ContainerInsights(
+                    enabled=enable,
+                    log_analytics_workspace_resource_id=workspace_resource_id if enable else None,
+                )
 
     instance.addon_profiles = addon_profiles
 
diff --git a/src/aks-preview/azext_aks_preview/managed_cluster_decorator.py b/src/aks-preview/azext_aks_preview/managed_cluster_decorator.py
index a9a88b3c8e8..7fef93edcb7 100644
--- a/src/aks-preview/azext_aks_preview/managed_cluster_decorator.py
+++ b/src/aks-preview/azext_aks_preview/managed_cluster_decorator.py
@@ -6662,6 +6662,11 @@ def update_monitoring_profile_flow_logs(self, mc: ManagedCluster) -> ManagedClus
                     config = monitoring_addon_profile.config or {}
                     config["enableRetinaNetworkFlags"] = str(container_network_logs_enabled)
                     mc.addon_profiles[monitoring_addon_key].config = config
+            container_insights = getattr(mc.azure_monitor_profile, "container_insights", None)
+            if container_insights is not None:
+                container_insights.container_network_logs = (
+                    "Enabled" if container_network_logs_enabled else "Disabled"
+                )
 
         # When enabling CNL, the DCR must be updated to add the high-scale stream.
         # Set the postprocessing intermediate so that the update path calls ensure_container_insights.
diff --git a/src/aks-preview/azext_aks_preview/tests/latest/data/setup_proxy.sh b/src/aks-preview/azext_aks_preview/tests/latest/data/setup_proxy.sh
index 879a0988688..9d0765970ae 100644
--- a/src/aks-preview/azext_aks_preview/tests/latest/data/setup_proxy.sh
+++ b/src/aks-preview/azext_aks_preview/tests/latest/data/setup_proxy.sh
@@ -1,5 +1,5 @@
 #!/usr/bin/env bash
-set -x
+set -euxo pipefail
 
 echo "setting up"
 WORKDIR="${1:-$(mktemp -d)}"
@@ -7,22 +7,14 @@ echo "setting up ${WORKDIR}"
 
 pushd "$WORKDIR"
 
-apt update -y && apt install -y apt-transport-https curl gnupg make gcc < /dev/null
-
-# add diladele apt key
-wget -qO - https://packages.diladele.com/diladele_pub.asc | apt-key add -
-
-# add new repo
-tee /etc/apt/sources.list.d/squid413-ubuntu20.diladele.com.list <<EOF
-deb https://squid413-ubuntu20.diladele.com/ubuntu/ focal main
-EOF
-
-# and install
-apt-get update && apt-get install -y squid-common squid-openssl squidclient libecap3 libecap3-dev < /dev/null
+apt-get -o Acquire::Retries=3 update
+DEBIAN_FRONTEND=noninteractive apt-get install -y ca-certificates curl squid-openssl < /dev/null
 
 mkdir -p /var/lib/squid
 
-/usr/lib/squid/security_file_certgen -c -s /var/lib/squid/ssl_db -M 4MB || true
+if [ ! -d /var/lib/squid/ssl_db ]; then
+    /usr/lib/squid/security_file_certgen -c -s /var/lib/squid/ssl_db -M 4MB
+fi
 
 chown -R proxy:proxy /var/lib/squid
 
@@ -129,11 +121,14 @@ update-ca-certificates
 sed -i 's~http_access deny all~http_access allow all~' /etc/squid/squid.conf
 sed -i "s~http_port 3128~http_port $HOST:3128\nhttps_port $HOST:3129 tls-cert=/etc/squid/squidc.pem tls-key=/etc/squid/squidk.pem~" /etc/squid/squid.conf
 
+squid -k parse
 systemctl restart squid
-systemctl status squid
-
-# validation, fails VM creation if commands fail
-curl -fsSl -o /dev/null -w '%{http_code}\n' -x http://${HOST}:3128/ -I http://www.google.com
-curl -fsSl -o /dev/null -w '%{http_code}\n' -x http://${HOST}:3128/ -I https://www.google.com
-curl -fsSl -o /dev/null -w '%{http_code}\n' -x https://${HOST}:3129/ -I http://www.google.com
-curl -fsSl -o /dev/null -w '%{http_code}\n' -x https://${HOST}:3129/ -I https://www.google.com
+systemctl is-active --quiet squid
+
+# Fail cloud-init unless both proxy listeners can reach the required endpoints.
+for proxy in "http://${HOST}:3128/" "https://${HOST}:3129/"; do
+    for endpoint in "http://packages.microsoft.com/" "https://mcr.microsoft.com/v2/"; do
+        curl -fsSL --connect-timeout 10 --max-time 30 --retry 2 --retry-connrefused \
+            -o /dev/null -w '%{http_code}\n' -x "$proxy" "$endpoint"
+    done
+done
diff --git a/src/aks-preview/azext_aks_preview/tests/latest/test_aks_commands.py b/src/aks-preview/azext_aks_preview/tests/latest/test_aks_commands.py
index 1a0c323a217..f0c7e8a2860 100644
--- a/src/aks-preview/azext_aks_preview/tests/latest/test_aks_commands.py
+++ b/src/aks-preview/azext_aks_preview/tests/latest/test_aks_commands.py
@@ -119,6 +119,7 @@ def _is_transient_operation_conflict(ex):
         return (
             "Another operation is in progress" in message or
             "Operation is not allowed because there's an in-progress" in message or
+            "AKSOperationPreempted" in message or
             "in-progress PutExtensionAddonHandler.PUT operation" in message or
             "is in Updating state, please wait for it to succeed" in message or
             "ProvisioningState of extension: Updating" in message or
@@ -399,6 +400,19 @@ def _cmd_with_retry(self, command, checks, expect_failure):
 
         return result
 
+    def _reimage_nodepool(self, resource_group, cluster_name, nodepool_name):
+        self.cmd(
+            f"aks nodepool upgrade --resource-group {resource_group} --cluster-name {cluster_name} "
+            f"--name {nodepool_name} --node-image-only --yes",
+            checks=[self.is_empty()],
+        )
+        # The upgradeNodeImageVersion action returns no resource body.
+        self.cmd(
+            f"aks nodepool show --resource-group {resource_group} --cluster-name {cluster_name} "
+            f"--name {nodepool_name}",
+            checks=[self.check("provisioningState", "Succeeded")],
+        )
+
     def _create_log_analytics_workspace(self, resource_group_location):
         workspace_name = self.create_random_name("clilaw", 16)
         workspace_location = (
@@ -494,8 +508,9 @@ def _cmd_or_skip_if_managed_system_unavailable(self, cmd, checks=None):
             return self.cmd(cmd, checks=checks)
         except Exception as ex:  # pylint: disable=broad-except
             message = str(ex)
-            if "ManagedSystem" in message and re.search(
-                r"not (?:whitelisted|allowed|enabled|supported|available|registered)",
+            if "managedsystem" in message.lower() and re.search(
+                r"not (?:whitelisted|allowed|enabled|supported|available|registered)"
+                r"|only supports? whitelisted subscriptions",
                 message,
                 re.IGNORECASE,
             ):
@@ -560,12 +575,12 @@ def _cmd_or_skip_if_os_sku_retired(self, cmd, os_sku, checks=None):
                 )
             raise
 
-    def _cmd_or_skip_if_feature_unavailable(self, cmd, feature_name, checks=None):
-        """Run a command that depends on a preview feature unlocked via `--aks-custom-headers`.
+    def _cmd_or_skip_if_feature_unavailable(self, cmd, feature_name, checks=None, feature_aliases=()):
+        """Run a command that requires a preview feature in the test environment.
 
         Some `AKSHTTPCustomFeatures` preview flags additionally require the test
         subscription to be on a service-side allowlist; the header alone is not
-        sufficient to unlock them (mirrors the verified behavior seen for the
+        sufficient to unlock them or make them available in every region (mirrors the behavior seen for the
         `ManagedSystem` agent pool mode preview). If the service rejects the request
         because this subscription isn't enrolled, skip with a precise reason instead
         of failing the test; any other failure (e.g. a real CLI/service regression)
@@ -575,17 +590,35 @@ def _cmd_or_skip_if_feature_unavailable(self, cmd, feature_name, checks=None):
             return self.cmd(cmd, checks=checks)
         except Exception as ex:  # pylint: disable=broad-except
             message = str(ex)
-            if feature_name.lower() in message.lower() and re.search(
-                r"not (?:whitelisted|allowed|enabled|supported|available|registered)",
+            feature_matches = any(
+                name.lower() in message.lower() for name in (feature_name, *feature_aliases)
+            )
+            if feature_matches and re.search(
+                r"not (?:yet )?(?:whitelisted|allowed|enabled|supported|available|registered)",
                 message,
                 re.IGNORECASE,
             ):
                 self.skipTest(
-                    f"This subscription is not enrolled for the {feature_name} preview "
-                    f"despite the aks-custom-headers override: {message}"
+                    f"The {feature_name} preview is unavailable in this test environment: {message}"
                 )
             raise
 
+    def _cmd_or_skip_if_region_unavailable(self, command, location, checks=None, vm_size=None):
+        try:
+            return self.cmd(command, checks=checks)
+        except (CLIError, HttpResponseError) as ex:
+            message = str(ex).lower()
+            expected_location = location.replace(" ", "").lower()
+            if f"'{expected_location}'" not in message:
+                raise
+            if vm_size is None:
+                unavailable = "(availabilityzonenotsupported)" in message
+            else:
+                unavailable = "(vmsizenotsupported)" in message and f"'{vm_size.lower()}'" in message
+            if unavailable:
+                self.skipTest(f"Required test capability is unavailable in {location}: {ex}")
+            raise
+
     def _get_lts_version(self, location):
         """Return the latest LTS version in the given location."""
         data = self.cmd(
@@ -928,8 +961,9 @@ def test_aks_create_and_update_with_managed_nat_gateway_standardv2(
             "--nat-gateway-managed-outbound-ip-count 2 "
             "--ssh-key-value={ssh_key_value}"
         )
-        self.cmd(
+        self._cmd_or_skip_if_feature_unavailable(
             create_cmd,
+            "managedNATGatewayV2",
             checks=[
                 self.check("provisioningState", "Succeeded"),
                 self.check("networkProfile.outboundType", "managedNATGateway"),
@@ -9860,6 +9894,21 @@ def test_aks_create_update_cluster_node_hardening(
             checks=[self.is_empty()],
         )
 
+    def _wait_for_http_proxy(self):
+        script = (
+            "timeout 600 cloud-init status --wait && systemctl is-active --quiet squid && "
+            "curl -fsSL --connect-timeout 10 --max-time 30 --retry 2 --retry-connrefused "
+            "-x http://cli-proxy-vm:3128/ https://mcr.microsoft.com/v2/ -o /dev/null && "
+            "curl -fsSL --connect-timeout 10 --max-time 30 --retry 2 --retry-connrefused "
+            "-x https://cli-proxy-vm:3129/ https://mcr.microsoft.com/v2/ -o /dev/null && "
+            "echo AKS_PROXY_READY"
+        )
+        self.cmd(
+            "vm run-command invoke --resource-group={resource_group} --name cli-proxy-vm "
+            f'--command-id RunShellScript --scripts "{script}" -o json',
+            checks=[self.check("contains(join('', value[].message), `\"\\nAKS_PROXY_READY\\n\"`)", True)],
+        )
+
     def _setup_http_proxy_cluster(self, resource_group, _resource_group_location, aks_name):
         """Shared setup for the http-proxy tests below: create a VNet/subnet,
         a proxy VM (with cloud-init that stands up an actual HTTP(S) proxy),
@@ -9873,8 +9922,8 @@ def _setup_http_proxy_cluster(self, resource_group, _resource_group_location, ak
         sequentially in a single live test run. Factoring the common
         VNet/VM/initial-create steps into this helper lets each of those
         follow-on scenarios (update / disable / re-enable) run as its own
-        independent live test, each well under the 1-hour live-test budget,
-        without dropping any of the original coverage.
+        independent live test. Verify the proxy before starting the cluster
+        so broken bootstrap/egress does not consume the full test timeout.
         """
         self.kwargs.update(
             {
@@ -9910,7 +9959,7 @@ def _setup_http_proxy_cluster(self, resource_group, _resource_group_location, ak
         create_vm_cmd = 'vm create \
             --resource-group={resource_group} \
             --name=cli-proxy-vm \
-            --image Canonical:0001-com-ubuntu-server-focal:20_04-lts:latest \
+            --image Canonical:ubuntu-24_04-lts:server:latest \
             --ssh-key-values @{ssh_key_value} \
             --public-ip-address "" \
             --custom-data {custom_data_path} \
@@ -9931,6 +9980,7 @@ def _setup_http_proxy_cluster(self, resource_group, _resource_group_location, ak
         assert subnet_id is not None
 
         self.cmd(create_vm_cmd)
+        self._wait_for_http_proxy()
 
         self.kwargs.update(
             {
@@ -9966,8 +10016,7 @@ def _setup_http_proxy_cluster(self, resource_group, _resource_group_location, ak
     # scenario is split below into three focused live tests that share the
     # ``_setup_http_proxy_cluster`` helper above. This preserves the exact
     # same assertions/coverage (initial create, config update, disable,
-    # re-enable) while keeping each individual test comfortably within the
-    # live-test time budget and letting them run in parallel (separate
+    # re-enable) while letting them run in parallel (separate
     # per-test resource groups).
     #
     # this case relatively frequently requires updating the corresponding recording file after network/virtualnetwork
@@ -10923,7 +10972,9 @@ def test_aks_enable_utlra_ssd(self, resource_group, resource_group_location):
             "--node-vm-size Standard_D2s_v3 --zones 1 2 3 --enable-ultra-ssd "
             "--ssh-key-value={ssh_key_value}"
         )
-        self.cmd(create_cmd, checks=[self.check("provisioningState", "Succeeded")])
+        self._cmd_or_skip_if_region_unavailable(
+            create_cmd, resource_group_location, checks=[self.check("provisioningState", "Succeeded")]
+        )
 
         # delete
         self.cmd(
@@ -15960,8 +16011,9 @@ def test_aks_availability_zones(self, resource_group, resource_group_location):
             "aks create --resource-group={resource_group} --name={name} "
             "--node-count=1 --ssh-key-value={ssh_key_value} --zones {zones}"
         )
-        self.cmd(
+        self._cmd_or_skip_if_region_unavailable(
             create_cmd,
+            resource_group_location,
             checks=[
                 self.check("provisioningState", "Succeeded"),
                 self.check("agentPoolProfiles[0].availabilityZones[0]", "1"),
@@ -21735,7 +21787,9 @@ def test_aks_create_acns_with_flow_logs(
                 self.check("networkProfile.advancedNetworking.observability.enabled", True),
                 self.check("networkProfile.advancedNetworking.security.enabled", True),
                 self.check("addonProfiles.omsagent.enabled", True),
-                self.check("addonProfiles.omsagent.config.enableRetinaNetworkFlags", "True"),
+                self.check(
+                    "contains(['True', 'true'], addonProfiles.omsagent.config.enableRetinaNetworkFlags)", True
+                ),
             ],
         ).get_output_in_json()
 
@@ -21760,7 +21814,9 @@ def test_aks_create_acns_with_flow_logs(
             disable_cmd,
             checks=[
                 self.check("provisioningState", "Succeeded"),
-                self.check("addonProfiles.omsagent.config.enableRetinaNetworkFlags", "False"),
+                self.check(
+                    "contains(['False', 'false'], addonProfiles.omsagent.config.enableRetinaNetworkFlags)", True
+                ),
             ],
         )
 
@@ -21770,7 +21826,9 @@ def test_aks_create_acns_with_flow_logs(
             enable_cmd_update,
             checks=[
                 self.check("provisioningState", "Succeeded"),
-                self.check("addonProfiles.omsagent.config.enableRetinaNetworkFlags", "True"),
+                self.check(
+                    "contains(['True', 'true'], addonProfiles.omsagent.config.enableRetinaNetworkFlags)", True
+                ),
             ],
         )
 
@@ -22802,8 +22860,7 @@ def test_aks_applicationloadbalancer_update(self, resource_group, resource_group
 
         # update (currently makes a PUT no-op)
         update_cmd = (
-            "aks applicationloadbalancer update --resource-group={resource_group} --name={aks_name} "
-            "--aks-custom-headers AKSHTTPCustomFeatures=Microsoft.ContainerService/ApplicationLoadBalancerPreview"
+            "aks applicationloadbalancer update --resource-group={resource_group} --name={aks_name}"
         )
 
         self.cmd(
@@ -24333,11 +24390,7 @@ def test_aks_network_isolated_cluster(self, resource_group, resource_group_locat
 
         # reimage the existing nodepool so nodes pick up the Cache-based bootstrap artifacts
         # while they still have outbound network access
-        reimage_nodepool_cmd = (
-            "aks nodepool upgrade --resource-group {resource_group} --cluster-name {aks_name_2} "
-            "--name nodepool1 --node-image-only --yes"
-        )
-        self.cmd(reimage_nodepool_cmd, checks=[self.check("provisioningState", "Succeeded")])
+        self._reimage_nodepool(resource_group, aks_name_2, "nodepool1")
 
         # wait for the cluster to settle after the reimage before removing outbound access
         wait_after_reimage_cmd = (
@@ -25003,6 +25056,7 @@ def test_aks_bastion_enable_update_disable(
         self._cmd_or_skip_if_feature_unavailable(
             enable_cmd,
             "ManagedBastion",
+            feature_aliases=("bastionProfile",),
             checks=[
                 self.check("provisioningState", "Succeeded"),
                 self.check("networkProfile.bastionProfile.enabled", True),
@@ -25097,6 +25151,7 @@ def test_aks_bastion_enable_disable_byo_public_ip(
         self._cmd_or_skip_if_feature_unavailable(
             enable_cmd,
             "ManagedBastion",
+            feature_aliases=("bastionProfile",),
             checks=[
                 self.check("provisioningState", "Succeeded"),
                 self.check("networkProfile.bastionProfile.enabled", True),
@@ -25923,9 +25978,6 @@ def test_aks_run_blue_green_upgrade(self, resource_group, resource_group_locatio
 
     @AllowLargeResponse()
     @AKSCustomResourceGroupPreparer(
-        # eastus is capacity-constrained for standard_dc16ads_cc_v5 (verified live:
-        # "SkuNotAvailable ... Following SKUs have failed for Capacity Restrictions");
-        # eastus2 has capacity for this confidential-compute SKU.
         random_name_length=17, name_prefix="clitest", location="eastus2"
     )
     def test_aks_jwtauthenticator_cmds(self, resource_group, resource_group_location):
@@ -25935,13 +25987,8 @@ def test_aks_jwtauthenticator_cmds(self, resource_group, resource_group_location
         aks_name = self.create_random_name('cliakstest', 16)
         jwt_auth_name = self.create_random_name('jwt', 10)
 
-        # The confidential-compute SKU below is only verified to have capacity in eastus2.
-        # A compliance-mandated AZURE_CLI_TEST_FORCE_RESOURCE_GROUP_LOCATION override always
-        # wins for the resource group's own location (by design, regardless of
-        # preserve_default_location), but an AKS cluster can be created in a different region
-        # than its resource group. Pin the cluster's --location explicitly to the
-        # capacity-verified region so this test keeps passing under a forced resource-group
-        # location instead of failing with SkuNotAvailable.
+        # Keep the confidential-compute requirement; subscription availability is checked
+        # explicitly rather than replacing this SKU with an ordinary VM.
         cluster_location = "eastus2"
 
         self.kwargs.update({
@@ -25959,9 +26006,10 @@ def test_aks_jwtauthenticator_cmds(self, resource_group, resource_group_location
             "--vm-set-type=VirtualMachines --node-count 1 --vm-size standard_dc16ads_cc_v5 "
             "--enable-managed-identity --ssh-key-value={ssh_key_value} "
         )
-        self.cmd(create_cmd, checks=[
-            self.check('provisioningState', 'Succeeded'),
-        ])
+        self._cmd_or_skip_if_region_unavailable(
+            create_cmd, cluster_location, vm_size="standard_dc16ads_cc_v5",
+            checks=[self.check('provisioningState', 'Succeeded')],
+        )
 
         add_jwt_cmd = (
             "aks jwtauthenticator add --resource-group={resource_group} --cluster-name={name} "
diff --git a/src/aks-preview/azext_aks_preview/tests/latest/test_aks_provisioning_retry.py b/src/aks-preview/azext_aks_preview/tests/latest/test_aks_provisioning_retry.py
index 550166a4c7c..86501134f28 100644
--- a/src/aks-preview/azext_aks_preview/tests/latest/test_aks_provisioning_retry.py
+++ b/src/aks-preview/azext_aks_preview/tests/latest/test_aks_provisioning_retry.py
@@ -4,6 +4,7 @@
 # --------------------------------------------------------------------------------------------
 
 import json
+import inspect
 import os
 import tempfile
 import unittest
@@ -196,6 +197,7 @@ def test_retries_only_known_transient_conflicts(
         messages = [
             "Operation is not allowed: Another operation is in progress.",
             "Operation is not allowed because there's an in-progress update managed cluster operation",
+            "(AKSOperationPreempted) This operation has been preempted by another operation. Please retry later.",
             "Operation is not allowed: in-progress PutExtensionAddonHandler.PUT operation",
             "The managed cluster test is in Updating state, please wait for it to succeed.",
             "ProvisioningState of extension: Updating",
@@ -688,6 +690,182 @@ def test_ambiguous_status_error_raises_with_expect_failure(
         mock_sleep.assert_not_called()
 
 
+class TestLiveScenarioRegressions(AKSRetryTestCase):
+    def test_proxy_readiness_requires_success_marker_not_just_run_command_status(self):
+        instance = self._make_instance()
+        for message, ready in [
+            ("[stdout]\nAKS_PROXY_READY\n\n[stderr]\n", True),
+            ("[stdout]\n\n[stderr]\nFailed to start squid\n", False),
+            ("Failed executing echo AKS_PROXY_READY", False),
+        ]:
+            with self.subTest(message=message):
+                result = self._result({"value": [{"code": "ProvisioningState/succeeded", "message": message}]})
+                instance.cmd = lambda command, checks: result.assert_with_checks(checks)
+                if ready:
+                    instance._wait_for_http_proxy()
+                else:
+                    with self.assertRaises(AssertionError):
+                        instance._wait_for_http_proxy()
+
+    def test_proxy_readiness_failure_prevents_cluster_creation(self):
+        instance = self._make_instance()
+        instance.generate_ssh_keys = MagicMock(return_value="key")
+        instance.cmd = MagicMock(return_value=self._result({"id": "/subnets/aks"}))
+        instance._wait_for_http_proxy = MagicMock(side_effect=AssertionError("Proxy not ready"))
+
+        with self.assertRaisesRegex(AssertionError, "Proxy not ready"):
+            instance._setup_http_proxy_cluster("rg", "westcentralus", "cluster")
+
+        commands = [call.args[0] for call in instance.cmd.call_args_list]
+        self.assertTrue(any(command.startswith("vm create") for command in commands))
+        self.assertFalse(any(command.startswith("aks create") for command in commands))
+
+    @patch("azure.cli.testsdk.base.execute")
+    def test_node_image_upgrade_accepts_real_empty_sdk_result(self, mock_execute):
+        from azure.cli.testsdk.base import ExecutionResult
+
+        cli_ctx = MagicMock()
+        cli_ctx.invoke.return_value = 0
+        action_result = ExecutionResult(cli_ctx, "aks nodepool upgrade --node-image-only")
+        pool_result = self._result({"provisioningState": "Succeeded"})
+        mock_execute.side_effect = [action_result, pool_result]
+        instance = self._make_instance()
+        instance.cmd = lambda command, checks=None: instance._cmd_with_retry(command, checks or [], False)
+
+        instance._reimage_nodepool("rg", "cluster", "pool")
+
+        self.assertEqual(mock_execute.call_count, 2)
+
+    def test_node_image_upgrade_checks_persisted_pool_not_empty_action_result(self):
+        instance = self._make_instance()
+        instance.cmd = MagicMock()
+
+        instance._reimage_nodepool("rg", "cluster", "pool")
+
+        self.assertEqual(instance.cmd.call_count, 2)
+        upgrade_call, show_call = instance.cmd.call_args_list
+        self.assertIn("--node-image-only", upgrade_call.args[0])
+        self.assertFalse(any(
+            instance._is_provisioning_state_check(check)
+            for check in upgrade_call.kwargs.get("checks", [])
+        ))
+        self.assertEqual(
+            show_call.args[0],
+            "aks nodepool show --resource-group rg --cluster-name cluster --name pool",
+        )
+        self.assertTrue(instance._is_provisioning_state_check(show_call.kwargs["checks"][0]))
+
+    def test_node_image_upgrade_failure_is_not_recovered_by_show(self):
+        instance = self._make_instance()
+        instance.cmd = MagicMock(side_effect=CLIError("Reimage failed"))
+
+        with self.assertRaisesRegex(CLIError, "Reimage failed"):
+            instance._reimage_nodepool("rg", "cluster", "pool")
+
+        instance.cmd.assert_called_once()
+
+    def test_alb_update_uses_supported_command_arguments(self):
+        from azext_aks_preview import ContainerServiceCommandsLoader
+        from azext_aks_preview.tests.latest.mocks import MockCLI
+
+        instance = self._make_instance()
+        instance.cmd = MagicMock()
+        instance._get_versions = MagicMock(return_value=("1.33.1", "1.34.1"))
+        instance.create_random_name = MagicMock(return_value="cluster")
+        instance.generate_ssh_keys = MagicMock(return_value="ssh-key")
+        scenario = inspect.unwrap(instance.test_aks_applicationloadbalancer_update)
+        scenario(instance, "rg", "eastus")
+
+        loader = ContainerServiceCommandsLoader(MockCLI())
+        loader.load_command_table(["aks", "applicationloadbalancer", "update"])
+        command = loader.command_table["aks applicationloadbalancer update"]
+        command.load_arguments()
+        self.assertNotIn("aks_custom_headers", command.arguments)
+        update_command = instance.cmd.call_args_list[1].args[0]
+        self.assertIn("aks applicationloadbalancer update", update_command)
+        self.assertNotIn("--aks-custom-headers", update_command)
+
+
+class TestFeatureAvailabilitySkip(AKSRetryTestCase):
+    def test_managed_system_allowlist_wording_skips(self):
+        instance = self._make_instance()
+        instance.cmd = MagicMock(side_effect=CLIError(
+            "(InvalidParameter) 'ManagedSystem' is in preview and only support whitelisted subscriptions"
+        ))
+        with self.assertRaises(unittest.SkipTest):
+            instance._cmd_or_skip_if_managed_system_unavailable("aks create")
+
+    def test_managed_system_validation_error_still_fails(self):
+        instance = self._make_instance()
+        instance.cmd = MagicMock(side_effect=CLIError(
+            "(InvalidParameter) Only one ManagedSystem pool is allowed per cluster"
+        ))
+        with self.assertRaises(CLIError):
+            instance._cmd_or_skip_if_managed_system_unavailable("aks nodepool add")
+
+    def test_feature_alias_and_region_errors_skip_precisely(self):
+        cases = [
+            ("ManagedBastion", ("bastionProfile",),
+             "(InvalidParameter) Field .properties.bastionProfile is not yet supported for managed cluster."),
+            ("managedNATGatewayV2", (),
+             "(UnsupportedOutboundType) Outbound type managedNATGatewayV2 is not supported in this region."),
+        ]
+        for feature, aliases, message in cases:
+            with self.subTest(feature=feature):
+                instance = self._make_instance()
+                instance.cmd = MagicMock(side_effect=CLIError(message))
+                with self.assertRaises(unittest.SkipTest):
+                    instance._cmd_or_skip_if_feature_unavailable(
+                        "aks update", feature, feature_aliases=aliases
+                    )
+
+    def test_unrelated_feature_and_validation_errors_still_fail(self):
+        messages = [
+            "(InvalidParameter) Field .properties.otherProfile is not yet supported for managed cluster.",
+            "(InvalidParameter) Field .properties.bastionProfile has an invalid value.",
+        ]
+        for message in messages:
+            with self.subTest(message=message):
+                instance = self._make_instance()
+                instance.cmd = MagicMock(side_effect=CLIError(message))
+                with self.assertRaises(CLIError):
+                    instance._cmd_or_skip_if_feature_unavailable(
+                        "aks update", "ManagedBastion", feature_aliases=("bastionProfile",)
+                    )
+
+    def test_region_or_required_vm_size_unavailable_skips(self):
+        cases = [
+            ("westcentralus", None, "(AvailabilityZoneNotSupported) The supported zones for location 'westcentralus' are ''"),
+            ("eastus2", "Standard_DC16ads_cc_v5",
+             "(VMSizeNotSupported) Virtual Machine size: 'standard_dc16ads_cc_v5' is not supported in location 'eastus2'."),
+        ]
+        for location, vm_size, message in cases:
+            with self.subTest(location=location):
+                instance = self._make_instance()
+                instance.cmd = MagicMock(side_effect=CLIError(message))
+                with self.assertRaises(unittest.SkipTest):
+                    instance._cmd_or_skip_if_region_unavailable(
+                        "aks create", location, vm_size=vm_size
+                    )
+
+    def test_region_skip_preserves_unrelated_and_capacity_failures(self):
+        cases = [
+            ("eastus", None, "(AvailabilityZoneNotSupported) The supported zones for location 'eastus2' are ''"),
+            ("eastus2", "Standard_DC16ads_cc_v5",
+             "(VMSizeNotSupported) Virtual Machine size: 'Standard_D2s_v3' is not supported in location 'eastus2'."),
+            ("eastus2", "Standard_DC16ads_cc_v5",
+             "(ErrCode_InsufficientVCPUQuota) Insufficient quota for Standard_DC16ads_cc_v5 in location 'eastus2'."),
+        ]
+        for location, vm_size, message in cases:
+            with self.subTest(message=message):
+                instance = self._make_instance()
+                instance.cmd = MagicMock(side_effect=CLIError(message))
+                with self.assertRaises(CLIError):
+                    instance._cmd_or_skip_if_region_unavailable(
+                        "aks create", location, vm_size=vm_size
+                    )
+
+
 class TestOsSkuRetirementSkip(AKSRetryTestCase):
     """
     Unit coverage for `_cmd_or_skip_if_os_sku_retired`, which was broadened this session
diff --git a/src/aks-preview/azext_aks_preview/tests/latest/test_custom.py b/src/aks-preview/azext_aks_preview/tests/latest/test_custom.py
index 8d2e76a7a08..4fb6eba9903 100644
--- a/src/aks-preview/azext_aks_preview/tests/latest/test_custom.py
+++ b/src/aks-preview/azext_aks_preview/tests/latest/test_custom.py
@@ -2,6 +2,7 @@
 # Copyright (c) Microsoft Corporation. All rights reserved.
 # Licensed under the MIT License. See License.txt in the project root for license information.
 # --------------------------------------------------------------------------------------------
+import json
 import unittest
 from unittest.mock import Mock, patch
 
@@ -231,6 +232,76 @@ def test_aks_enable_addons_virtual_node_with_none_agent_pool_profiles(self):
             self.assertNotIn("NoneType", str(type(e)))
 
 
+class TestMonitoringProfileConsistency(unittest.TestCase):
+    @staticmethod
+    def _cluster():
+        from azext_aks_preview.vendored_sdks.azure_mgmt_preview_aks import models
+
+        return models.ManagedCluster(
+            location="westus2",
+            addon_profiles={"omsagent": models.ManagedClusterAddonProfile(
+                enabled=True,
+                config={"enableRetinaNetworkFlags": "true", "useAADAuth": "true"},
+            )},
+            azure_monitor_profile=models.ManagedClusterAzureMonitorProfile(
+                container_insights=models.ManagedClusterAzureMonitorProfileContainerInsights(
+                    enabled=True, container_network_logs="Enabled",
+                ),
+                metrics=models.ManagedClusterAzureMonitorProfileMetrics(enabled=True),
+            ),
+        )
+
+    @staticmethod
+    def _wire_properties(cluster):
+        from azext_aks_preview.vendored_sdks.azure_mgmt_preview_aks import ContainerServiceClient
+
+        class RequestCaptured(Exception):
+            pass
+
+        with ContainerServiceClient(Mock(), "sub") as client:
+            with patch.object(client._client._pipeline, "run", side_effect=RequestCaptured) as send:
+                try:
+                    client.managed_clusters.begin_create_or_update("rg", "cluster", cluster)
+                except RequestCaptured:
+                    return json.loads(send.call_args.args[0].body)["properties"]
+        raise AssertionError("The preview SDK did not construct the managed cluster PUT")
+
+    def test_disable_and_reenable_monitoring_updates_both_wire_profiles(self):
+        from azext_aks_preview.custom import _update_addons
+
+        register_aks_preview_resource_type()
+        cmd = MockCmd(MockCLI())
+        cluster = self._cluster()
+        workspace = "/subscriptions/sub/resourceGroups/rg/providers/Microsoft.OperationalInsights/workspaces/workspace"
+        for enable in (False, True):
+            with self.subTest(enable=enable):
+                _update_addons(cmd, cluster, "sub", "rg", "cluster", "monitoring", enable,
+                               workspace_resource_id=workspace)
+                payload = self._wire_properties(cluster)
+                insights = payload["azureMonitorProfile"]["containerInsights"]
+                self.assertEqual(insights["enabled"], enable)
+                self.assertEqual(payload["addonProfiles"]["omsagent"]["enabled"], enable)
+                self.assertNotEqual(insights.get("containerNetworkLogs"), "Enabled")
+                if enable:
+                    self.assertEqual(insights["logAnalyticsWorkspaceResourceId"], workspace)
+                self.assertTrue(payload["azureMonitorProfile"]["metrics"]["enabled"])
+
+    def test_disable_flow_logs_updates_both_wire_profiles(self):
+        from azext_aks_preview.managed_cluster_decorator import AKSPreviewManagedClusterUpdateDecorator
+
+        register_aks_preview_resource_type()
+        cluster = self._cluster()
+        decorator = AKSPreviewManagedClusterUpdateDecorator(
+            MockCmd(MockCLI()), Mock(), {"disable_container_network_logs": True}, CUSTOM_MGMT_AKS_PREVIEW,
+        )
+        decorator.context.attach_mc(cluster)
+        decorator.update_monitoring_profile_flow_logs(cluster)
+        payload = self._wire_properties(cluster)
+        self.assertEqual(payload["azureMonitorProfile"]["containerInsights"]["containerNetworkLogs"], "Disabled")
+        self.assertEqual(payload["addonProfiles"]["omsagent"]["config"]["enableRetinaNetworkFlags"].lower(), "false")
+        self.assertTrue(payload["azureMonitorProfile"]["metrics"]["enabled"])
+
+
 class TestAksAgentPoolGetBootstrapData(unittest.TestCase):
     def setUp(self):
         register_aks_preview_resource_type()