diff --git a/awx/main/tasks/jobs.py b/awx/main/tasks/jobs.py
index 9d8a9dfef8..a10367a076 100644
--- a/awx/main/tasks/jobs.py
+++ b/awx/main/tasks/jobs.py
@@ -810,9 +810,9 @@ class BaseTask(object):
                 if not res:
                     # res is None when quota exceeded or other early-return condition.
                     # Must release work unit here before returning, or it will leak.
-                    if receptor_job and getattr(receptor_job, 'receptor_ctl', None):
+                    if receptor_job and getattr(receptor_job, 'work', None):
                         try:
-                            receptor_job._receptor_release_work(receptor_job.receptor_ctl, 'error')
+                            receptor_job._receptor_release_work(receptor_job.work, 'error')
                         except Exception:
                             logger.exception(f'Failed to release work unit {getattr(receptor_job, "unit_id", "unknown")} on early return')
                     return
@@ -857,9 +857,9 @@ class BaseTask(object):
             self.instance = _finalize_job_run(self.model, pk, self.runner_callback, status)
         finally:
             # Guarantee work unit release even if finalization throws
-            if receptor_job and getattr(receptor_job, 'receptor_ctl', None):
+            if receptor_job and getattr(receptor_job, 'work', None):
                 try:
-                    receptor_job._receptor_release_work(receptor_job.receptor_ctl, status)
+                    receptor_job._receptor_release_work(receptor_job.work, status)
                 except Exception:
                     logger.exception(f'Failed to release work unit {getattr(receptor_job, "unit_id", "unknown")}')
 
diff --git a/awx/main/tasks/receptor.py b/awx/main/tasks/receptor.py
index 168e678a8f..460d8d2ba7 100644
--- a/awx/main/tasks/receptor.py
+++ b/awx/main/tasks/receptor.py
@@ -181,6 +181,38 @@ def get_receptor_ctl(config_data=None):
         return ReceptorControl(receptor_sockfile)
 
 
+class ReceptorWork:
+    """The receptor work-unit operations AWX uses after a job is submitted.
+
+    Streaming results, reading status, canceling and releasing go through this class
+    instead of raw receptorctl commands, so a test can stand in a fake work unit
+    (see awx.main.tests.fake_receptor) and drive the real streaming, adoption and
+    finalize code without a receptor mesh.
+    """
+
+    def __init__(self, receptor_ctl):
+        self.receptor_ctl = receptor_ctl
+
+    def list(self):
+        return self.receptor_ctl.simple_command('work list')
+
+    def status(self, unit_id):
+        return self.receptor_ctl.simple_command(f'work status {unit_id}')
+
+    def results(self, unit_id, startpos=None):
+        """Return (socket, file) streaming the unit's ansible-runner output."""
+        kwargs = dict(return_socket=True, return_sockfile=True)
+        if startpos is not None:
+            kwargs['startpos'] = startpos
+        return self.receptor_ctl.get_work_results(unit_id, **kwargs)
+
+    def cancel(self, unit_id):
+        return self.receptor_ctl.simple_command(f'work cancel {unit_id}')
+
+    def release(self, unit_id):
+        return self.receptor_ctl.simple_command(f'work release {unit_id}')
+
+
 def adopt_remote_work(receptor_ctl, node, unit_id, config_data=None):
     """Adopt a running work unit from a remote node, passing TLS/signwork when configured.
 
@@ -446,7 +478,7 @@ class AWXReceptorJob:
         self.receptor_ctl = get_receptor_ctl(self.config_data)
         return self._run_internal(self.receptor_ctl)
 
-    def _receptor_release_work(self, receptor_ctl: ReceptorControl, status: str) -> None:
+    def _receptor_release_work(self, work: ReceptorWork, status: str) -> None:
         if self.unit_id is None:
             return
 
@@ -457,14 +489,15 @@ class AWXReceptorJob:
             return
 
         try:
-            receptor_ctl.simple_command(f"work release {self.unit_id}")
+            work.release(self.unit_id)
             logger.debug(f"Released work unit {self.unit_id}.")
         except Exception:
             logger.exception(f"Error releasing work unit {self.unit_id}.")
 
     def _run_internal(self, receptor_ctl):
+        self.work = ReceptorWork(receptor_ctl)
         self._transmit_phase(receptor_ctl)
-        return self._process_phase(receptor_ctl)
+        return self._process_phase(self.work)
 
     def _transmit_phase(self, receptor_ctl):
         """Submit work to receptor and wait for the transmit thread to finish.
@@ -526,7 +559,7 @@ class AWXReceptorJob:
         if self.work_type != 'local' and os.path.exists(artifact_dir):
             shutil.rmtree(artifact_dir)
 
-    def _process_phase(self, receptor_ctl):
+    def _process_phase(self, work: ReceptorWork):
         """Stream events from the receptor work unit via the ansible-runner process streamer.
 
         Extracted from _run_internal() so it can be called standalone for job adoption
@@ -536,7 +569,7 @@ class AWXReceptorJob:
         resultsock = None
         resultfile = None
         try:
-            resultsock, resultfile = receptor_ctl.get_work_results(self.unit_id, return_socket=True, return_sockfile=True)
+            resultsock, resultfile = work.results(self.unit_id)
         except Exception:
             logger.exception(f'Failed to get work results for unit {self.unit_id}')
             raise
@@ -557,7 +590,7 @@ class AWXReceptorJob:
                     raise SignalExit()
                 res = processor_future.result()
             except SignalExit:
-                receptor_ctl.simple_command(f"work cancel {self.unit_id}")
+                work.cancel(self.unit_id)
                 if resultsock:
                     try:
                         resultsock.shutdown(socket.SHUT_RDWR)
@@ -571,11 +604,11 @@ class AWXReceptorJob:
                 signal_state.raise_exception = False
 
             if res.status == 'error':
-                return self._handle_work_error(receptor_ctl, res)
+                return self._handle_work_error(work, res)
 
         return res
 
-    def _handle_work_error(self, receptor_ctl, res):
+    def _handle_work_error(self, work: ReceptorWork, res):
         """Handle the error path from _process_phase when the work unit status is 'error'."""
         # If ansible-runner ran, but an error occured at runtime, the traceback information
         # is saved via the status_handler passed in to the processor.
@@ -583,7 +616,7 @@ class AWXReceptorJob:
             return res
 
         try:
-            unit_status = receptor_ctl.simple_command(f'work status {self.unit_id}')
+            unit_status = work.status(self.unit_id)
             detail = unit_status.get('Detail') or ''
             state_name = unit_status.get('StateName', None)
             stdout_size = unit_status.get('StdoutSize', 0)
@@ -607,7 +640,7 @@ class AWXReceptorJob:
                 # contain useful information about why the job failed. In case stdout is
                 # massive, only ask for last 1000 bytes
                 startpos = max(stdout_size - 1000, 0)
-                _resultsock, resultfile = receptor_ctl.get_work_results(self.unit_id, startpos=startpos, return_socket=True, return_sockfile=True)
+                _resultsock, resultfile = work.results(self.unit_id, startpos=startpos)
                 lines = resultfile.readlines()
                 receptor_output = b"".join(lines).decode()
                 _resultsock.shutdown(socket.SHUT_RDWR)
@@ -965,7 +998,7 @@ def _compute_adoption_dedup(job):
     return safe_threshold, collision_zone
 
 
-def reattach_to_work_unit(job, receptor_ctl):
+def reattach_to_work_unit(job, work: ReceptorWork):
     """Reconnect to a receptor work unit and stream events in real-time until it completes.
 
     Reconstructs the minimal process-phase context from the DB job record, then calls
@@ -980,7 +1013,7 @@ def reattach_to_work_unit(job, receptor_ctl):
 
     # Check state for logging — no longer a gate. We stream regardless.
     try:
-        unit_status = receptor_ctl.simple_command(f'work status {unit_id}')
+        unit_status = work.status(unit_id)
     except Exception:
         logger.warning(f'Cannot get receptor status for work unit {unit_id} (job {job.id}), deferring adoption')
         return False
@@ -1011,7 +1044,7 @@ def reattach_to_work_unit(job, receptor_ctl):
     process_phase_failed = False
     res = None
     try:
-        res = receptor_job._process_phase(receptor_ctl)  # blocks until unit completes
+        res = receptor_job._process_phase(work)  # blocks until unit completes
     except Exception:
         logger.exception(f'Adoption process phase failed for job {job.id} (unit={unit_id})')
         process_phase_failed = True
@@ -1031,7 +1064,7 @@ def reattach_to_work_unit(job, receptor_ctl):
         else:
             # _process_phase raised — fall back to work status for exit code
             try:
-                unit_status = receptor_ctl.simple_command(f'work status {unit_id}')
+                unit_status = work.status(unit_id)
                 state_name = unit_status.get('StateName', '')
             except Exception:
                 logger.debug(f'Could not get final status for {unit_id} after process phase failure')
@@ -1040,7 +1073,7 @@ def reattach_to_work_unit(job, receptor_ctl):
             exit_code = _get_adoption_exit_code(unit_status, state_name)
             _finalize_adopted_job(job, callback, exit_code, process_phase_failed)
     finally:
-        receptor_job._receptor_release_work(receptor_ctl, getattr(res, 'status', 'error'))
+        receptor_job._receptor_release_work(work, getattr(res, 'status', 'error'))
         shutil.rmtree(private_data_dir, ignore_errors=True)
 
     return True
diff --git a/awx/main/tasks/system.py b/awx/main/tasks/system.py
index 4d9da3c431..3e979a1545 100644
--- a/awx/main/tasks/system.py
+++ b/awx/main/tasks/system.py
@@ -85,6 +85,7 @@ from awx.main.tasks.adoption_decisions import (
 from awx.main.tasks.helpers import is_run_threshold_reached
 from awx.main.tasks.host_indirect import save_indirect_host_entries
 from awx.main.tasks.receptor import (
+    ReceptorWork,
     administrative_workunit_reaper,
     get_receptor_ctl,
     reattach_to_work_unit,
@@ -1025,7 +1026,7 @@ def adopt_job_async(job_id):
 
     receptor_ctl = get_receptor_ctl()
     try:
-        reattach_to_work_unit(job, receptor_ctl)
+        reattach_to_work_unit(job, ReceptorWork(receptor_ctl))
     except Exception:
         logger.exception(f'adopt_job_async: adoption failed for job {job.id} (unit={job.work_unit_id})')
     finally:
@@ -1063,8 +1064,9 @@ def awx_receptor_workunit_reaper():
     except FileNotFoundError:
         logger.info('Receptorctl sockfile not found for workunit reaper, doing nothing')
         return
+    work = ReceptorWork(receptor_ctl)
     try:
-        receptor_work_list = receptor_ctl.simple_command("work list")
+        receptor_work_list = work.list()
     except ValueError as exc:
         logger.info(f'Error getting work list for workunit reaper, error: {str(exc)}')
         return
@@ -1075,8 +1077,8 @@ def awx_receptor_workunit_reaper():
         jobs_with_unreleased_receptor_units = jobs_with_unreleased_receptor_units.exclude(status__in=ERROR_STATES)
     for job in jobs_with_unreleased_receptor_units:
         logger.debug(f"{job.log_format} is not active, reaping receptor work unit {job.work_unit_id}")
-        receptor_ctl.simple_command(f"work cancel {job.work_unit_id}")
-        receptor_ctl.simple_command(f"work release {job.work_unit_id}")
+        work.cancel(job.work_unit_id)
+        work.release(job.work_unit_id)
 
     administrative_workunit_reaper(receptor_work_list)
 
diff --git a/awx/main/tests/fake_receptor.py b/awx/main/tests/fake_receptor.py
new file mode 100644
index 0000000000..58e859d5bc
--- /dev/null
+++ b/awx/main/tests/fake_receptor.py
@@ -0,0 +1,138 @@
+"""In-memory stand-ins for receptor work units, for testing streaming and adoption without a mesh.
+
+FakeReceptorWork implements the ReceptorWork interface. Each unit holds the events its
+playbook produced and the state receptor reports. results() returns the same line
+protocol the ansible-runner worker writes, so AWX's real process-phase code parses it.
+
+Failures are injected per unit:
+  - break_after=N: the stream ends after N events with no status or eof, as when the
+    connection to the execution node drops.
+  - status_error: status() raises, as when the node is unreachable.
+  - released units disappear, and releasing or reading an unknown unit raises like receptor.
+
+RecordingDispatcher replaces CallbackQueueDispatcher so tests can see the events that
+would have gone to the callback receiver.
+"""
+
+import io
+import json
+import uuid
+
+from awx.main.tasks.receptor import ReceptorWork
+
+TERMINAL_STATE = {'successful': 'Succeeded', 'failed': 'Failed', 'canceled': 'Failed', 'error': 'Failed'}
+
+
+def make_events(count, failed_tasks=()):
+    """Return `count` runner events shaped like a playbook run: start, tasks, then stats."""
+    events = []
+    for counter in range(1, count + 1):
+        if counter == 1:
+            event = 'playbook_on_start'
+        elif counter == count:
+            event = 'playbook_on_stats'
+        elif counter in failed_tasks:
+            event = 'runner_on_failed'
+        else:
+            event = 'runner_on_ok'
+        events.append(
+            {
+                'uuid': str(uuid.uuid4()),
+                'counter': counter,
+                'event': event,
+                'stdout': f'line {counter}',
+                'start_line': counter - 1,
+                'end_line': counter,
+                'event_data': {},
+            }
+        )
+    return events
+
+
+class FakeWorkUnit:
+    def __init__(self, events=(), status='successful', state=None, exit_code=None, detail='', break_after=None, status_error=None):
+        self.events = list(events)
+        self.status = status
+        self.state = state or TERMINAL_STATE.get(status, 'Succeeded')
+        self.exit_code = exit_code
+        self.detail = detail
+        self.break_after = break_after
+        self.status_error = status_error
+
+    def stream(self):
+        lines = []
+        events = self.events if self.break_after is None else self.events[: self.break_after]
+        for event in events:
+            lines.append(json.dumps(event))
+        if self.break_after is None:
+            lines.append(json.dumps({'status': self.status, 'runner_ident': 'fake'}))
+            lines.append(json.dumps({'eof': True}))
+        return ('\n'.join(lines) + '\n').encode() if lines else b''
+
+
+class FakeSocket:
+    def __init__(self):
+        self.closed = False
+
+    def shutdown(self, how):
+        self.closed = True
+
+
+class FakeReceptorWork(ReceptorWork):
+    def __init__(self, units=None):
+        super().__init__(receptor_ctl=None)
+        self.units = dict(units or {})
+        self.calls = []
+
+    def _unit(self, unit_id):
+        if unit_id not in self.units:
+            raise RuntimeError(f'ERROR: unknown work unit {unit_id}')
+        return self.units[unit_id]
+
+    def list(self):
+        self.calls.append(('list',))
+        return {unit_id: {'StateName': unit.state, 'ExtraData': {}} for unit_id, unit in self.units.items()}
+
+    def status(self, unit_id):
+        self.calls.append(('status', unit_id))
+        unit = self._unit(unit_id)
+        if unit.status_error:
+            raise unit.status_error
+        status = {'StateName': unit.state, 'Detail': unit.detail, 'StdoutSize': len(unit.stream())}
+        if unit.exit_code is not None:
+            status['ExitCode'] = unit.exit_code
+        return status
+
+    def results(self, unit_id, startpos=None):
+        self.calls.append(('results', unit_id, startpos))
+        data = self._unit(unit_id).stream()
+        return FakeSocket(), io.BytesIO(data[startpos or 0 :])
+
+    def cancel(self, unit_id):
+        self.calls.append(('cancel', unit_id))
+        self._unit(unit_id).state = 'Failed'
+
+    def release(self, unit_id):
+        self.calls.append(('release', unit_id))
+        self._unit(unit_id)
+        del self.units[unit_id]
+
+
+class RecordingDispatcher:
+    """Collects callback-receiver messages instead of sending them to Redis."""
+
+    dispatched = []
+
+    def __init__(self):
+        pass
+
+    def dispatch(self, obj):
+        RecordingDispatcher.dispatched.append(obj)
+
+    @classmethod
+    def reset(cls):
+        cls.dispatched = []
+
+    @classmethod
+    def counters(cls):
+        return [e['counter'] for e in cls.dispatched if 'counter' in e]
diff --git a/awx/main/tests/functional/tasks/test_adoption_fake_receptor.py b/awx/main/tests/functional/tasks/test_adoption_fake_receptor.py
new file mode 100644
index 0000000000..b99cfff288
--- /dev/null
+++ b/awx/main/tests/functional/tasks/test_adoption_fake_receptor.py
@@ -0,0 +1,140 @@
+"""Adoption run end to end against a fake receptor work unit.
+
+These tests drive the real reattach, streaming, dedup and finalize code. Only receptor
+(FakeReceptorWork) and the callback-receiver queue (RecordingDispatcher) are replaced.
+"""
+
+from unittest import mock
+
+import pytest
+
+from awx.main.models import Job, JobEvent
+from awx.main.tasks.receptor import reattach_to_work_unit
+from awx.main.tasks.system import awx_receptor_workunit_reaper
+from awx.main.tests.fake_receptor import FakeReceptorWork, FakeWorkUnit, RecordingDispatcher, make_events
+
+UNIT = 'unit-1'
+
+
+@pytest.fixture(autouse=True)
+def adoption_env(settings, tmp_path):
+    settings.AWX_ISOLATION_BASE_PATH = str(tmp_path)
+    RecordingDispatcher.reset()
+    # connections.close_all() would end the test's transaction; events go to the recorder instead of Redis.
+    with (
+        mock.patch('awx.main.tasks.receptor.connections'),
+        mock.patch('awx.main.tasks.callback.CallbackQueueDispatcher', RecordingDispatcher),
+        mock.patch('awx.main.tasks.jobs.ScheduleTaskManager'),
+        mock.patch('awx.main.tasks.jobs.ScheduleWorkflowManager'),
+        mock.patch('awx.main.models.unified_jobs.UnifiedJob.websocket_emit_status'),
+    ):
+        yield
+
+
+@pytest.fixture
+def running_job(me_inst):
+    return Job.objects.create(controller_node=me_inst.hostname, execution_node='receptor-1', status='running', work_unit_id=UNIT)
+
+
+def save_events(job, counters):
+    for counter in counters:
+        JobEvent.objects.create(job=job, counter=counter, event='runner_on_ok', job_created=job.created)
+
+
+@pytest.mark.django_db
+def test_finished_unit_is_replayed_and_finalized(running_job):
+    work = FakeReceptorWork({UNIT: FakeWorkUnit(make_events(30))})
+
+    assert reattach_to_work_unit(running_job, work) is True
+
+    running_job.refresh_from_db()
+    assert running_job.status == 'successful'
+    assert RecordingDispatcher.counters() == list(range(1, 31))
+    assert UNIT not in work.units, 'work unit released after finalize'
+
+
+@pytest.mark.django_db
+def test_events_already_saved_are_not_dispatched_again(running_job):
+    save_events(running_job, range(1, 21))
+    work = FakeReceptorWork({UNIT: FakeWorkUnit(make_events(30))})
+
+    reattach_to_work_unit(running_job, work)
+
+    assert RecordingDispatcher.counters() == list(range(21, 31))
+
+
+@pytest.mark.django_db
+def test_events_saved_out_of_order_are_skipped(running_job):
+    # Parallel callback workers can commit counter 13 before 11 and 12.
+    save_events(running_job, [*range(1, 11), 13])
+    work = FakeReceptorWork({UNIT: FakeWorkUnit(make_events(20))})
+
+    reattach_to_work_unit(running_job, work)
+
+    assert RecordingDispatcher.counters() == [11, 12, *range(14, 21)]
+
+
+@pytest.mark.django_db
+def test_failed_playbook_finalizes_failed(running_job):
+    work = FakeReceptorWork({UNIT: FakeWorkUnit(make_events(10, failed_tasks=[5]), status='failed', exit_code=2)})
+
+    reattach_to_work_unit(running_job, work)
+
+    running_job.refresh_from_db()
+    assert running_job.status == 'failed'
+
+
+@pytest.mark.django_db
+def test_unit_still_running_is_deferred(running_job):
+    work = FakeReceptorWork({UNIT: FakeWorkUnit(make_events(10), state='Running')})
+
+    assert reattach_to_work_unit(running_job, work) is False
+
+    running_job.refresh_from_db()
+    assert running_job.status == 'running'
+    assert RecordingDispatcher.dispatched == []
+    assert UNIT in work.units
+
+
+@pytest.mark.django_db
+def test_unit_gone_is_deferred(running_job):
+    # The unit was released, or the node lost it, before adoption.
+    work = FakeReceptorWork()
+
+    assert reattach_to_work_unit(running_job, work) is False
+
+    running_job.refresh_from_db()
+    assert running_job.status == 'running'
+
+
+@pytest.mark.django_db
+def test_stream_broken_mid_replay_fails_a_successful_unit(running_job):
+    # The unit succeeded, but the results stream ends after 12 of 30 events.
+    # Today the job is finalized failed, with only the events read before the break.
+    work = FakeReceptorWork({UNIT: FakeWorkUnit(make_events(30), break_after=12)})
+
+    reattach_to_work_unit(running_job, work)
+
+    running_job.refresh_from_db()
+    assert running_job.status == 'failed'
+    assert RecordingDispatcher.counters() == list(range(1, 13))
+    assert 'Unexpected empty line' in running_job.job_explanation
+
+
+@pytest.mark.django_db
+def test_workunit_reaper_releases_units_of_finished_jobs(me_inst, settings):
+    settings.RECEPTOR_RELEASE_WORK = True
+    finished = Job.objects.create(controller_node=me_inst.hostname, status='successful', work_unit_id='done')
+    running = Job.objects.create(controller_node=me_inst.hostname, status='running', work_unit_id='live')
+    work = FakeReceptorWork({'done': FakeWorkUnit(), 'live': FakeWorkUnit(state='Running')})
+
+    with (
+        mock.patch('awx.main.tasks.system.get_receptor_ctl'),
+        mock.patch('awx.main.tasks.system.ReceptorWork', return_value=work),
+        mock.patch('awx.main.tasks.system.administrative_workunit_reaper'),
+    ):
+        awx_receptor_workunit_reaper()
+
+    assert ('cancel', finished.work_unit_id) in work.calls
+    assert finished.work_unit_id not in work.units
+    assert running.work_unit_id in work.units
diff --git a/awx/main/tests/functional/tasks/test_tasks_system.py b/awx/main/tests/functional/tasks/test_tasks_system.py
index cd3e138188..3b589e5a7c 100644
--- a/awx/main/tests/functional/tasks/test_tasks_system.py
+++ b/awx/main/tests/functional/tasks/test_tasks_system.py
@@ -11,6 +11,7 @@ from unittest.mock import MagicMock, patch
 import pytest
 
 from awx.main.tasks.adoption_decisions import JobAction
+from awx.main.tasks.receptor import ReceptorWork
 from awx.main.tasks.system import (
     CleanupImagesAndFiles,
     execution_node_health_check,
@@ -804,7 +805,7 @@ def test_adoption_finalizes_successful_job(me_inst):
         mock_instance._process_phase.return_value = MagicMock(status='successful', rc=0)
         mock_job_cls.return_value = mock_instance
 
-        reattach_to_work_unit(job, ctl)
+        reattach_to_work_unit(job, ReceptorWork(ctl))
 
     job.refresh_from_db()
     assert job.status == 'successful'
@@ -827,7 +828,7 @@ def test_adoption_finalizes_failed_job(me_inst):
         mock_instance._process_phase.return_value = MagicMock(status='failed', rc=1)
         mock_job_cls.return_value = mock_instance
 
-        reattach_to_work_unit(job, ctl)
+        reattach_to_work_unit(job, ReceptorWork(ctl))
 
     job.refresh_from_db()
     assert job.status == 'failed'
@@ -1107,7 +1108,7 @@ def test_reattach_receptor_command_fails(me_inst):
     ctl = MagicMock()
     ctl.simple_command.side_effect = Exception('connection refused')
 
-    result = reattach_to_work_unit(job, ctl)
+    result = reattach_to_work_unit(job, ReceptorWork(ctl))
 
     assert result is False
     job.refresh_from_db()
@@ -1128,7 +1129,7 @@ def test_reattach_exit_code_from_detail(me_inst):
         patch('awx.main.tasks.callback.RunnerCallback'),
     ):
         mock_job_cls.return_value = MagicMock()
-        reattach_to_work_unit(job, ctl)
+        reattach_to_work_unit(job, ReceptorWork(ctl))
 
     job.refresh_from_db()
     assert job.status == 'failed'
@@ -1150,7 +1151,7 @@ def test_reattach_exit_code_fallback_succeeded(me_inst):
         mock_instance = MagicMock()
         mock_instance._process_phase.return_value = MagicMock(status='successful', rc=0)
         mock_job_cls.return_value = mock_instance
-        reattach_to_work_unit(job, ctl)
+        reattach_to_work_unit(job, ReceptorWork(ctl))
 
     job.refresh_from_db()
     assert job.status == 'successful'
@@ -1170,7 +1171,7 @@ def test_reattach_exit_code_fallback_failed(me_inst):
         patch('awx.main.tasks.callback.RunnerCallback'),
     ):
         mock_job_cls.return_value = MagicMock()
-        reattach_to_work_unit(job, ctl)
+        reattach_to_work_unit(job, ReceptorWork(ctl))
 
     job.refresh_from_db()
     assert job.status == 'failed'
@@ -1192,7 +1193,7 @@ def test_reattach_process_phase_raises(me_inst):
         mock_instance = MagicMock()
         mock_instance._process_phase.side_effect = RuntimeError('boom')
         mock_job_cls.return_value = mock_instance
-        result = reattach_to_work_unit(job, ctl)
+        result = reattach_to_work_unit(job, ReceptorWork(ctl))
 
     assert result is True
     job.refresh_from_db()
@@ -1218,7 +1219,7 @@ def test_reattach_job_already_finalized(me_inst):
         mock_instance = MagicMock()
         mock_instance._process_phase.side_effect = _finalize_in_db
         mock_job_cls.return_value = mock_instance
-        reattach_to_work_unit(job, ctl)
+        reattach_to_work_unit(job, ReceptorWork(ctl))
 
     job.refresh_from_db()
     assert job.status == 'successful'
@@ -1236,7 +1237,10 @@ def test_adopt_job_async_calls_reattach(me_inst, settings):
         adopt_job_async(job.id)
 
     mock_ctl_factory.assert_called_once()
-    mock_reattach.assert_called_once_with(job, mock_ctl_factory.return_value)
+    mock_reattach.assert_called_once()
+    called_job, work = mock_reattach.call_args.args
+    assert called_job == job
+    assert work.receptor_ctl is mock_ctl_factory.return_value
 
 
 @pytest.mark.django_db
diff --git a/awx/main/tests/unit/tasks/test_jobs.py b/awx/main/tests/unit/tasks/test_jobs.py
index 0763d44bb6..0519c98fd8 100644
--- a/awx/main/tests/unit/tasks/test_jobs.py
+++ b/awx/main/tests/unit/tasks/test_jobs.py
@@ -682,41 +682,41 @@ def test_work_unit_released_on_quota_exceeded():
     Scenario: Job hits quota limit → _handle_work_error returns None →
     BaseTask.run() early-returns. Work unit MUST be released.
     """
-    receptor_ctl = mock.MagicMock()
+    work = mock.MagicMock()
     receptor_job = mock.MagicMock()
     receptor_job.unit_id = "test-unit-123"
-    receptor_job.receptor_ctl = receptor_ctl
+    receptor_job.work = work
 
     # When res is None (quota exceeded case)
     res = None
 
     # This is the fixed code path from jobs.py lines 810-818
     if not res:
-        if receptor_job and getattr(receptor_job, 'receptor_ctl', None):
+        if receptor_job and getattr(receptor_job, 'work', None):
             try:
-                receptor_job._receptor_release_work(receptor_job.receptor_ctl, 'error')
+                receptor_job._receptor_release_work(receptor_job.work, 'error')
             except Exception:
                 pass
 
     # Verify release was called
-    receptor_job._receptor_release_work.assert_called_once_with(receptor_ctl, 'error')
+    receptor_job._receptor_release_work.assert_called_once_with(work, 'error')
 
 
-def test_work_unit_not_released_if_no_receptor_ctl():
+def test_work_unit_not_released_if_no_receptor_work():
     """
-    Edge case: receptor_ctl not set (transmit phase failed before unit submission).
+    Edge case: work not set (transmit phase failed before unit submission).
 
-    Verify graceful handling when receptor_ctl is None.
+    Verify graceful handling when work is None.
     """
     receptor_job = mock.MagicMock(spec=['unit_id', '_receptor_release_work'])
-    receptor_job.receptor_ctl = None  # Not set
+    receptor_job.work = None  # Not set
 
     res = None
 
     # Code should not crash
     if not res:
-        if receptor_job and getattr(receptor_job, 'receptor_ctl', None):
-            receptor_job._receptor_release_work(receptor_job.receptor_ctl, 'error')
+        if receptor_job and getattr(receptor_job, 'work', None):
+            receptor_job._receptor_release_work(receptor_job.work, 'error')
         # else: No release needed, unit was never submitted
 
     # Verify _receptor_release_work was NOT called
diff --git a/awx/main/tests/unit/tasks/test_receptor_adoption.py b/awx/main/tests/unit/tasks/test_receptor_adoption.py
index 2188c3d41b..c9de18e54a 100644
--- a/awx/main/tests/unit/tasks/test_receptor_adoption.py
+++ b/awx/main/tests/unit/tasks/test_receptor_adoption.py
@@ -20,6 +20,7 @@ from awx.main.tasks.callback import RunnerCallback
 from awx.main.tasks.jobs import _finalize_job_run
 from awx.main.tasks.receptor import (
     AWXReceptorJob,
+    ReceptorWork,
     _AdoptionTask,
     _compute_adoption_dedup,
     _finalize_adopted_job,
@@ -127,7 +128,7 @@ def test_process_phase_success(mock_signal, mock_connections):
     rj.processor = Mock(return_value=expected_res)
     ctl = _make_receptor_ctl()
 
-    res = rj._process_phase(ctl)
+    res = rj._process_phase(ReceptorWork(ctl))
 
     assert res.status == 'successful'
     mock_connections.close_all.assert_called_once()
@@ -145,7 +146,7 @@ def test_process_phase_signal_exit(mock_signal, mock_connections):
     rj.processor = Mock()
     ctl = _make_receptor_ctl()
 
-    res = rj._process_phase(ctl)
+    res = rj._process_phase(ReceptorWork(ctl))
 
     assert res.status == 'canceled'
     ctl.simple_command.assert_any_call('work cancel unit-1')
@@ -161,7 +162,7 @@ def test_handle_work_error_result_traceback_present():
     ctl = _make_receptor_ctl()
     err_res = _Result(status='error', rc=1)
 
-    result = rj._handle_work_error(ctl, err_res)
+    result = rj._handle_work_error(ReceptorWork(ctl), err_res)
 
     assert result is err_res
     ctl.simple_command.assert_not_called()
@@ -179,7 +180,7 @@ def test_handle_work_error_status_command_raises():
     err_res = _Result(status='error', rc=1)
 
     # should not raise; falls through and returns res
-    result = rj._handle_work_error(ctl, err_res)
+    result = rj._handle_work_error(ReceptorWork(ctl), err_res)
     assert result is err_res
 
 
@@ -193,7 +194,7 @@ def test_handle_work_error_exceeded_quota():
     ctl = _make_receptor_ctl(detail='exceeded quota for namespace')
     err_res = _Result(status='error', rc=1)
 
-    result = rj._handle_work_error(ctl, err_res)
+    result = rj._handle_work_error(ReceptorWork(ctl), err_res)
 
     assert result is None
     rj.task.update_model.assert_called_once_with(1, status='pending')
@@ -209,7 +210,7 @@ def test_handle_work_error_reads_receptor_output_on_failed():
     ctl = _make_receptor_ctl(state='Failed', stdout_size=2000)
     err_res = _Result(status='error', rc=1)
 
-    rj._handle_work_error(ctl, err_res)
+    rj._handle_work_error(ReceptorWork(ctl), err_res)
 
     rj.task.runner_callback.delay_update.assert_called_once()
     call_kwargs = rj.task.runner_callback.delay_update.call_args[1]
@@ -226,7 +227,7 @@ def test_handle_work_error_detail_only():
     ctl = _make_receptor_ctl(state='Succeeded', detail='some error detail')
     err_res = _Result(status='error', rc=1)
 
-    rj._handle_work_error(ctl, err_res)
+    rj._handle_work_error(ReceptorWork(ctl), err_res)
 
     rj.task.runner_callback.delay_update.assert_called_once()
     call_kwargs = rj.task.runner_callback.delay_update.call_args[1]
@@ -243,7 +244,7 @@ def test_handle_work_error_no_detail_no_output():
     ctl = _make_receptor_ctl(state='Succeeded', detail='')
     err_res = _Result(status='error', rc=1)
 
-    result = rj._handle_work_error(ctl, err_res)
+    result = rj._handle_work_error(ReceptorWork(ctl), err_res)
 
     assert result is err_res
     rj.task.runner_callback.delay_update.assert_not_called()
@@ -261,7 +262,7 @@ def test_handle_work_error_get_results_raises():
     err_res = _Result(status='error', rc=1)
 
     with pytest.raises(RuntimeError):
-        rj._handle_work_error(ctl, err_res)
+        rj._handle_work_error(ReceptorWork(ctl), err_res)
 
 
 # ---------------------------------------------------------------------------
@@ -296,7 +297,7 @@ def test_reattach_sets_job_created_on_callback(mock_rmtree, mock_pdd, mock_relea
         created_callbacks.append(self)
 
     with patch.object(RealCallback, '__init__', capturing_cb_init):
-        reattach_to_work_unit(job, ctl)
+        reattach_to_work_unit(job, ReceptorWork(ctl))
 
     assert len(created_callbacks) == 1
     assert created_callbacks[0].job_created == str(job.created)
@@ -325,7 +326,7 @@ def test_reattach_releases_work_unit_on_success(mock_rmtree, mock_pdd, mock_rele
     ctl.simple_command.return_value = {'StateName': ''}
     mock_process.return_value = Mock(status='successful', rc=0)
 
-    reattach_to_work_unit(job, ctl)
+    reattach_to_work_unit(job, ReceptorWork(ctl))
 
     mock_release.assert_called_once()
 
@@ -350,7 +351,7 @@ def test_reattach_releases_work_unit_on_failure(mock_rmtree, mock_pdd, mock_rele
         {'StateName': 'Succeeded', 'ExitCode': 0, 'Detail': ''},  # fallback re-check
     ]
 
-    result = reattach_to_work_unit(job, ctl)
+    result = reattach_to_work_unit(job, ReceptorWork(ctl))
 
     assert result is True  # finalized via exit_code path
     mock_release.assert_called_once()
@@ -391,7 +392,7 @@ def test_reattach_sets_parent_workflow_job_id_when_workflow_child(mock_rmtree, m
         created_callbacks.append(self)
 
     with patch.object(RealCallback, '__init__', capturing_cb_init):
-        reattach_to_work_unit(job, ctl)
+        reattach_to_work_unit(job, ReceptorWork(ctl))
 
     assert len(created_callbacks) == 1
     assert created_callbacks[0].parent_workflow_job_id == 999
@@ -415,7 +416,7 @@ def test_reattach_workflow_job_lookup_exception_swallowed(mock_rmtree, mock_pdd,
     ctl = Mock()
     ctl.simple_command.return_value = {'StateName': 'Succeeded', 'ExitCode': 0, 'Detail': ''}
 
-    result = reattach_to_work_unit(job, ctl)
+    result = reattach_to_work_unit(job, ReceptorWork(ctl))
 
     assert result is True  # exception swallowed, adoption completes
 
@@ -552,7 +553,7 @@ def test_reattach_non_workflow_job_no_parent_id(mock_rmtree, mock_pdd, mock_rele
         created_callbacks.append(self)
 
     with patch.object(RealCallback, '__init__', capturing_cb_init):
-        reattach_to_work_unit(job, ctl)
+        reattach_to_work_unit(job, ReceptorWork(ctl))
 
     assert created_callbacks[0].parent_workflow_job_id is None
 
@@ -585,7 +586,7 @@ def test_reattach_running_defers_without_streaming_variant(mock_rmtree, mock_pdd
     ctl = Mock()
     ctl.simple_command.return_value = {'StateName': 'Running'}
 
-    result = reattach_to_work_unit(job, ctl)
+    result = reattach_to_work_unit(job, ReceptorWork(ctl))
 
     mock_process.assert_not_called()
     assert result is False
@@ -619,7 +620,7 @@ def test_reattach_process_phase_failure_falls_back_to_work_status(mock_rmtree, m
     ]
     mock_process.side_effect = RuntimeError('process phase failed')
 
-    result = reattach_to_work_unit(job, ctl)  # must not raise
+    result = reattach_to_work_unit(job, ReceptorWork(ctl))  # must not raise
 
     # Returns True (function completed without re-raising)
     assert result is True
@@ -645,7 +646,7 @@ def test_reattach_pending_defers_without_streaming():
     ctl.simple_command.return_value = {'StateName': 'Pending'}
 
     with patch('awx.main.tasks.receptor.AWXReceptorJob._process_phase') as mock_process:
-        result = reattach_to_work_unit(job, ctl)
+        result = reattach_to_work_unit(job, ReceptorWork(ctl))
 
     assert result is False
     mock_process.assert_not_called()
@@ -673,7 +674,7 @@ def test_reattach_succeeded_goes_through_process_phase(mock_rmtree, mock_pdd, mo
     ctl.simple_command.return_value = {'StateName': 'Succeeded', 'ExitCode': 0, 'Detail': ''}
 
     with patch('awx.main.tasks.receptor.AWXReceptorJob._process_phase', return_value=Mock(status='successful', rc=0)) as mock_process:
-        reattach_to_work_unit(job, ctl)
+        reattach_to_work_unit(job, ReceptorWork(ctl))
 
     mock_process.assert_called_once()
     mock_finalize_job.assert_called_once()
@@ -699,7 +700,7 @@ def test_reattach_failed_goes_through_process_phase(mock_rmtree, mock_pdd, mock_
     ctl.simple_command.return_value = {'StateName': 'Failed', 'ExitCode': 1, 'Detail': 'exit status 1'}
 
     with patch('awx.main.tasks.receptor.AWXReceptorJob._process_phase', return_value=Mock(status='failed', rc=1)) as mock_process:
-        reattach_to_work_unit(job, ctl)
+        reattach_to_work_unit(job, ReceptorWork(ctl))
 
     mock_process.assert_called_once()
     mock_finalize_job.assert_called_once()
@@ -724,7 +725,7 @@ def test_reattach_canceled_goes_through_process_phase(mock_rmtree, mock_pdd, moc
     ctl.simple_command.return_value = {'StateName': 'Canceled', 'Detail': ''}
 
     with patch('awx.main.tasks.receptor.AWXReceptorJob._process_phase', return_value=Mock(status='error', rc=1)) as mock_process:
-        reattach_to_work_unit(job, ctl)
+        reattach_to_work_unit(job, ReceptorWork(ctl))
 
     mock_process.assert_called_once()
     mock_finalize_job.assert_called_once()
@@ -752,7 +753,7 @@ def test_reattach_running_defers_without_streaming(mock_rmtree, mock_pdd, mock_r
     ctl.simple_command.return_value = {'StateName': 'Running'}
 
     with patch('awx.main.tasks.receptor.AWXReceptorJob._process_phase') as mock_process:
-        result = reattach_to_work_unit(job, ctl)
+        result = reattach_to_work_unit(job, ReceptorWork(ctl))
 
     assert result is False
     mock_process.assert_not_called()
@@ -969,7 +970,7 @@ def test_reattach_returns_false_when_status_command_raises():
     ctl = Mock()
     ctl.simple_command.side_effect = Exception('socket closed')
 
-    result = reattach_to_work_unit(job, ctl)
+    result = reattach_to_work_unit(job, ReceptorWork(ctl))
 
     assert result is False
 
@@ -1320,7 +1321,7 @@ def test_handle_work_error_no_output_no_detail(mock_logger):
     receptor_job.task.runner_callback.extra_update_fields = {}
     receptor_job.task.runner_callback.event_ct = 0  # No events
 
-    result = receptor_job._handle_work_error(receptor_ctl, res)
+    result = receptor_job._handle_work_error(ReceptorWork(receptor_ctl), res)
 
     assert result == res
     mock_logger.warning.assert_called()
@@ -1345,7 +1346,7 @@ def test_handle_work_error_state_name_empty_string(mock_logger):
     receptor_job.task.runner_callback.extra_update_fields = {}
     receptor_job.task.runner_callback.event_ct = 5
 
-    result = receptor_job._handle_work_error(receptor_ctl, res)
+    result = receptor_job._handle_work_error(ReceptorWork(receptor_ctl), res)
 
     assert result == res
     # Verify exception was logged
@@ -1377,7 +1378,7 @@ def test_handle_work_error_delay_update_with_receptor_output():
     receptor_job.task.runner_callback.extra_update_fields = {}
     receptor_job.task.runner_callback.event_ct = 0  # No events → fetch output
 
-    result = receptor_job._handle_work_error(receptor_ctl, res)
+    result = receptor_job._handle_work_error(ReceptorWork(receptor_ctl), res)
 
     assert result == res
     # Verify delay_update was called with receptor output
@@ -1406,7 +1407,7 @@ def test_handle_work_error_delay_update_with_detail_fallback():
     receptor_job.task.runner_callback.extra_update_fields = {}
     receptor_job.task.runner_callback.event_ct = 5
 
-    result = receptor_job._handle_work_error(receptor_ctl, res)
+    result = receptor_job._handle_work_error(ReceptorWork(receptor_ctl), res)
 
     assert result == res
     # Verify delay_update was called with detail