diff --git a/src/ansible_runner/interface.py b/src/ansible_runner/interface.py
index 0b3dbc62..1d77df7a 100644
--- a/src/ansible_runner/interface.py
+++ b/src/ansible_runner/interface.py
@@ -133,7 +133,35 @@ def init_runner(**kwargs):
                   finished_callback=finished_callback)
 
 
-def run(**kwargs):
+def run(*,
+        private_data_dir,
+        ident,
+        json_mode,
+        playbook,
+        module, module_args,
+        host_pattern, inventory,
+        role, roles_path,
+        envvars, extravars,
+        passwords,
+        settings,
+        ssh_key,
+        cmdline,
+        suppress_env_files,
+        limit, forks, verbosity, quiet,
+        artifact_dir, project_dir,
+        rotate_artifacts,
+        timeout,
+        streamer,
+        _input, _output,
+        event_handler, status_handler, artifacts_handler,
+        cancel_callback, finished_callback,
+        process_isolation, process_isolation_executable, process_isolation_path,
+        process_isolation_hide_paths, process_isolation_show_paths, process_isolation_ro_paths,
+        container_image, container_volume_mounts, container_options,
+        directory_isolation_base_path,
+        fact_cache, fact_cache_type,
+        omit_event_data, only_failed_event_data, check_job_event_data,
+        ):
     '''
     Run an Ansible Runner task in the foreground and return a Runner object when complete.
 
diff --git a/src/ansible_runner/streaming.py b/src/ansible_runner/streaming.py
index bf205d22..ea9a872d 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 inspect
 import json
 import os
 import stat
@@ -188,6 +189,17 @@ def run(self):
                     return self.status, self.rc
 
                 if 'kwargs' in data:
+                    spec = inspect.getfullargspec(ansible_runner.interface.run)
+                    diff = set(data['kwargs']).difference(set(spec.kwonlyargs))
+                    if diff:
+                        self.status_handler(
+                            {
+                                'status': 'error',
+                                'job_explanation': f'Unhandled keyword argument(s) in transmitted data: {diff}'
+                            },
+                            None)
+                        self.finished_callback(None)  # send eof line
+                        return self.status, self.rc
                     self.job_kwargs = self.update_paths(data['kwargs'])
                 elif 'zipfile' in data:
                     try:
diff --git a/test/unit/test_streaming.py b/test/unit/test_streaming.py
index c185cf0e..8b98149c 100644
--- a/test/unit/test_streaming.py
+++ b/test/unit/test_streaming.py
@@ -1,6 +1,7 @@
+import io
 import os
 
-from ansible_runner.streaming import Processor
+from ansible_runner.streaming import Processor, Transmitter, Worker
 
 
 class TestProcessor:
@@ -14,3 +15,66 @@ def test_artifact_dir_with_int_ident(self, tmp_path):
         assert p.artifact_dir == os.path.join(kwargs['private_data_dir'],
                                               'artifacts',
                                               str(kwargs['ident']))
+
+class TestTransmitter:
+
+    def test_job_arguments(self, tmp_path, project_fixtures):
+        """
+        Test format of sending job arguments.
+        """
+        transmit_dir = project_fixtures / 'debug'
+        outgoing_buffer_file = tmp_path / 'buffer_out'
+        outgoing_buffer_file.touch()
+
+        kwargs = {
+            'playbook': 'debug.yml',
+            'only_transmit_kwargs': True
+        }
+
+        with outgoing_buffer_file.open('b+r') as outgoing_buffer:
+            transmitter = Transmitter(
+                _output=outgoing_buffer,
+                private_data_dir=transmit_dir,
+                **kwargs)
+            transmitter.run()
+            outgoing_buffer.seek(0)
+            sent = outgoing_buffer.read()
+
+        expected = b'{"kwargs": {"playbook": "debug.yml"}}\n{"eof": true}\n'
+        assert sent == expected
+
+    def test_unhandled_argument(self, project_fixtures):
+        transmit_dir = project_fixtures / 'debug'
+        transmit_buffer = io.BytesIO()
+        output_buffer = io.BytesIO()
+
+        for buffer in (transmit_buffer, output_buffer):
+            buffer.name = 'foo'
+
+        kwargs = {
+            'playbook': 'debug.yml',
+            'oopsie': True,
+            'only_transmit_kwargs': True
+        }
+
+        status, rc = Transmitter(
+                _output=transmit_buffer,
+                private_data_dir=transmit_dir,
+                **kwargs).run()
+
+        assert rc in (None, 0)
+        assert status == 'unstarted'
+        transmit_buffer.seek(0)
+
+        worker = Worker(_input=transmit_buffer,
+                        _output=output_buffer)
+
+        status, rc = worker.run()
+
+        assert status == 'error'
+        assert rc in (None, 0)
+
+        output_buffer.seek(0)
+        output = output_buffer.read()
+
+        assert output == b'{"status": "error", "job_explanation": "Unhandled keyword argument(s) in transmitted data: {\'oopsie\'}"}\n{"eof": true}\n'