diff --git a/src/ansible_runner/_internal/__init__.py b/src/ansible_runner/_internal/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/src/ansible_runner/_internal/_dump_artifacts.py b/src/ansible_runner/_internal/_dump_artifacts.py
new file mode 100644
index 000000000..cc88c6ee4
--- /dev/null
+++ b/src/ansible_runner/_internal/_dump_artifacts.py
@@ -0,0 +1,124 @@
+from __future__ import annotations
+
+import fcntl
+import hashlib
+import json
+import os
+import stat
+import tempfile
+
+from collections.abc import MutableMapping
+
+from ansible_runner.config.runner import RunnerConfig
+from ansible_runner.utils import isinventory, isplaybook
+
+
+def dump_artifacts(config: RunnerConfig) -> None:
+    """Introspect the arguments and dump objects to disk"""
+    if config.role:
+        role = {'name': config.role}
+        if config.role_vars:
+            role['vars'] = config.role_vars
+
+        hosts = config.host_pattern or 'all'
+        play = [{'hosts': hosts, 'roles': [role]}]
+
+        if config.role_skip_facts:
+            play[0]['gather_facts'] = False
+
+        config.playbook = play
+
+        if config.envvars is None:
+            config.envvars = {}
+
+        roles_path = config.roles_path
+        if not roles_path:
+            roles_path = os.path.join(config.private_data_dir, 'roles')
+        else:
+            roles_path += f":{os.path.join(config.private_data_dir, 'roles')}"
+
+        config.envvars['ANSIBLE_ROLES_PATH'] = roles_path
+
+    playbook = config.playbook
+    if playbook:
+        # Ensure the play is a list of dictionaries
+        if isinstance(playbook, MutableMapping):
+            playbook = [playbook]
+
+        if isplaybook(playbook):
+            path = os.path.join(config.private_data_dir, 'project')
+            config.playbook = dump_artifact(json.dumps(playbook), path, 'main.json')
+
+    obj = config.inventory
+    if obj and isinventory(obj):
+        path = os.path.join(config.private_data_dir, 'inventory')
+        if isinstance(obj, MutableMapping):
+            config.inventory = dump_artifact(json.dumps(obj), path, 'hosts.json')
+        elif isinstance(obj, str):
+            if not os.path.exists(os.path.join(path, obj)):
+                config.inventory = dump_artifact(obj, path, 'hosts')
+            elif os.path.isabs(obj):
+                config.inventory = obj
+            else:
+                config.inventory = os.path.join(path, obj)
+
+    if not config.suppress_env_files:
+        for key in ('envvars', 'extravars', 'passwords', 'settings'):
+            obj = getattr(config, key, None)
+            if obj and not os.path.exists(os.path.join(config.private_data_dir, 'env', key)):
+                path = os.path.join(config.private_data_dir, 'env')
+                dump_artifact(json.dumps(obj), path, key)
+
+        for key in ('ssh_key', 'cmdline'):
+            obj = getattr(config, key, None)
+            if obj and not os.path.exists(os.path.join(config.private_data_dir, 'env', key)):
+                path = os.path.join(config.private_data_dir, 'env')
+                dump_artifact(obj, path, key)
+
+
+def dump_artifact(obj: str,
+                  path: str,
+                  filename: str | None = None
+                  ) -> str:
+    """Write the artifact to disk at the specified path
+
+    :param str obj: The string object to be dumped to disk in the specified
+        path. The artifact filename will be automatically created.
+    :param str path: The full path to the artifacts data directory.
+    :param str filename: The name of file to write the artifact to.
+        If the filename is not provided, then one will be generated.
+
+    :return: The full path filename for the artifact that was generated.
+    """
+    if not os.path.exists(path):
+        os.makedirs(path, mode=0o700)
+
+    p_sha1 = hashlib.sha1()
+    p_sha1.update(obj.encode(encoding='UTF-8'))
+
+    if filename is None:
+        _, fn = tempfile.mkstemp(dir=path)
+    else:
+        fn = os.path.join(path, filename)
+
+    if os.path.exists(fn):
+        c_sha1 = hashlib.sha1()
+        with open(fn) as f:
+            contents = f.read()
+        c_sha1.update(contents.encode(encoding='UTF-8'))
+
+    if not os.path.exists(fn) or p_sha1.hexdigest() != c_sha1.hexdigest():
+        lock_fp = os.path.join(path, '.artifact_write_lock')
+        lock_fd = os.open(lock_fp, os.O_RDWR | os.O_CREAT, stat.S_IRUSR | stat.S_IWUSR)
+        fcntl.lockf(lock_fd, fcntl.LOCK_EX)
+
+        try:
+            with open(fn, 'w') as f:
+                os.chmod(fn, stat.S_IRUSR | stat.S_IWUSR)
+                f.write(str(obj))
+        finally:
+            fcntl.lockf(lock_fd, fcntl.LOCK_UN)
+            os.close(lock_fd)
+            os.remove(lock_fp)
+
+    return fn
diff --git a/src/ansible_runner/config/_base.py b/src/ansible_runner/config/_base.py
index d704a3cd4..6696aa5f0 100644
--- a/src/ansible_runner/config/_base.py
+++ b/src/ansible_runner/config/_base.py
@@ -32,7 +32,7 @@
 from dataclasses import dataclass, field
 from enum import Enum
 from uuid import uuid4
-from collections.abc import Mapping
+from collections.abc import Callable, Mapping
 from typing import Any
 
 import pexpect
@@ -64,6 +64,13 @@ class BaseExecutionMode(Enum):
 
 @dataclass
 class BaseConfig:
+    """The base configuration object.
+
+    This object has multiple initialization responsibilities, including:
+        - guaranteeing the `private_data_dir` directory exists
+        - guaranteeing that `ident` value is set
+        - setting the various work directory attributes based on `private_data_dir`
+    """
 
     # This MUST be the first field we define to handle the use case where a RunnerConfig object
     # is instantiated with private_data_dir as the first positional argument (non-keyword).
@@ -97,6 +104,12 @@ class BaseConfig:
     suppress_env_files: bool = field(metadata={}, default=False)
     timeout: int | None = field(metadata={}, default=None)
 
+    event_handler: Callable[[dict], None] | None = None
+    status_handler: Callable[[dict, BaseConfig], bool] | None = None
+    artifacts_handler: Callable[[str], None] | None = None
+    cancel_callback: Callable[[], bool] | None = None
+    finished_callback: Callable[[BaseConfig], None] | None = None
+
     _CONTAINER_ENGINES = ('docker', 'podman')
 
     def __post_init__(self) -> None:
@@ -115,7 +128,10 @@ def __post_init__(self) -> None:
             # Note that os.makedirs, exist_ok=True is dangerous.  If there's a directory writable
             # by someone other than the user anywhere in the path to be created, an attacker can
             # attempt to compromise the directories via a race.
-            os.makedirs(self.private_data_dir, exist_ok=True, mode=0o700)
+            try:
+                os.makedirs(self.private_data_dir, exist_ok=True, mode=0o700)
+            except Exception as error:
+                raise ConfigurationError(f"Unable to create private_data_dir {self.private_data_dir}") from error
         else:
             self.private_data_dir = tempfile.mkdtemp(prefix=defaults.AUTO_CREATE_NAMING, dir=defaults.AUTO_CREATE_DIR)
 
diff --git a/src/ansible_runner/config/runner.py b/src/ansible_runner/config/runner.py
index 44f73a7de..d4c1bcd30 100644
--- a/src/ansible_runner/config/runner.py
+++ b/src/ansible_runner/config/runner.py
@@ -79,12 +79,15 @@ class RunnerConfig(BaseConfig):
     module_args: str | None = field(metadata={}, default=None)
     omit_event_data: bool = field(metadata={}, default=False)
     only_failed_event_data: bool = field(metadata={}, default=False)
-    playbook: str | None = field(metadata={}, default=None)
+    playbook: str | dict | list | None = field(metadata={}, default=None)
     process_isolation_hide_paths: str | list | None = field(metadata={}, default=None)
     process_isolation_ro_paths: str | list | None = field(metadata={}, default=None)
     process_isolation_show_paths: str | list | None = field(metadata={}, default=None)
     process_isolation_path: str | None = field(metadata={}, default=None)
+    role: str = ""
+    role_skip_facts: bool = False
     roles_path: str | None = field(metadata={}, default=None)
+    role_vars: dict[str, str] | None = None
     skip_tags: str | None = field(metadata={}, default=None)
     suppress_ansible_output: bool = field(metadata={}, default=False)
     suppress_output_file: bool = field(metadata={}, default=False)
@@ -121,6 +124,21 @@ def directory_isolation_path(self):
     def directory_isolation_path(self, value):
         self.directory_isolation_base_path = value
 
+    @property
+    def hosts(self):
+        """
+        Alias for backward compatibility.
+
+        dump_artifacts() makes reference to 'hosts' kwargs (API) value, even though it
+        is undocumented as an API parameter to interface.run(). We make it equivalent
+        to 'host_pattern' here to not break anyone.
+        """
+        return self.host_pattern
+
+    @hosts.setter
+    def hosts(self, value):
+        self.host_pattern = value
+
     @property
     def extra_vars(self):
         """ Alias for backward compatibility. """
diff --git a/src/ansible_runner/interface.py b/src/ansible_runner/interface.py
index 0b3dbc627..a7cc0ebc6 100644
--- a/src/ansible_runner/interface.py
+++ b/src/ansible_runner/interface.py
@@ -17,13 +17,18 @@
 # specific language governing permissions and limitations
 # under the License.
 #
+from __future__ import annotations
+
+import io
 import os
 import json
 import sys
 import threading
 import logging
+from dataclasses import asdict
 
 from ansible_runner import output
+from ansible_runner._internal._dump_artifacts import dump_artifacts
 from ansible_runner.config.runner import RunnerConfig
 from ansible_runner.config.command import CommandConfig
 from ansible_runner.config.inventory import InventoryConfig
@@ -32,7 +37,6 @@
 from ansible_runner.runner import Runner
 from ansible_runner.streaming import Transmitter, Worker, Processor
 from ansible_runner.utils import (
-    dump_artifacts,
     check_isolation_executable_installed,
     sanitize_json_response,
     signal_handler,
@@ -41,7 +45,12 @@
 logging.getLogger('ansible-runner').addHandler(logging.NullHandler())
 
 
-def init_runner(**kwargs):
+def init_runner(
+        config: RunnerConfig,
+        streamer: str,
+        only_transmit_kwargs: bool,
+        _input: io.FileIO | None = None,
+        _output: io.FileIO | None = None):
     '''
     Initialize the Runner() instance
 
@@ -51,89 +60,74 @@ def init_runner(**kwargs):
     See parameters given to :py:func:`ansible_runner.interface.run`
     '''
 
-    # Handle logging first thing
-    debug = kwargs.pop('debug', None)
-    logfile = kwargs.pop('logfile', None)
-
-    if not kwargs.pop("ignore_logging", True):
-        output.configure()
-        if debug in (True, False):
-            output.set_debug('enable' if debug is True else 'disable')
-
-        if logfile:
-            output.set_logfile(logfile)
-
     # If running via the transmit-worker-process method, we must only extract things as read-only
     # inside of one of these commands. That could be either transmit or worker.
-    if kwargs.get('streamer') not in ('worker', 'process'):
-        dump_artifacts(kwargs)
+    if streamer not in ('worker', 'process'):
+        dump_artifacts(config)
 
-    if kwargs.get('streamer'):
+    if streamer:
         # undo any full paths that were dumped by dump_artifacts above in the streamer case
-        private_data_dir = kwargs['private_data_dir']
+        private_data_dir = config.private_data_dir
         project_dir = os.path.join(private_data_dir, 'project')
 
-        playbook_path = kwargs.get('playbook') or ''
+        playbook_path = config.playbook or ''
         if os.path.isabs(playbook_path) and playbook_path.startswith(project_dir):
-            kwargs['playbook'] = os.path.relpath(playbook_path, project_dir)
+            config.playbook = os.path.relpath(playbook_path, project_dir)
 
-        inventory_path = kwargs.get('inventory') or ''
+        inventory_path = config.inventory or ''
         if os.path.isabs(inventory_path) and inventory_path.startswith(private_data_dir):
-            kwargs['inventory'] = os.path.relpath(inventory_path, private_data_dir)
+            config.inventory = os.path.relpath(inventory_path, private_data_dir)
 
-        roles_path = kwargs.get('envvars', {}).get('ANSIBLE_ROLES_PATH') or ''
+        envvars = config.envvars or {}
+        roles_path = envvars.get('ANSIBLE_ROLES_PATH') or ''
         if os.path.isabs(roles_path) and roles_path.startswith(private_data_dir):
-            kwargs['envvars']['ANSIBLE_ROLES_PATH'] = os.path.relpath(roles_path, private_data_dir)
+            config.envvars['ANSIBLE_ROLES_PATH'] = os.path.relpath(roles_path, private_data_dir)
 
-    event_callback_handler = kwargs.pop('event_handler', None)
-    status_callback_handler = kwargs.pop('status_handler', None)
-    artifacts_handler = kwargs.pop('artifacts_handler', None)
-    cancel_callback = kwargs.pop('cancel_callback', None)
-    if cancel_callback is None:
+    if config.cancel_callback is None:
         # attempt to load signal handler.
         # will return None if we are not in the main thread
-        cancel_callback = signal_handler()
-    finished_callback = kwargs.pop('finished_callback', None)
+        config.cancel_callback = signal_handler()
 
-    streamer = kwargs.pop('streamer', None)
-    if streamer:
-        if streamer == 'transmit':
-            stream_transmitter = Transmitter(**kwargs)
-            return stream_transmitter
-
-        if streamer == 'worker':
-            stream_worker = Worker(**kwargs)
-            return stream_worker
-
-        if streamer == 'process':
-            stream_processor = Processor(event_handler=event_callback_handler,
-                                         status_handler=status_callback_handler,
-                                         artifacts_handler=artifacts_handler,
-                                         cancel_callback=cancel_callback,
-                                         finished_callback=finished_callback,
-                                         **kwargs)
-            return stream_processor
-
-    if kwargs.get("process_isolation", False):
-        pi_executable = kwargs.get("process_isolation_executable", "podman")
-        if not check_isolation_executable_installed(pi_executable):
-            print(f'Unable to find process isolation executable: {pi_executable}')
-            sys.exit(1)
+    if streamer == 'transmit':
+        kwargs = asdict(config)
+        stream_transmitter = Transmitter(only_transmit_kwargs, _output=_output, **kwargs)
+        return stream_transmitter
 
-    kwargs.pop('_input', None)
-    kwargs.pop('_output', None)
-    rc = RunnerConfig(**kwargs)
-    rc.prepare()
+    if streamer == 'worker':
+        kwargs = asdict(config)
+        stream_worker = Worker(_input=_input, _output=_output, **kwargs)
+        return stream_worker
 
-    return Runner(rc,
-                  event_handler=event_callback_handler,
-                  status_handler=status_callback_handler,
-                  artifacts_handler=artifacts_handler,
-                  cancel_callback=cancel_callback,
-                  finished_callback=finished_callback)
+    if streamer == 'process':
+        kwargs = asdict(config)
+        stream_processor = Processor(_input=_input, **kwargs)
+        return stream_processor
 
+    if config.process_isolation:
+        pi_executable = config.process_isolation_executable
+        if not check_isolation_executable_installed(pi_executable):
+            print(f'Unable to find process isolation executable: {pi_executable}')
+            sys.exit(1)
 
-def run(**kwargs):
+    config.prepare()
+
+    return Runner(config,
+                  event_handler=config.event_handler,
+                  status_handler=config.status_handler,
+                  artifacts_handler=config.artifacts_handler,
+                  cancel_callback=config.cancel_callback,
+                  finished_callback=config.finished_callback)
+
+
+def run(config: RunnerConfig | None = None,
+        streamer: str = "",
+        debug: bool = False,
+        logfile: str = "",
+        ignore_logging: bool = True,
+        _input: io.FileIO | None = None,
+        _output: io.FileIO | None = None,
+        only_transmit_kwargs: bool = False,
+        **kwargs):
     '''
     Run an Ansible Runner task in the foreground and return a Runner object when complete.
 
@@ -209,7 +203,19 @@ def run(**kwargs):
 
     :returns: A :py:class:`ansible_runner.runner.Runner` object, or a simple object containing ``rc`` if run remotely
     '''
-    r = init_runner(**kwargs)
+
+    # Initialize logging
+    if not ignore_logging:
+        output.configure(debug, logfile)
+
+    if not config:
+        config = RunnerConfig(**kwargs)
+
+    r = init_runner(
+        config=config, streamer=streamer,
+        only_transmit_kwargs=only_transmit_kwargs,
+        _input=_input, _output=_output,
+    )
     r.run()
     return r
 
diff --git a/src/ansible_runner/output.py b/src/ansible_runner/output.py
index e3e15d3e6..955abc923 100644
--- a/src/ansible_runner/output.py
+++ b/src/ansible_runner/output.py
@@ -64,7 +64,7 @@ def set_traceback(value: str) -> None:
     TRACEBACK_ENABLED = value.lower() == 'enable'
 
 
-def configure() -> None:
+def configure(debug: bool, logfile: str) -> None:
     '''
     Configures the logging facility
 
@@ -89,3 +89,7 @@ def configure() -> None:
         formatter = logging.Formatter('%(message)s')
         stdout_handler.setFormatter(formatter)
         _display_logger.addHandler(stdout_handler)
+
+    set_debug('enable' if debug is True else 'disable')
+    if logfile:
+        set_logfile(logfile)
diff --git a/src/ansible_runner/streaming.py b/src/ansible_runner/streaming.py
index bf205d22b..041f3b4ba 100644
--- a/src/ansible_runner/streaming.py
+++ b/src/ansible_runner/streaming.py
@@ -1,6 +1,7 @@
 from __future__ import annotations  # allow newer type syntax until 3.10 is our minimum
 
 import codecs
+import io
 import json
 import os
 import stat
@@ -14,6 +15,7 @@
 from threading import Event, RLock, Thread
 
 import ansible_runner
+from ansible_runner.config.runner import RunnerConfig
 from ansible_runner.exceptions import ConfigurationError
 from ansible_runner.loader import ArtifactLoader
 import ansible_runner.plugins
@@ -37,12 +39,12 @@ def __init__(self, settings):
 
 
 class Transmitter:
-    def __init__(self, _output=None, **kwargs):
+    def __init__(self, only_transmit_kwargs: bool, _output: io.FileIO | None, **kwargs):
         if _output is None:
             _output = sys.stdout.buffer
         self._output = _output
-        self.private_data_dir = os.path.abspath(kwargs.pop('private_data_dir'))
-        self.only_transmit_kwargs = kwargs.pop('only_transmit_kwargs', False)
+        self.private_data_dir = os.path.abspath(kwargs['private_data_dir'])
+        self.only_transmit_kwargs = only_transmit_kwargs
         if 'keepalive_seconds' in kwargs:
             kwargs.pop('keepalive_seconds')  # don't confuse older runners with this Worker-only arg
 
diff --git a/src/ansible_runner/utils/__init__.py b/src/ansible_runner/utils/__init__.py
index f4a661ee5..85ff60da8 100644
--- a/src/ansible_runner/utils/__init__.py
+++ b/src/ansible_runner/utils/__init__.py
@@ -6,10 +6,7 @@
 import re
 import os
 import stat
-import fcntl
 import shutil
-import hashlib
-import tempfile
 import subprocess
 import base64
 import threading
@@ -120,55 +117,6 @@ def check_isolation_executable_installed(isolation_executable: str) -> bool:
     return False
 
 
-def dump_artifact(obj: str,
-                  path: str,
-                  filename: str | None = None
-                  ) -> str:
-    '''
-    Write the artifact to disk at the specified path
-
-    :param str obj: The string object to be dumped to disk in the specified
-        path. The artifact filename will be automatically created.
-    :param str path: The full path to the artifacts data directory.
-    :param str filename: The name of file to write the artifact to.
-        If the filename is not provided, then one will be generated.
-
-    :return: The full path filename for the artifact that was generated.
-    '''
-    if not os.path.exists(path):
-        os.makedirs(path, mode=0o700)
-
-    p_sha1 = hashlib.sha1()
-    p_sha1.update(obj.encode(encoding='UTF-8'))
-
-    if filename is None:
-        _, fn = tempfile.mkstemp(dir=path)
-    else:
-        fn = os.path.join(path, filename)
-
-    if os.path.exists(fn):
-        c_sha1 = hashlib.sha1()
-        with open(fn) as f:
-            contents = f.read()
-        c_sha1.update(contents.encode(encoding='UTF-8'))
-
-    if not os.path.exists(fn) or p_sha1.hexdigest() != c_sha1.hexdigest():
-        lock_fp = os.path.join(path, '.artifact_write_lock')
-        lock_fd = os.open(lock_fp, os.O_RDWR | os.O_CREAT, stat.S_IRUSR | stat.S_IWUSR)
-        fcntl.lockf(lock_fd, fcntl.LOCK_EX)
-
-        try:
-            with open(fn, 'w') as f:
-                os.chmod(fn, stat.S_IRUSR | stat.S_IWUSR)
-                f.write(str(obj))
-        finally:
-            fcntl.lockf(lock_fd, fcntl.LOCK_UN)
-            os.close(lock_fd)
-            os.remove(lock_fp)
-
-    return fn
-
-
 def cleanup_artifact_dir(path: str, num_keep: int = 0) -> None:
     # 0 disables artifact dir cleanup/rotation
     if num_keep < 1:
@@ -182,80 +130,6 @@ def cleanup_artifact_dir(path: str, num_keep: int = 0) -> None:
         shutil.rmtree(all_paths[f])
 
 
-def dump_artifacts(kwargs: dict) -> None:
-    '''
-    Introspect the kwargs and dump objects to disk
-    '''
-    private_data_dir = kwargs.get('private_data_dir')
-    if not private_data_dir:
-        private_data_dir = tempfile.mkdtemp()
-        kwargs['private_data_dir'] = private_data_dir
-
-    if not os.path.exists(private_data_dir):
-        raise ValueError('private_data_dir path is either invalid or does not exist')
-
-    if 'role' in kwargs:
-        role = {'name': kwargs.pop('role')}
-        if 'role_vars' in kwargs:
-            role['vars'] = kwargs.pop('role_vars')
-
-        play = [{'hosts': kwargs.pop('hosts', 'all'), 'roles': [role]}]
-
-        if kwargs.pop('role_skip_facts', False):
-            play[0]['gather_facts'] = False
-
-        kwargs['playbook'] = play
-
-        if 'envvars' not in kwargs:
-            kwargs['envvars'] = {}
-
-        roles_path = kwargs.pop('roles_path', None)
-        if not roles_path:
-            roles_path = os.path.join(private_data_dir, 'roles')
-        else:
-            roles_path += f":{os.path.join(private_data_dir, 'roles')}"
-
-        kwargs['envvars']['ANSIBLE_ROLES_PATH'] = roles_path
-
-    playbook = kwargs.get('playbook')
-    if playbook:
-        # Ensure the play is a list of dictionaries
-        if isinstance(playbook, MutableMapping):
-            playbook = [playbook]
-
-        if isplaybook(playbook):
-            path = os.path.join(private_data_dir, 'project')
-            kwargs['playbook'] = dump_artifact(json.dumps(playbook), path, 'main.json')
-
-    obj = kwargs.get('inventory')
-    if obj and isinventory(obj):
-        path = os.path.join(private_data_dir, 'inventory')
-        if isinstance(obj, MutableMapping):
-            kwargs['inventory'] = dump_artifact(json.dumps(obj), path, 'hosts.json')
-        elif isinstance(obj, str):
-            if not os.path.exists(os.path.join(path, obj)):
-                kwargs['inventory'] = dump_artifact(obj, path, 'hosts')
-            elif os.path.isabs(obj):
-                kwargs['inventory'] = obj
-            else:
-                kwargs['inventory'] = os.path.join(path, obj)
-
-    if not kwargs.get('suppress_env_files', False):
-        for key in ('envvars', 'extravars', 'passwords', 'settings'):
-            obj = kwargs.get(key)
-            if obj and not os.path.exists(os.path.join(private_data_dir, 'env', key)):
-                path = os.path.join(private_data_dir, 'env')
-                dump_artifact(json.dumps(obj), path, key)
-                kwargs.pop(key)
-
-        for key in ('ssh_key', 'cmdline'):
-            obj = kwargs.get(key)
-            if obj and not os.path.exists(os.path.join(private_data_dir, 'env', key)):
-                path = os.path.join(private_data_dir, 'env')
-                dump_artifact(str(kwargs[key]), path, key)
-                kwargs.pop(key)
-
-
 def collect_new_events(event_path: str, old_events: dict) -> Iterator[tuple[dict, dict]]:
     '''
     Collect new events for the 'events' generator property
diff --git a/test/unit/config/test__base.py b/test/unit/config/test__base.py
index 9c33e423c..66569db5c 100644
--- a/test/unit/config/test__base.py
+++ b/test/unit/config/test__base.py
@@ -2,6 +2,7 @@
 
 import os
 import re
+import tempfile
 from functools import partial
 
 from test.utils.common import RSAKey
@@ -22,6 +23,19 @@ def load_file_side_effect(path, value, *args, **kwargs):
     raise ConfigurationError
 
 
+def test_base_config_empty_pvt_data_dir():
+    """Make sure we create a pvt data dir even when not supplied"""
+    rc = BaseConfig()
+    tmpdir = tempfile.gettempdir()
+    assert tmpdir in rc.private_data_dir
+
+
+def test_base_config_invalid_pvt_data_dir():
+    """A ConfigurationError should be raised if we cannot create the requested pvt data dir"""
+    with pytest.raises(ConfigurationError, match="Unable to create private_data_dir"):
+        BaseConfig("/not/a/writable/path")
+
+
 def test_base_config_init_defaults(tmp_path):
     rc = BaseConfig(private_data_dir=tmp_path.as_posix())
     assert rc.private_data_dir == tmp_path.as_posix()
diff --git a/test/unit/utils/test_dump_artifacts.py b/test/unit/utils/test_dump_artifacts.py
index ef5cfb400..8b8c30565 100644
--- a/test/unit/utils/test_dump_artifacts.py
+++ b/test/unit/utils/test_dump_artifacts.py
@@ -1,27 +1,7 @@
 import pytest
 
-from ansible_runner.utils import dump_artifacts
-
-
-def test_dump_artifacts_private_data_dir_does_not_exists():
-    data_dir = '/not/a/path'
-    kwargs = {'private_data_dir': data_dir}
-
-    with pytest.raises(ValueError, match='invalid or does not exist'):
-        dump_artifacts(kwargs)
-
-    assert kwargs['private_data_dir'] == data_dir
-
-
-def test_dump_artifacts_private_data_dir_create_tempfile(mocker):
-    mocker.patch('ansible_runner.utils.os.path.exists', side_effect=AttributeError('Raised intentionally'))
-    mocker.patch('ansible_runner.utils.tempfile.mkdtemp', return_value='/tmp/dir')
-
-    kwargs = {}
-    with pytest.raises(AttributeError, match='Raised intentionally'):
-        dump_artifacts(kwargs)
-
-    assert kwargs['private_data_dir'] == '/tmp/dir'
+from ansible_runner.config.runner import RunnerConfig
+from ansible_runner._internal._dump_artifacts import dump_artifacts
 
 
 @pytest.mark.parametrize(
@@ -31,20 +11,20 @@ def test_dump_artifacts_private_data_dir_create_tempfile(mocker):
     )
 )
 def test_dump_artifacts_playbook_object(mocker, playbook):
-    mock_dump_artifact = mocker.patch('ansible_runner.utils.dump_artifact', side_effect=AttributeError('Raised intentionally'))
+    mock_dump_artifact = mocker.patch('ansible_runner._internal._dump_artifacts.dump_artifact', side_effect=AttributeError('Raised intentionally'))
     mocker.patch('ansible_runner.utils.isplaybook', return_value=True)
 
     playbook_string = '[{"playbook": [{"hosts": "all"}]}]'
     kwargs = {'private_data_dir': '/tmp', 'playbook': playbook}
 
     with pytest.raises(AttributeError, match='Raised intentionally'):
-        dump_artifacts(kwargs)
+        dump_artifacts(RunnerConfig(**kwargs))
 
     mock_dump_artifact.assert_called_once_with(playbook_string, '/tmp/project', 'main.json')
 
 
 def test_dump_artifacts_role(mocker):
-    mock_dump_artifact = mocker.patch('ansible_runner.utils.dump_artifact')
+    mock_dump_artifact = mocker.patch('ansible_runner._internal._dump_artifacts.dump_artifact')
 
     kwargs = {
         'private_data_dir': '/tmp',
@@ -52,14 +32,14 @@ def test_dump_artifacts_role(mocker):
         'playbook': [{'playbook': [{'hosts': 'all'}]}],
     }
 
-    dump_artifacts(kwargs)
+    dump_artifacts(RunnerConfig(**kwargs))
 
     assert mock_dump_artifact.call_count == 2
     mock_dump_artifact.assert_called_with('{"ANSIBLE_ROLES_PATH": "/tmp/roles"}', '/tmp/env', 'envvars')
 
 
 def test_dump_artifacts_roles_path(mocker):
-    mock_dump_artifact = mocker.patch('ansible_runner.utils.dump_artifact')
+    mock_dump_artifact = mocker.patch('ansible_runner._internal._dump_artifacts.dump_artifact')
 
     kwargs = {
         'private_data_dir': '/tmp',
@@ -68,14 +48,14 @@ def test_dump_artifacts_roles_path(mocker):
         'playbook': [{'playbook': [{'hosts': 'all'}]}],
     }
 
-    dump_artifacts(kwargs)
+    dump_artifacts(RunnerConfig(**kwargs))
 
     assert mock_dump_artifact.call_count == 2
     mock_dump_artifact.assert_called_with('{"ANSIBLE_ROLES_PATH": "/tmp/altrole:/tmp/roles"}', '/tmp/env', 'envvars')
 
 
 def test_dump_artifacts_role_vars(mocker):
-    mock_dump_artifact = mocker.patch('ansible_runner.utils.dump_artifact', side_effect=AttributeError('Raised intentionally'))
+    mock_dump_artifact = mocker.patch('ansible_runner._internal._dump_artifacts.dump_artifact', side_effect=AttributeError('Raised intentionally'))
 
     kwargs = {
         'private_data_dir': '/tmp',
@@ -85,7 +65,7 @@ def test_dump_artifacts_role_vars(mocker):
     }
 
     with pytest.raises(AttributeError, match='Raised intentionally'):
-        dump_artifacts(kwargs)
+        dump_artifacts(RunnerConfig(**kwargs))
 
     mock_dump_artifact.assert_called_once_with(
         '[{"hosts": "all", "roles": [{"name": "test", "vars": {"name": "nginx"}}]}]',
@@ -95,7 +75,7 @@ def test_dump_artifacts_role_vars(mocker):
 
 
 def test_dump_artifacts_role_skip_facts(mocker):
-    mock_dump_artifact = mocker.patch('ansible_runner.utils.dump_artifact', side_effect=AttributeError('Raised intentionally'))
+    mock_dump_artifact = mocker.patch('ansible_runner._internal._dump_artifacts.dump_artifact', side_effect=AttributeError('Raised intentionally'))
 
     kwargs = {
         'private_data_dir': '/tmp',
@@ -105,7 +85,7 @@ def test_dump_artifacts_role_skip_facts(mocker):
     }
 
     with pytest.raises(AttributeError, match='Raised intentionally'):
-        dump_artifacts(kwargs)
+        dump_artifacts(RunnerConfig(**kwargs))
 
     mock_dump_artifact.assert_called_once_with(
         '[{"hosts": "all", "roles": [{"name": "test"}], "gather_facts": false}]',
@@ -115,21 +95,21 @@ def test_dump_artifacts_role_skip_facts(mocker):
 
 
 def test_dump_artifacts_inventory_string(mocker):
-    mock_dump_artifact = mocker.patch('ansible_runner.utils.dump_artifact')
+    mock_dump_artifact = mocker.patch('ansible_runner._internal._dump_artifacts.dump_artifact')
 
     inv = '[all]\nlocalhost'
     kwargs = {'private_data_dir': '/tmp', 'inventory': inv}
-    dump_artifacts(kwargs)
+    dump_artifacts(RunnerConfig(**kwargs))
 
     mock_dump_artifact.assert_called_once_with(inv, '/tmp/inventory', 'hosts')
 
 
 def test_dump_artifacts_inventory_path(mocker):
-    mock_dump_artifact = mocker.patch('ansible_runner.utils.dump_artifact')
+    mock_dump_artifact = mocker.patch('ansible_runner._internal._dump_artifacts.dump_artifact')
 
     inv = '/tmp'
     kwargs = {'private_data_dir': '/tmp', 'inventory': inv}
-    dump_artifacts(kwargs)
+    dump_artifacts(RunnerConfig(**kwargs))
 
     assert mock_dump_artifact.call_count == 0
     assert mock_dump_artifact.called is False
@@ -137,12 +117,12 @@ def test_dump_artifacts_inventory_path(mocker):
 
 
 def test_dump_artifacts_inventory_object(mocker):
-    mock_dump_artifact = mocker.patch('ansible_runner.utils.dump_artifact')
+    mock_dump_artifact = mocker.patch('ansible_runner._internal._dump_artifacts.dump_artifact')
 
     inv = {'foo': 'bar'}
     inv_string = '{"foo": "bar"}'
     kwargs = {'private_data_dir': '/tmp', 'inventory': inv}
-    dump_artifacts(kwargs)
+    dump_artifacts(RunnerConfig(**kwargs))
 
     mock_dump_artifact.assert_called_once_with(inv_string, '/tmp/inventory', 'hosts.json')
 
@@ -152,9 +132,10 @@ def test_dump_artifacts_inventory_string_path(mocker):
 
     inv_string = 'site1'
     kwargs = {'private_data_dir': '/tmp', 'inventory': inv_string}
-    dump_artifacts(kwargs)
+    rc = RunnerConfig(**kwargs)
+    dump_artifacts(rc)
 
-    assert kwargs['inventory'] == '/tmp/inventory/site1'
+    assert rc.inventory == '/tmp/inventory/site1'
 
 
 def test_dump_artifacts_inventory_string_abs_path(mocker):
@@ -162,13 +143,14 @@ def test_dump_artifacts_inventory_string_abs_path(mocker):
 
     inv_string = '/tmp/site1'
     kwargs = {'private_data_dir': '/tmp', 'inventory': inv_string}
-    dump_artifacts(kwargs)
+    rc = RunnerConfig(**kwargs)
+    dump_artifacts(rc)
 
-    assert kwargs['inventory'] == '/tmp/site1'
+    assert rc.inventory == '/tmp/site1'
 
 
 def test_dump_artifacts_passwords(mocker):
-    mock_dump_artifact = mocker.patch('ansible_runner.utils.dump_artifact')
+    mock_dump_artifact = mocker.patch('ansible_runner._internal._dump_artifacts.dump_artifact')
 
     kwargs = {
         'private_data_dir': '/tmp',
@@ -177,7 +159,7 @@ def test_dump_artifacts_passwords(mocker):
         'ssh_key': 'asdfg1234',
     }
 
-    dump_artifacts(kwargs)
+    dump_artifacts(RunnerConfig(**kwargs))
 
     assert mock_dump_artifact.call_count == 3
     mock_dump_artifact.assert_any_call('{"a": "b"}', '/tmp/env', 'passwords')
@@ -186,7 +168,7 @@ def test_dump_artifacts_passwords(mocker):
 
 
 def test_dont_dump_artifacts_passwords(mocker):
-    mock_dump_artifact = mocker.patch('ansible_runner.utils.dump_artifact')
+    mock_dump_artifact = mocker.patch('ansible_runner._internal._dump_artifacts.dump_artifact')
 
     kwargs = {
         'private_data_dir': '/tmp',
@@ -196,7 +178,7 @@ def test_dont_dump_artifacts_passwords(mocker):
         'suppress_env_files': True
     }
 
-    dump_artifacts(kwargs)
+    dump_artifacts(RunnerConfig(**kwargs))
 
     assert mock_dump_artifact.call_count == 0
 
@@ -211,12 +193,12 @@ def test_dont_dump_artifacts_passwords(mocker):
     )
 )
 def test_dump_artifacts_extra_keys(mocker, key, value, value_str):
-    mock_dump_artifact = mocker.patch('ansible_runner.utils.dump_artifact')
+    mock_dump_artifact = mocker.patch('ansible_runner._internal._dump_artifacts.dump_artifact')
 
     kwargs = {'private_data_dir': '/tmp'}
     kwargs.update({key: value})
 
-    dump_artifacts(kwargs)
+    rc = RunnerConfig(**kwargs)
+    dump_artifacts(rc)
 
     mock_dump_artifact.assert_called_once_with(value_str, '/tmp/env', key)
-    assert 'settings' not in kwargs