diff --git a/bin/ansible-connection b/bin/ansible-connection
deleted file mode 120000
index a20affdbe6a..00000000000
--- a/bin/ansible-connection
+++ /dev/null
@@ -1 +0,0 @@
-../lib/ansible/cli/scripts/ansible_connection_cli_stub.py
\ No newline at end of file
diff --git a/changelogs/fragments/ansible_connection_path.yml b/changelogs/fragments/ansible_connection_path.yml
new file mode 100644
index 00000000000..d1eb1866fbe
--- /dev/null
+++ b/changelogs/fragments/ansible_connection_path.yml
@@ -0,0 +1,8 @@
+bugfixes:
+  - persistent connection plugins - The correct Ansible persistent connection helper is now always used.
+    Previously, the wrong script could be used, depending on the value of the ``PATH`` environment variable.
+    As a result, users were sometimes required to set ``ANSIBLE_CONNECTION_PATH`` to use the correct script.
+deprecated_features:
+  - persistent connection plugins - The ``ANSIBLE_CONNECTION_PATH`` config option no longer has any effect, and will be removed in a future release.
+breaking_changes:
+  - persistent connection plugins - The ``ANSIBLE_CONNECTION_PATH`` config option no longer has any effect.
diff --git a/lib/ansible/cli/scripts/ansible_connection_cli_stub.py b/lib/ansible/cli/scripts/ansible_connection_cli_stub.py
old mode 100755
new mode 100644
index 9455b9851a9..701dcdaa198
--- a/lib/ansible/cli/scripts/ansible_connection_cli_stub.py
+++ b/lib/ansible/cli/scripts/ansible_connection_cli_stub.py
@@ -1,10 +1,8 @@
-#!/usr/bin/env python
 # Copyright: (c) 2017, Ansible Project
 # GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
 from __future__ import annotations
 
 import fcntl
-import hashlib
 import io
 import os
 import pickle
@@ -40,13 +38,6 @@ def read_stream(byte_stream):
     if len(data) < size:
         raise Exception("EOF found before data was complete")
 
-    data_hash = to_text(byte_stream.readline().strip())
-    if data_hash != hashlib.sha1(data).hexdigest():
-        raise Exception("Read {0} bytes, but data did not match checksum".format(size))
-
-    # restore escaped loose \r characters
-    data = data.replace(br'\r', b'\r')
-
     return data
 
 
@@ -221,7 +212,7 @@ def main(args=None):
     """ Called to initiate the connect to the remote device
     """
 
-    parser = opt_help.create_base_parser(prog='ansible-connection')
+    parser = opt_help.create_base_parser(prog=None)
     opt_help.add_verbosity_options(parser)
     parser.add_argument('playbook_pid')
     parser.add_argument('task_uuid')
diff --git a/lib/ansible/config/base.yml b/lib/ansible/config/base.yml
index 20795407a34..f032793e98e 100644
--- a/lib/ansible/config/base.yml
+++ b/lib/ansible/config/base.yml
@@ -1,6 +1,14 @@
 # Copyright (c) 2017 Ansible Project
 # GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
 ---
+_ANSIBLE_CONNECTION_PATH:
+  env:
+  - name: _ANSIBLE_CONNECTION_PATH
+  name: Overrides the location of the Ansible persistent connection helper script.
+  description:
+    - For internal use only.
+  type: path
+  version_added: "2.18"
 ANSIBLE_HOME:
   name: The Ansible home path
   description:
@@ -25,6 +33,9 @@ ANSIBLE_CONNECTION_PATH:
   - {key: ansible_connection_path, section: persistent_connection}
   yaml: {key: persistent_connection.ansible_connection_path}
   version_added: "2.8"
+  deprecated:
+    why: This setting has no effect.
+    version: "2.22"
 ANSIBLE_COW_SELECTION:
   name: Cowsay filter selection
   default: default
diff --git a/lib/ansible/executor/task_executor.py b/lib/ansible/executor/task_executor.py
index a8e24f5f36a..b65dc420cca 100644
--- a/lib/ansible/executor/task_executor.py
+++ b/lib/ansible/executor/task_executor.py
@@ -4,23 +4,23 @@
 from __future__ import annotations
 
 import os
-import pty
 import time
 import json
+import pathlib
 import signal
 import subprocess
 import sys
-import termios
 import traceback
 
 from ansible import constants as C
+from ansible.cli import scripts
 from ansible.errors import AnsibleError, AnsibleParserError, AnsibleUndefinedVariable, AnsibleConnectionFailure, AnsibleActionFail, AnsibleActionSkip
 from ansible.executor.task_result import TaskResult
 from ansible.executor.module_common import get_action_args_with_defaults
 from ansible.module_utils.parsing.convert_bool import boolean
 from ansible.module_utils.six import binary_type
 from ansible.module_utils.common.text.converters import to_text, to_native
-from ansible.module_utils.connection import write_to_file_descriptor
+from ansible.module_utils.connection import write_to_stream
 from ansible.playbook.conditional import Conditional
 from ansible.playbook.task import Task
 from ansible.plugins import get_plugin_class
@@ -1179,26 +1179,19 @@ class TaskExecutor:
         return handler, module
 
 
+CLI_STUB_NAME = 'ansible_connection_cli_stub.py'
+
+
 def start_connection(play_context, options, task_uuid):
     '''
     Starts the persistent connection
     '''
-    candidate_paths = [C.ANSIBLE_CONNECTION_PATH or os.path.dirname(sys.argv[0])]
-    candidate_paths.extend(os.environ.get('PATH', '').split(os.pathsep))
-    for dirname in candidate_paths:
-        ansible_connection = os.path.join(dirname, 'ansible-connection')
-        if os.path.isfile(ansible_connection):
-            display.vvvv("Found ansible-connection at path {0}".format(ansible_connection))
-            break
-    else:
-        raise AnsibleError("Unable to find location of 'ansible-connection'. "
-                           "Please set or check the value of ANSIBLE_CONNECTION_PATH")
 
     env = os.environ.copy()
     env.update({
         # HACK; most of these paths may change during the controller's lifetime
         # (eg, due to late dynamic role includes, multi-playbook execution), without a way
-        # to invalidate/update, ansible-connection won't always see the same plugins the controller
+        # to invalidate/update, the persistent connection helper won't always see the same plugins the controller
         # can.
         'ANSIBLE_BECOME_PLUGINS': become_loader.print_paths(),
         'ANSIBLE_CLICONF_PLUGINS': cliconf_loader.print_paths(),
@@ -1211,30 +1204,19 @@ def start_connection(play_context, options, task_uuid):
     verbosity = []
     if display.verbosity:
         verbosity.append('-%s' % ('v' * display.verbosity))
-    python = sys.executable
-    master, slave = pty.openpty()
+
+    if not (cli_stub_path := C.config.get_config_value('_ANSIBLE_CONNECTION_PATH')):
+        cli_stub_path = str(pathlib.Path(scripts.__file__).parent / CLI_STUB_NAME)
+
     p = subprocess.Popen(
-        [python, ansible_connection, *verbosity, to_text(os.getppid()), to_text(task_uuid)],
-        stdin=slave, stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=env
+        [sys.executable, cli_stub_path, *verbosity, to_text(os.getppid()), to_text(task_uuid)],
+        stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=env,
     )
-    os.close(slave)
-
-    # We need to set the pty into noncanonical mode. This ensures that we
-    # can receive lines longer than 4095 characters (plus newline) without
-    # truncating.
-    old = termios.tcgetattr(master)
-    new = termios.tcgetattr(master)
-    new[3] = new[3] & ~termios.ICANON
-
-    try:
-        termios.tcsetattr(master, termios.TCSANOW, new)
-        write_to_file_descriptor(master, options)
-        write_to_file_descriptor(master, play_context.serialize())
-
-        (stdout, stderr) = p.communicate()
-    finally:
-        termios.tcsetattr(master, termios.TCSANOW, old)
-    os.close(master)
+
+    write_to_stream(p.stdin, options)
+    write_to_stream(p.stdin, play_context.serialize())
+
+    (stdout, stderr) = p.communicate()
 
     if p.returncode == 0:
         result = json.loads(to_text(stdout, errors='surrogate_then_replace'))
diff --git a/lib/ansible/module_utils/connection.py b/lib/ansible/module_utils/connection.py
index cc889696e70..b6720125855 100644
--- a/lib/ansible/module_utils/connection.py
+++ b/lib/ansible/module_utils/connection.py
@@ -29,8 +29,8 @@
 from __future__ import annotations
 
 import os
-import hashlib
 import json
+import pickle
 import socket
 import struct
 import traceback
@@ -40,30 +40,14 @@ from functools import partial
 from ansible.module_utils.common.text.converters import to_bytes, to_text
 from ansible.module_utils.common.json import AnsibleJSONEncoder
 from ansible.module_utils.six import iteritems
-from ansible.module_utils.six.moves import cPickle
 
 
-def write_to_file_descriptor(fd, obj):
-    """Handles making sure all data is properly written to file descriptor fd.
+def write_to_stream(stream, obj):
+    """Write a length+newline-prefixed pickled object to a stream."""
+    src = pickle.dumps(obj)
 
-    In particular, that data is encoded in a character stream-friendly way and
-    that all data gets written before returning.
-    """
-    # Need to force a protocol that is compatible with both py2 and py3.
-    # That would be protocol=2 or less.
-    # Also need to force a protocol that excludes certain control chars as
-    # stdin in this case is a pty and control chars will cause problems.
-    # that means only protocol=0 will work.
-    src = cPickle.dumps(obj, protocol=0)
-
-    # raw \r characters will not survive pty round-trip
-    # They should be rehydrated on the receiving end
-    src = src.replace(b'\r', br'\r')
-    data_hash = to_bytes(hashlib.sha1(src).hexdigest())
-
-    os.write(fd, b'%d\n' % len(src))
-    os.write(fd, src)
-    os.write(fd, b'%s\n' % data_hash)
+    stream.write(b'%d\n' % len(src))
+    stream.write(src)
 
 
 def send_data(s, data):
@@ -146,7 +130,7 @@ class Connection(object):
             data = json.dumps(req, cls=AnsibleJSONEncoder, vault_to_text=True)
         except TypeError as exc:
             raise ConnectionError(
-                "Failed to encode some variables as JSON for communication with ansible-connection. "
+                "Failed to encode some variables as JSON for communication with the persistent connection helper. "
                 "The original exception was: %s" % to_text(exc)
             )
 
@@ -176,7 +160,7 @@ class Connection(object):
         if response['id'] != reqid:
             raise ConnectionError('invalid json-rpc id received')
         if "result_type" in response:
-            response["result"] = cPickle.loads(to_bytes(response["result"]))
+            response["result"] = pickle.loads(to_bytes(response["result"], errors="surrogateescape"))
 
         return response
 
diff --git a/lib/ansible/utils/display.py b/lib/ansible/utils/display.py
index e425a867dc9..ddf59179876 100644
--- a/lib/ansible/utils/display.py
+++ b/lib/ansible/utils/display.py
@@ -424,7 +424,7 @@ class Display(metaclass=Singleton):
                 msg2 = msg2 + u'\n'
 
             # Note: After Display() class is refactored need to update the log capture
-            # code in 'bin/ansible-connection' (and other relevant places).
+            # code in 'cli/scripts/ansible_connection_cli_stub.py' (and other relevant places).
             if not stderr:
                 fileobj = sys.stdout
             else:
diff --git a/lib/ansible/utils/jsonrpc.py b/lib/ansible/utils/jsonrpc.py
index 37b286a0a52..82d1c02ea12 100644
--- a/lib/ansible/utils/jsonrpc.py
+++ b/lib/ansible/utils/jsonrpc.py
@@ -83,7 +83,7 @@ class JsonRpcServer(object):
             result = to_text(result)
         if not isinstance(result, text_type):
             response["result_type"] = "pickle"
-            result = to_text(pickle.dumps(result, protocol=0))
+            result = to_text(pickle.dumps(result), errors='surrogateescape')
         response['result'] = result
         return response
 
diff --git a/setup.cfg b/setup.cfg
index 0b6b627e96c..29fb6ef2671 100644
--- a/setup.cfg
+++ b/setup.cfg
@@ -100,7 +100,6 @@ ansible_test =
 #     ansible-playbook = ansible.cli.playbook:main
 #     ansible-pull = ansible.cli.pull:main
 #     ansible-vault = ansible.cli.vault:main
-#     ansible-connection = ansible.cli.scripts.ansible_connection_cli_stub:main
 #     ansible-test = ansible_test._util.target.cli.ansible_test_cli_stub:main
 
 [flake8]
diff --git a/setup.py b/setup.py
index 26e680e36fc..7f56b203d96 100644
--- a/setup.py
+++ b/setup.py
@@ -24,7 +24,6 @@ setup(
             'ansible-playbook=ansible.cli.playbook:main',
             'ansible-pull=ansible.cli.pull:main',
             'ansible-vault=ansible.cli.vault:main',
-            'ansible-connection=ansible.cli.scripts.ansible_connection_cli_stub:main',
         ],
     },
 )
diff --git a/test/integration/targets/packaging_cli-doc/verify.py b/test/integration/targets/packaging_cli-doc/verify.py
index 7fe9d1d371d..f556968f6ce 100755
--- a/test/integration/targets/packaging_cli-doc/verify.py
+++ b/test/integration/targets/packaging_cli-doc/verify.py
@@ -6,7 +6,6 @@ import pathlib
 import sys
 
 exclude_programs = {
-    'ansible-connection',
     'ansible-test',
 }
 
diff --git a/test/lib/ansible_test/_internal/ansible_util.py b/test/lib/ansible_test/_internal/ansible_util.py
index 885489f4ba9..c2d363506b3 100644
--- a/test/lib/ansible_test/_internal/ansible_util.py
+++ b/test/lib/ansible_test/_internal/ansible_util.py
@@ -115,14 +115,16 @@ def ansible_environment(args: CommonConfig, color: bool = True, ansible_config:
         # enabled even when not using code coverage to surface warnings when worker processes do not exit cleanly
         ANSIBLE_WORKER_SHUTDOWN_POLL_COUNT='100',
         ANSIBLE_WORKER_SHUTDOWN_POLL_DELAY='0.1',
+        # ansible-test specific environment variables require an 'ANSIBLE_TEST_' prefix to distinguish them from ansible-core env vars defined by config
+        ANSIBLE_TEST_ANSIBLE_LIB_ROOT=ANSIBLE_LIB_ROOT,  # used by the coverage injector
     )
 
     if isinstance(args, IntegrationConfig) and args.coverage:
-        # standard path injection is not effective for ansible-connection, instead the location must be configured
-        # ansible-connection only requires the injector for code coverage
+        # standard path injection is not effective for the persistent connection helper, instead the location must be configured
+        # it only requires the injector for code coverage
         # the correct python interpreter is already selected using the sys.executable used to invoke ansible
         ansible.update(
-            ANSIBLE_CONNECTION_PATH=os.path.join(get_injector_path(), 'ansible-connection'),
+            _ANSIBLE_CONNECTION_PATH=os.path.join(get_injector_path(), 'ansible_connection_cli_stub.py'),
         )
 
     if isinstance(args, PosixIntegrationConfig):
diff --git a/test/lib/ansible_test/_internal/constants.py b/test/lib/ansible_test/_internal/constants.py
index fdf2d954e60..985f4954c00 100644
--- a/test/lib/ansible_test/_internal/constants.py
+++ b/test/lib/ansible_test/_internal/constants.py
@@ -37,7 +37,6 @@ SECCOMP_CHOICES = [
 ANSIBLE_BIN_SYMLINK_MAP = {
     'ansible': '../lib/ansible/cli/adhoc.py',
     'ansible-config': '../lib/ansible/cli/config.py',
-    'ansible-connection': '../lib/ansible/cli/scripts/ansible_connection_cli_stub.py',
     'ansible-console': '../lib/ansible/cli/console.py',
     'ansible-doc': '../lib/ansible/cli/doc.py',
     'ansible-galaxy': '../lib/ansible/cli/galaxy.py',
diff --git a/test/lib/ansible_test/_internal/util_common.py b/test/lib/ansible_test/_internal/util_common.py
index a6904971b8a..dfda54593e6 100644
--- a/test/lib/ansible_test/_internal/util_common.py
+++ b/test/lib/ansible_test/_internal/util_common.py
@@ -300,6 +300,7 @@ def get_injector_path() -> str:
     injector_names = sorted(list(ANSIBLE_BIN_SYMLINK_MAP) + [
         'importer.py',
         'pytest',
+        'ansible_connection_cli_stub.py',
     ])
 
     scripts = (
diff --git a/test/lib/ansible_test/_util/target/injector/python.py b/test/lib/ansible_test/_util/target/injector/python.py
index 82f59b1d191..8b9fa86455e 100644
--- a/test/lib/ansible_test/_util/target/injector/python.py
+++ b/test/lib/ansible_test/_util/target/injector/python.py
@@ -6,12 +6,15 @@ import importlib.util
 import os
 import sys
 
+NETWORKING_CLI_STUB_SCRIPT = 'ansible_connection_cli_stub.py'
+
 
 def main():
     """Main entry point."""
     name = os.path.basename(__file__)
     args = [sys.executable]
 
+    ansible_lib_root = os.environ.get('ANSIBLE_TEST_ANSIBLE_LIB_ROOT')
     coverage_config = os.environ.get('COVERAGE_CONF')
     coverage_output = os.environ.get('COVERAGE_FILE')
 
@@ -33,6 +36,8 @@ def main():
         args += ['-m', 'pytest']
     elif name == 'importer.py':
         args += [find_program(name, False)]
+    elif name == NETWORKING_CLI_STUB_SCRIPT:
+        args += [os.path.join(ansible_lib_root, 'cli/scripts', NETWORKING_CLI_STUB_SCRIPT)]
     else:
         args += [find_program(name, True)]
 
diff --git a/test/lib/ansible_test/config/config.yml b/test/lib/ansible_test/config/config.yml
index 9fca7afb9b0..ce43713c0a8 100644
--- a/test/lib/ansible_test/config/config.yml
+++ b/test/lib/ansible_test/config/config.yml
@@ -17,7 +17,7 @@ modules:
   #                   This is the default value if no configuration is provided.
   #  - 'controller' - All Python versions supported by the Ansible controller.
   #                   This indicates the modules/module_utils can only run on the controller.
-  #                   Intended for use only with modules/module_utils that depend on ansible-connection, which only runs on the controller.
+  #                   Intended for use only with modules/module_utils that depend on the Ansible persistent connection helper, which only runs on the controller.
   #                   Unit tests for modules/module_utils will be permitted to import any Ansible code, instead of only module_utils.
   #  - SpecifierSet - A PEP 440 specifier set indicating the supported Python versions.
   #                   This is only needed when modules/module_utils do not support all Python versions supported by Ansible.
diff --git a/test/sanity/ignore.txt b/test/sanity/ignore.txt
index dc8fd0ee3f0..a4cfe5c5748 100644
--- a/test/sanity/ignore.txt
+++ b/test/sanity/ignore.txt
@@ -1,4 +1,3 @@
-lib/ansible/cli/scripts/ansible_connection_cli_stub.py shebang
 lib/ansible/config/base.yml no-unwanted-files
 lib/ansible/executor/powershell/async_watchdog.ps1 pslint:PSCustomUseLiteralPath
 lib/ansible/executor/powershell/async_wrapper.ps1 pslint:PSCustomUseLiteralPath