`ansible-runner process` puts job event artifacts in the wrong folder
Compare the two commands:

```bash
ansible-runner run demo -p test.yml
ansible-runner transmit ./demo -p test.yml | ansible-runner worker | ansible-runner process ./demo
```

### First case, `run` command

```
(env) [alancoding@alan-red-hat ansible-runner]$ tree demo
demo
├── artifacts
│   └── 6f0caada-8d38-452c-ac50-3f9bf443ccc8
│       ├── command
│       ├── fact_cache
│       │   └── localhost
│       ├── job_events
│       │   ├── 10-24418c34-c5b5-1070-847f-000000000008.json
│       │   ├── 11-4118027c-ea58-4d49-82a4-3feb28dafa15.json
│       │   ├── 12-7ccb9eaf-9031-44b3-a07f-919b940e15ca.json
│       │   ├── 13-702939c0-6843-4ade-ac04-e8e0fde48ceb.json
│       │   ├── 1-c7ff42a2-e57f-4e3d-9564-0fd09df12305.json
│       │   ├── 2-1064b388-b54b-4544-acf5-6ec64b3da011.json
│       │   ├── 3-2d8f38bb-77e0-4430-a649-f7b934306bb9.json
│       │   ├── 4-20ab0e64-5bbc-4320-8394-32405b0e681a.json
│       │   ├── 5-56299102-1dd8-4bd7-8852-d9070e14a352.json
│       │   ├── 6-24418c34-c5b5-1070-847f-000000000006.json
│       │   ├── 7-24418c34-c5b5-1070-847f-00000000000c.json
│       │   ├── 8-006c79ff-a5d2-418a-a659-dc03fc2fd7a8.json
│       │   └── 9-5f702b54-19c9-40f8-b720-2b4a74a48cef.json
│       ├── rc
│       ├── status
│       ├── stderr
│       └── stdout
├── daemon.log
├── env
│   ├── envvars
│   ├── extravars
│   ├── passwords
│   ├── settings
│   └── ssh_key
├── inventory
│   └── hosts
└── project
    ├── roles
    │   └── testrole
    │       ├── defaults
    │       │   └── main.yml
    │       ├── handlers
    │       │   └── main.yml
    │       ├── meta
    │       │   └── main.yml
    │       ├── README.md
    │       ├── tasks
    │       │   └── main.yml
    │       ├── tests
    │       │   ├── inventory
    │       │   └── test.yml
    │       └── vars
    │           └── main.yml
    └── test.yml

15 directories, 35 files
```

### Second case, streamed job

```
(env) [alancoding@alan-red-hat ansible-runner]$ tree demo
demo
├── artifacts
│   ├── command
│   ├── fact_cache
│   │   └── localhost
│   ├── job_events
│   │   ├── 10-24418c34-c5b5-ec99-f24c-000000000008.json
│   │   ├── 11-7b0daf27-2902-4610-8fca-0fbb3127c6c7.json
│   │   ├── 12-4b0e2521-ee27-45a9-99d8-eb8085ae9239.json
│   │   ├── 13-64fba58b-c91a-42a2-b136-e69d46aa8d29.json
│   │   ├── 1-6f076358-6f45-4210-92f3-062aa46a3914.json
│   │   ├── 2-28f40541-b2c1-49d4-a259-f7bd6dbf6057.json
│   │   ├── 3-94856c3f-3de6-4496-aede-17f3e9478aeb.json
│   │   ├── 4-982e6438-49cb-4089-a92d-633c354486da.json
│   │   ├── 5-1c155393-d9dc-4ff6-8a33-b131f50500c4.json
│   │   ├── 6-24418c34-c5b5-ec99-f24c-000000000006.json
│   │   ├── 7-24418c34-c5b5-ec99-f24c-00000000000c.json
│   │   ├── 8-de07d8ba-b2ad-4032-a07c-b7253680e702.json
│   │   └── 9-6b5ae4d3-eb24-40bf-ae66-0f6e80abc42a.json
│   ├── rc
│   ├── status
│   ├── stderr
│   └── stdout
├── daemon.log
├── env
│   ├── envvars
│   ├── extravars
│   ├── passwords
│   ├── settings
│   └── ssh_key
├── inventory
│   └── hosts
└── project
    ├── roles
    │   └── testrole
    │       ├── defaults
    │       │   └── main.yml
    │       ├── handlers
    │       │   └── main.yml
    │       ├── meta
    │       │   └── main.yml
    │       ├── README.md
    │       ├── tasks
    │       │   └── main.yml
    │       ├── tests
    │       │   ├── inventory
    │       │   └── test.yml
    │       └── vars
    │           └── main.yml
    └── test.yml

14 directories, 35 files
```

The remote job execution somehow drops the `ident`, although this should get passed back-and-fourth between transmit / worker / process.
diff --git a/ansible_runner/__main__.py b/ansible_runner/__main__.py
index d6222a1..0344567 100644
--- a/ansible_runner/__main__.py
+++ b/ansible_runner/__main__.py
@@ -31,7 +31,6 @@ import os
 import shutil
 import textwrap
 import tempfile
-import atexit
 
 from contextlib import contextmanager
 from uuid import uuid4
@@ -40,7 +39,7 @@ from yaml import safe_dump, safe_load
 
 from ansible_runner import run
 from ansible_runner import output
-from ansible_runner.utils import dump_artifact, Bunch
+from ansible_runner.utils import dump_artifact, Bunch, register_for_cleanup
 from ansible_runner.utils.capacity import get_cpu_count, get_mem_in_bytes, get_uuid
 from ansible_runner.runner import Runner
 from ansible_runner.exceptions import AnsibleRunnerException
@@ -408,26 +407,6 @@ DEFAULT_CLI_ARGS = {
             )
         ),
     ),
-    "execenv_cli_group": (
-        (
-            ('--container-runtime',),
-            dict(
-                dest='container_runtime',
-                default='podman',
-                help="OCI Compliant container runtime to use. Examples: podman, docker"
-            ),
-        ),
-        (
-            ('--keep-files',),
-            dict(
-                dest='keep_files',
-                action='store_true',
-                default=False,
-                help="Keep temporary files persistent on disk instead of cleaning them automatically. "
-                     "(Useful for debugging)"
-            ),
-        ),
-    ),
 }
 
 logger = logging.getLogger('ansible-runner')
@@ -643,6 +622,17 @@ def main(sys_args=None):
         action="store_true",
         help="show the execution node's Ansible Runner version along with its memory and CPU capacities"
     )
+    worker_subparser.add_argument(
+        "--delete",
+        dest="delete_directory",
+        action="store_true",
+        default=False,
+        help=(
+            "Delete existing folder (and everything in it) in the location specified by --private-data-dir. "
+            "The directory will be re-populated when the streamed data is unpacked. "
+            "Using this will also assure that the directory is deleted when the job finishes."
+        )
+    )
     process_subparser = subparser.add_parser(
         'process',
         help="Receive the output of remote ansible-runner work and distribute the results"
@@ -763,7 +753,7 @@ def main(sys_args=None):
 
     vargs = vars(args)
 
-    if vargs.get('command') in ('worker', 'process'):
+    if vargs.get('command') == 'worker':
         if vargs.get('worker_info'):
             cpu = get_cpu_count()
             mem = get_mem_in_bytes()
@@ -783,15 +773,25 @@ def main(sys_args=None):
                     }
             print(safe_dump(info, default_flow_style=True))
             parser.exit(0)
+
+        cleanup_data_dir = True
+        if vargs.get('private_data_dir'):
+            if os.path.exists(vargs.get('private_data_dir')):
+                if vargs.get('delete_directory', False):
+                    shutil.rmtree(vargs['private_data_dir'])
+                else:
+                    cleanup_data_dir = False
+        else:
+            temp_private_dir = tempfile.mkdtemp()
+            vargs['private_data_dir'] = temp_private_dir
+        if cleanup_data_dir:
+            register_for_cleanup(vargs['private_data_dir'])
+
+    if vargs.get('command') == 'process':
+        # the process command is the final destination of artifacts, user expects private_data_dir to not be cleaned up
         if not vargs.get('private_data_dir'):
             temp_private_dir = tempfile.mkdtemp()
             vargs['private_data_dir'] = temp_private_dir
-            if vargs.get('keep_files', False):
-                print("ANSIBLE-RUNNER: keeping temporary data directory: {}".format(temp_private_dir))
-            else:
-                @atexit.register
-                def conditonally_clean_cli_execenv_tempdir():
-                    shutil.rmtree(temp_private_dir)
 
     if vargs.get('command') in ('start', 'run', 'transmit'):
         if vargs.get('hosts') and not (vargs.get('module') or vargs.get('role')):
diff --git a/ansible_runner/config/runner.py b/ansible_runner/config/runner.py
index 27ed7ef..a3dcca1 100644
--- a/ansible_runner/config/runner.py
+++ b/ansible_runner/config/runner.py
@@ -31,6 +31,7 @@ from ansible_runner import output
 from ansible_runner.config._base import BaseConfig, BaseExecutionMode
 from ansible_runner.exceptions import ConfigurationError
 from ansible_runner.output import debug
+from ansible_runner.utils import register_for_cleanup
 
 
 logger = logging.getLogger('ansible-runner')
@@ -355,6 +356,9 @@ class RunnerConfig(BaseConfig):
         '''
         path = tempfile.mkdtemp(prefix='ansible_runner_pi_', dir=self.process_isolation_path)
         os.chmod(path, stat.S_IRUSR | stat.S_IWUSR | stat.S_IXUSR)
+
+        register_for_cleanup(path)
+
         return path
 
     def wrap_args_with_cgexec(self, args):
diff --git a/ansible_runner/streaming.py b/ansible_runner/streaming.py
index 484fc3a..475759d 100644
--- a/ansible_runner/streaming.py
+++ b/ansible_runner/streaming.py
@@ -15,6 +15,7 @@ import ansible_runner
 from ansible_runner.exceptions import ConfigurationError
 from ansible_runner.loader import ArtifactLoader
 import ansible_runner.plugins
+from ansible_runner.utils import register_for_cleanup
 from ansible_runner.utils.streaming import stream_dir, unstream_dir
 
 
@@ -74,6 +75,7 @@ class Worker(object):
         private_data_dir = kwargs.get('private_data_dir')
         if private_data_dir is None:
             private_data_dir = tempfile.mkdtemp()
+            register_for_cleanup(private_data_dir)
         self.private_data_dir = private_data_dir
 
         self.status = "unstarted"
diff --git a/ansible_runner/utils/__init__.py b/ansible_runner/utils/__init__.py
index 310c638..93ed65c 100644
--- a/ansible_runner/utils/__init__.py
+++ b/ansible_runner/utils/__init__.py
@@ -14,6 +14,7 @@ import threading
 import pipes
 import uuid
 import codecs
+import atexit
 
 from distutils.spawn import find_executable
 from ansible_runner.exceptions import ConfigurationError
@@ -26,6 +27,21 @@ from io import StringIO
 from six import string_types, PY2, PY3, text_type, binary_type
 
 
+def _cleanup_folder(folder):
+    try:
+        shutil.rmtree(folder)
+    except FileNotFoundError:
+        pass
+
+
+def register_for_cleanup(folder):
+    '''
+    Provide the path to a folder to make sure it is deleted when execution finishes.
+    The folder need not exist at the time when this is called.
+    '''
+    atexit.register(_cleanup_folder, folder)
+
+
 class Bunch(object):
 
     '''
diff --git a/docs/remote_jobs.rst b/docs/remote_jobs.rst
index 80fffec..f8cadd6 100644
--- a/docs/remote_jobs.rst
+++ b/docs/remote_jobs.rst
@@ -27,6 +27,7 @@ This command optionally accepts the `--private-data-dir` option.
 If provided, it will extract the contents sent from `ansible-runner transmit` into that directory.
 If no `--private-data-dir` is given, then it will extract the contents to a temporary directory,
 which will be deleted at the end of execution.
+You can use the `--delete` flag to assure that files are deleted from a location specified by `--private-data-dir` as well.
 
 The `ansible-runner process` command accepts the result stream from the worker, and fires all the normal callbacks
 and does job event processing.  In the command above, this results in printing the playbook output and saving
diff --git a/test/conftest.py b/test/conftest.py
index 9cc5cfc..6e2159a 100644
--- a/test/conftest.py
+++ b/test/conftest.py
@@ -25,6 +25,6 @@ def skipif_pre_ansible28(is_pre_ansible28):
         pytest.skip("Valid only on Ansible 2.8+")
 
 
-@pytest.fixture
+@pytest.fixture(scope='session')
 def test_data_dir():
     return os.path.join(os.path.dirname(__file__), 'data')
diff --git a/test/integration/conftest.py b/test/integration/conftest.py
index fe8a2ec..2ab20af 100644
--- a/test/integration/conftest.py
+++ b/test/integration/conftest.py
@@ -1,5 +1,8 @@
 import os
 import shutil
+import subprocess
+import json
+import yaml
 
 import pytest
 import pexpect
@@ -71,3 +74,56 @@ def clear_integration_artifacts(request):
             shutil.rmtree(path)
 
     request.addfinalizer(rm_integration_artifacts)
+
+
+class CompletedProcessProxy(object):
+
+    def __init__(self, result):
+        self.result = result
+
+    def __getattr__(self, attr):
+        return getattr(self.result, attr)
+
+    @property
+    def json(self):
+        try:
+            response_json = json.loads(self.stdout)
+        except json.JSONDecodeError:
+            pytest.fail(
+                f"Unable to convert the response to a valid json - stdout: {self.stdout}, stderr: {self.stderr}"
+            )
+        return response_json
+
+    @property
+    def yaml(self):
+        return yaml.safe_load(self.stdout)
+
+
+@pytest.fixture(scope='function')
+def cli(request):
+    def run(args, *a, **kw):
+        if not kw.pop('bare', None):
+            args = ['ansible-runner'] + args
+        kw['encoding'] = 'utf-8'
+        if 'check' not in kw:
+            # By default we want to fail if a command fails to run. Tests that
+            # want to skip this can pass check=False when calling this fixture
+            kw['check'] = True
+        if 'stdout' not in kw:
+            kw['stdout'] = subprocess.PIPE
+        if 'stderr' not in kw:
+            kw['stderr'] = subprocess.PIPE
+
+        kw.setdefault('env', os.environ.copy()).update({
+            'LANG': 'en_US.UTF-8'
+        })
+
+        try:
+            ret = CompletedProcessProxy(subprocess.run(' '.join(args), shell=True, *a, **kw))
+        except subprocess.CalledProcessError as err:
+            pytest.fail(
+                f"Running {err.cmd} resulted in a non-zero return code: {err.returncode} - stdout: {err.stdout}, stderr: {err.stderr}"
+            )
+
+        return ret
+    return run
diff --git a/test/integration/containerized/conftest.py b/test/integration/containerized/conftest.py
index 8e8f826..c5a25f9 100644
--- a/test/integration/containerized/conftest.py
+++ b/test/integration/containerized/conftest.py
@@ -1,8 +1,3 @@
-import json
-import os
-import subprocess
-import yaml
-
 from tempfile import NamedTemporaryFile
 
 import pytest
@@ -16,56 +11,3 @@ def tmp_file_maker():
             tempf.write(bytes(text, 'UTF-8'))
         return tempf.name
     return tmp_file
-
-
-class CompletedProcessProxy(object):
-
-    def __init__(self, result):
-        self.result = result
-
-    def __getattr__(self, attr):
-        return getattr(self.result, attr)
-
-    @property
-    def json(self):
-        try:
-            response_json = json.loads(self.stdout)
-        except json.JSONDecodeError:
-            pytest.fail(
-                f"Unable to convert the response to a valid json - stdout: {self.stdout}, stderr: {self.stderr}"
-            )
-        return response_json
-
-    @property
-    def yaml(self):
-        return yaml.safe_load(self.stdout)
-
-
-@pytest.fixture(scope='function')
-def cli(request):
-    def run(args, *a, **kw):
-        if not kw.pop('bare', None):
-            args = ['ansible-runner'] + args
-        kw['encoding'] = 'utf-8'
-        if 'check' not in kw:
-            # By default we want to fail if a command fails to run. Tests that
-            # want to skip this can pass check=False when calling this fixture
-            kw['check'] = True
-        if 'stdout' not in kw:
-            kw['stdout'] = subprocess.PIPE
-        if 'stderr' not in kw:
-            kw['stderr'] = subprocess.PIPE
-
-        kw.setdefault('env', os.environ.copy()).update({
-            'LANG': 'en_US.UTF-8'
-        })
-
-        try:
-            ret = CompletedProcessProxy(subprocess.run(' '.join(args), shell=True, *a, **kw))
-        except subprocess.CalledProcessError as err:
-            pytest.fail(
-                f"Running {err.cmd} resulted in a non-zero return code: {err.returncode} - stdout: {err.stdout}, stderr: {err.stderr}"
-            )
-
-        return ret
-    return run
diff --git a/test/integration/test_transmit_worker_process.py b/test/integration/test_transmit_worker_process.py
index 977a8d6..5be4115 100644
--- a/test/integration/test_transmit_worker_process.py
+++ b/test/integration/test_transmit_worker_process.py
@@ -158,6 +158,40 @@ class TestStreamingUsage:
         self.check_artifacts(process_dir, job_type)
 
 
+@pytest.fixture(scope='session')
+def transmit_stream(test_data_dir):
+    outgoing_buffer = tempfile.NamedTemporaryFile()
+    transmit_dir = os.path.join(test_data_dir, 'debug')
+    transmitter = Transmitter(_output=outgoing_buffer, private_data_dir=transmit_dir, playbook='debug.yml')
+    status, rc = transmitter.run()
+    assert rc in (None, 0)
+    assert status == 'unstarted'
+    return outgoing_buffer
+
+
+@pytest.mark.parametrize('delete', [False, True])
+def test_worker_preserve_or_delete_dir(tmp_path, cli, transmit_stream, delete):
+    worker_dir = str(tmp_path / 'for_worker')
+    os.mkdir(worker_dir)
+
+    test_file_path = os.path.join(worker_dir, 'test_file.txt')
+    with open(os.path.join(worker_dir, 'test_file.txt'), 'w') as f:
+        f.write('foobar')
+
+    with open(transmit_stream.name, 'r') as f:
+        worker_args = ['worker', '--private-data-dir', worker_dir]
+        if delete is True:
+            worker_args.append('--delete')
+        r = cli(worker_args, stdin=f)
+
+    assert '{"eof": true}' in r.stdout
+    for test_path in (test_file_path, os.path.join(worker_dir, 'project', 'debug.yml')):
+        if delete:
+            assert not os.path.exists(test_path)
+        else:
+            assert os.path.exists(test_path)
+
+
 def test_missing_private_dir_transmit(tmpdir):
     outgoing_buffer = io.BytesIO()
 
['docs/remote_jobs.rst', 'test/integration/test_transmit_worker_process.py', 'ansible_runner/__main__.py', 'ansible_runner/config/runner.py', 'test/integration/conftest.py', 'ansible_runner/utils/__init__.py', 'test/integration/containerized/conftest.py', 'test/conftest.py', 'ansible_runner/streaming.py']