code: def _mutate_user_tags_for_run(
      cls, flow_id, run_id, tags_to_add=None, tags_to_remove=None
  ):
      min_service_version_with_tag_mutation = "2.3.0"
      if cls._supports_tag_mutation is None:
          version = cls._version(None)
          cls._supports_tag_mutation = (
              version is not None
              and version_parse(version)
              >= version_parse(min_service_version_with_tag_mutation)
          )
      if not cls._supports_tag_mutation:
          raise ServiceException(
              "Adding or removing tags on a run requires the Metaflow service to be "
              "at least version %s. Please upgrade your service."
              % (min_service_version_with_tag_mutation,)
          )
  
      url = (
          ServiceMetadataProvider._obj_path(flow_id, run_id)
          + "/tag/mutate"
      )
      tag_mutation_data = {
          # mutate_user_tags_for_run() should have already ensured that this is a list, so let's be tolerant here
          "tags_to_add": list(tags_to_add or []),
          "tags_to_remove": list(tags_to_remove or []),
      }
      tries = 1
      status_codes_seen = set()
      # try up to 10 times, with a gentle exponential backoff (1.4-1.6x)
      while True:
          resp, _ = cls._request(
              None,
              url,
              "PATCH",
              data=tag_mutation_data,
              return_raw_resp=True,
          )
          status_codes_seen.add(resp.status_code)
          # happy path
          if resp.status_code < 300:
              return frozenset(resp.json()["tags"])
          # definitely NOT retriable
          if resp.status_code in (400, 422):
              raise MetaflowTaggingError(
                  "Metadata service says: %s" % (resp.text,)
              )
          # if we get here, mutation failure is possibly retriable
          if tries >= 10:
              # if we ever received 409 on any of our attempts, report "conflicting updates" blurb to user
              if 409 in status_codes_seen:
                  raise MetaflowTaggingError(
                      "Tagging failed due to too many conflicting updates from other processes"
                  )
              # No 409's seen... raise a more generic error
              raise MetaflowTaggingError(
                  "Tagging failed after %d tries" % tries
              )
          time.sleep(0.3 * random.uniform(1.4, 1.6) ** tries)
          tries += 1
  
imports: import os
  import random
  import requests
  import time
  from metaflow.exception import MetaflowException, MetaflowTaggingError, MetaflowInternalError
  from metaflow.metaflow_config import SERVICE_RETRY_COUNT, SERVICE_HEADERS, SERVICE_URL
  from metaflow.metadata import MetadataProvider
  from metaflow.metadata.heartbeat import HB_URL_KEY
  from metaflow.sidecar import Message, MessageTypes, Sidecar
  from metaflow.util import version_parse
label_desc: Use image node_id 16 for calling the ServiceMetadataProvider obj's underlying member method code with example usage: obj._mutate_user_tags_for_run(cls, flow_id, run_id, tags_to_add, tags_to_remove) and returns: frozenset