code: def aot_wrapper_dedupe(
      flat_fn,
      flat_args: List[Tensor],
      aot_config: AOTConfig,
      *,
      compiler_fn,
      fw_metadata,
  ):
      # Use information about whether or not flat_fn mutates its arguments
      # or not to handle dupe args
  
      # Strategy 1: For any input that is not mutated, we can leafify it if we
      # need to remove a duplicate.
      leaf_flat_args = []
      args_set = set()
      ok = True
  
      for i, a in enumerate(flat_args):
          if not isinstance(a, torch.Tensor):
              leaf_flat_args.append(a)
          elif a not in args_set:
              args_set.add(a)
              leaf_flat_args.append(a)
          elif (
              not fw_metadata.input_info[i].mutates_data
              and not fw_metadata.input_info[i].mutates_metadata
          ):
              leaf_flat_args.append(
                  a.detach().requires_grad_(a.requires_grad)
              )
          else:
              ok = False
              break
  
      if ok:
          return compiler_fn(
              flat_fn,
              leaf_flat_args,
              aot_config,
              fw_metadata=fw_metadata,
          )
  
      if requires_subclass_dispatch(leaf_flat_args, fw_metadata):
          raise RuntimeError(
              """\
  Encountered duplicate inputs that are mutated in the graph, but at least one input/output
  to the graph is a tensor subclass. This is not supported today. You can try to
  remove the aliasing yourself as a workaround, or otherwise file an issue on github."""
          )
  
      # export path: ban duplicate inputs for now, add later if requested.
      if aot_config.is_export:
          raise RuntimeError(
              f"""\
  Encountered duplicated inputs that are mutated in the graph you are trying to export.
  This functionality is currently not supported. If needed, please file a github issue.
  
  fw_metadata={str(fw_metadata)}
          """
          )
  
      # Strategy 2: Duplicate specialize.
      #
      # In Haskell types, suppose you have:
      #
      #   add_dupe_args :: DedupedArgs -> Args
      #   remove_dupe_args :: Args -> DedupedArgs
      #
      #   compiler_fn
      #       :: (DedupedArgs -> R) -> DedupedArgs -> AOTConfig -> (DedupedArgs -> R)
      #   deped_compiler_fn
      #       :: (Args -> R) -> Args -> AOTConfig -> (Args -> R)
      #
      # Then the code below can be written in point-free style as:
      #
      #   deduped_compiler_fn f a c =
      #       compiler_fn (f . add_dupe_args) (remove_dupe_args a) c . remove_dupe_args
      #
      # Suppose you have:
      #
      #   [a, b, a, c]
      #
      # We want:
      #
      #   remove_dupe_args([a, b, a, c]) == [a, b, c]
      #   add_dupe_args([a, b, c]) == [a, b, a, c]
      #
      # This is done via (respectively):
      #
      #   seen_args = {a: 0, b: 1, c: 2}
      #   enumerate(add_dupe_map) = [  # how to get args from the deduped list
      #       (0, 0),
      #       (1, 1),
      #       (2, 0),
      #       (3, 2),
      #   ]
      #   keep_arg_mask = [True, True, False, True]
  
      seen_args: Dict[Tensor, int] = {}
      keep_arg_mask = []
      # Implicitly map duped arg position (list index) to de-duped arg position
      add_dupe_map: List[int] = []
      duped_arg_len = len(flat_args)
  
      j = 0  # index into deduped_flat_args
      for t in flat_args:
          if isinstance(t, torch.Tensor):
              if t in seen_args:
                  keep_arg_mask.append(False)
                  add_dupe_map.append(seen_args[t])
                  continue
              seen_args[t] = j
  
          keep_arg_mask.append(True)
          add_dupe_map.append(j)
          j += 1
      assert (
          len(add_dupe_map) == duped_arg_len
      ), f"Expects add_dupe_map to have length {duped_arg_len} but got {len(add_dupe_map)}"
  
      # NB: Hot path, avoid set lookups here
      # TODO: Can avoid the zip here too, probably
      def remove_dupe_args(args):
          return [t for t, keep in zip(args, keep_arg_mask) if keep]
  
      def add_dupe_args(args):
          return [args[add_dupe_map[i]] for i in range(duped_arg_len)]
  
      deduped_flat_args = remove_dupe_args(flat_args)
  
      # Update our input metadata to remove duped input metadata.
      updated_fw_metadata = remove_dupe_metadata(
          fw_metadata, keep_arg_mask, add_dupe_map
      )
  
      if (
          tracing_context := TracingContext.try_get()
          and aot_config.aot_autograd_arg_pos_to_source
      ):
          # TODO(voz): This structure is 1:1, we could consider an alternate structure like
          # kept_pos:[dupe_arg_pos], however, add_dupe_map is 1:1 so we would need a new structure there,
          # which feels like needless complexity for a tiny bit of efficiency at this point.
          for dupe_arg_pos, (kept_pos, keep_arg) in enumerate(
              zip(add_dupe_map, keep_arg_mask)
          ):
              if not keep_arg:
                  dupe_arg_source = (
                      aot_config.aot_autograd_arg_pos_to_source[
                          dupe_arg_pos
                      ]
                  )
                  kept_arg_source = (
                      aot_config.aot_autograd_arg_pos_to_source[
                          kept_pos
                      ]
                  )
                  tracing_context.guards_context.aotautograd_guards.append(  # type: ignore[attr-defined]
                      DuplicateInputs(kept_arg_source, dupe_arg_source)
                  )
  
      @wraps(flat_fn)
      def wrapped_flat_fn(*args):
          return flat_fn(*add_dupe_args(args))
  
      if config.debug_assert:
          ref_fw_metadata = run_functionalized_fw_and_collect_metadata(
              wrapped_flat_fn,
              keep_input_mutations=fw_metadata.keep_input_mutations,
              is_train=fw_metadata.is_train,
          )(*deduped_flat_args)
          assert (
              ref_fw_metadata == updated_fw_metadata
          ), f"ref_metadata={str(ref_fw_metadata)}, actual_metadata={str(updated_fw_metadata)}"
  
      compiled_fn = compiler_fn(
          wrapped_flat_fn,
          deduped_flat_args,
          aot_config,
          fw_metadata=updated_fw_metadata,
      )
  
      if not hasattr(compiled_fn, "_boxed_call"):
          compiled_fn = make_boxed_func(compiled_fn)
  
      @wraps(compiled_fn)
      def wrapped_compiled_fn(args):
          deduped_args = remove_dupe_args(args)
          args.clear()
          return compiled_fn(deduped_args)
  
      wrapped_compiled_fn._boxed_call = True  # type: ignore[attr-defined]
  
      # This can be uncommented when we properly guard for duplicates,
      # but right now we must not do it.
      # if not config.debug_assert:
      #     return wrapped_compiled_fn
  
      @wraps(wrapped_compiled_fn)
      def debugged_compiled_fn(args):
          # Test that the computed remove/add arg functions are an inverse
          new_args = add_dupe_args(remove_dupe_args(args))
          seen: Dict[Any, None] = {}
          for i, (x, y) in enumerate(zip(new_args, args)):
              seen[y] = None
              assert x is y, format_guard_bug_msg(
                  aot_config,
                  f"{describe_input(i, aot_config)} would be a duplicate of "
                  f"{describe_input(add_dupe_map[i], aot_config)}",
              )
          # This is only an error if there is metadata mutation on both of
          # the duped arguments; in this case, we need to know what order
          # the metadata mutation applies in.  You'll get the correct result
          # otherwise, because a graph that assumes distinct inputs works if
          # you dupe the inputs (the gradient contributions from each input
          # will get summed up appropriately.)
          #
          # TODO: work out how to setup this assert correctly
          """
          assert len(seen) == unique_args, format_guard_bug_msg(aot_config,
              f"there would be {unique_args} distinct arguments"
          )
          """
          return wrapped_compiled_fn(args)
  
      debugged_compiled_fn._boxed_call = True  # type: ignore[attr-defined]
  
      return debugged_compiled_fn
  
imports: import collections
  import pprint
  from functools import wraps
  from typing import Any, Callable, Dict, List, Optional, Tuple, Union
  import torch
  import torch.utils.dlpack
  from torch import Tensor
  from torch._guards import DuplicateInputs, TracingContext
  from torch._prims_common import CUDARngStateHelper
  from torch.multiprocessing.reductions import StorageWeakRef
  from ..None import config
  from .collect_metadata_analysis import run_functionalized_fw_and_collect_metadata
  from .functional_utils import gen_alias_from_base
  from .input_output_analysis import compute_overlapping_inputs, create_synthetic_base_metadata, remove_dupe_metadata
  from .logging_utils import describe_input, format_guard_bug_msg
  from .schemas import AOTConfig, InputAliasInfo, OutputType, SubclassCreationMeta, TensorAlias, ViewAndMutationMeta
  from .subclass_utils import requires_subclass_dispatch, unwrap_tensor_subclasses, wrap_tensor_subclasses
  from .utils import call_func_at_runtime_with_args, make_boxed_func, partial_flatten_asdict, strict_zip
label_desc: Use image node_id 4 for calling a global function with example usage: aot_wrapper_dedupe(flat_fn, flat_args, aot_config) and returns: debugged_compiled_fn, compiler_fn, unknown, unknown, flat_fn, compiled_fn, wrapped_compiled_fn