async def METHOD_NAME(
    console: Console,
    repl_subsystem: ReplSubsystem,
    specified_targets: FilteredTargets,
    union_membership: UnionMembership,
    complete_env: CompleteEnvironmentVars,
) -> Repl:
    await _warn_on_non_local_environments(specified_targets, "the `repl` goal")
    # TODO: When we support multiple languages, detect the default repl to use based
    #  on the targets.  For now we default to the python repl.
    repl_shell_name = repl_subsystem.shell or "python"
    implementations = {impl.name: impl for impl in union_membership[ReplImplementation]}
    repl_implementation_cls = implementations.get(repl_shell_name)
    if repl_implementation_cls is None:
        available = sorted(implementations.keys())
        console.print_stderr(
            softwrap(
                f"""
                {repr(repl_shell_name)} is not a registered REPL. Available REPLs (which may
                be specified through the option `--repl-shell`): {available}
                """
            )
        )
        return Repl(-1)
    repl_impl = repl_implementation_cls(targets=specified_targets)
    request = await Get(ReplRequest, ReplImplementation, repl_impl)
    env = {**complete_env, **request.extra_env}
    result = await Effect(
        InteractiveProcessResult,
        InteractiveProcess(
            argv=request.args,
            env=env,
            input_digest=request.digest,
            run_in_workspace=request.run_in_workspace,
            restartable=repl_subsystem.restartable,
            immutable_input_digests=request.immutable_input_digests,
            append_only_caches=request.append_only_caches,
        ),
    )
    return Repl(result.exit_code)