diff --git a/awx/main/tests/adoption_cluster.py b/awx/main/tests/adoption_cluster.py
new file mode 100644
index 0000000000..1a6285c224
--- /dev/null
+++ b/awx/main/tests/adoption_cluster.py
@@ -0,0 +1,92 @@
+"""A small cluster in the test database, for job adoption failure scenarios.
+
+The scenarios were first found with the failpoint harness on a two-controller
+docker-compose cluster. Here the same cluster is rows in the test database:
+
+  ctrl-0      this controller, running the heartbeat under test
+  ctrl-1      a peer controller, judged lost when its last_seen is stale
+  receptor-1  the execution node the jobs run on
+
+Receptor is replaced with FakeReceptorWork and the callback-receiver queue with
+RecordingDispatcher (see awx.main.tests.fake_receptor), so the real heartbeat, reap,
+adoption and finalize code runs without a mesh.
+"""
+
+import contextlib
+from unittest import mock
+
+from django.utils.timezone import now, timedelta
+
+from awx.main.models import Instance
+from awx.main.tasks.system import _heartbeat_handle_lost_instances, _heartbeat_instance_management
+from awx.main.tests.fake_receptor import RecordingDispatcher
+
+THIS = 'ctrl-0'
+PEER = 'ctrl-1'
+EXEC = 'receptor-1'
+UNIT = 'unit-1'
+GRACE = 120  # CLUSTER_NODE_HEARTBEAT_PERIOD * CLUSTER_NODE_MISSED_HEARTBEAT_TOLERANCE
+
+
+def configure(settings):
+    settings.CLUSTER_HOST_ID = THIS
+    settings.AWX_AUTO_DEPROVISION_INSTANCES = False
+    settings.CLUSTER_NODE_HEARTBEAT_PERIOD = 60
+    settings.CLUSTER_NODE_MISSED_HEARTBEAT_TOLERANCE = 2
+
+
+def make_cluster(peer_seconds_since_seen=GRACE + 80):
+    """Create ctrl-0 (healthy), ctrl-1 (silent for peer_seconds_since_seen) and receptor-1."""
+    t = now()
+    this = Instance.objects.create(hostname=THIS, node_type='control', node_state='ready', last_seen=t)
+    peer = Instance.objects.create(hostname=PEER, node_type='control', node_state='ready', last_seen=t - timedelta(seconds=peer_seconds_since_seen))
+    Instance.objects.create(hostname=EXEC, node_type='execution', node_state='ready', last_seen=t)
+    return this, peer
+
+
+def mesh(*connected):
+    """Receptor status as ctrl-0 sees it, with ctrl-0 connected to each node in connected."""
+    return {'KnownConnectionCosts': {THIS: {node: 1 for node in connected}}, 'Advertisements': []}
+
+
+def run_lost_instance_handling(mesh_status, work=None, clock=None):
+    """Run ctrl-0's heartbeat up to and including lost-instance handling. Return the lost hostnames.
+
+    work, when given, is returned wherever system.py opens ReceptorWork, so a test can see
+    whether a work unit was canceled.
+    """
+    ctl = mock.MagicMock()
+    ctl.simple_command.return_value = mesh_status
+    with contextlib.ExitStack() as stack:
+        stack.enter_context(mock.patch('awx.main.tasks.system.get_receptor_ctl', return_value=ctl))
+        stack.enter_context(mock.patch('awx.main.tasks.system.inspect_execution_and_hop_nodes'))
+        stack.enter_context(mock.patch.object(Instance, 'local_health_check'))
+        if work is not None:
+            stack.enter_context(mock.patch('awx.main.tasks.system.ReceptorWork', return_value=work))
+        this_inst, _, lost, _ = _heartbeat_instance_management(clock)
+        _heartbeat_handle_lost_instances(lost, this_inst)
+    return sorted(inst.hostname for inst in lost)
+
+
+@contextlib.contextmanager
+def adoption_env(settings, tmp_path):
+    """Patch what adoption touches outside the database: Redis, websockets and the schedulers."""
+    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
+
+
+def assert_playbook_not_left_running(job, work):
+    """Fail when the job is no longer active but its work unit was never canceled."""
+    job.refresh_from_db()
+    still_active = job.status in ('running', 'waiting')
+    unit_canceled = ('cancel', job.work_unit_id) in work.calls
+    assert still_active or unit_canceled, f'job is {job.status!r} but its work unit {job.work_unit_id} was never canceled'
diff --git a/awx/main/tests/functional/tasks/test_adoption_clock.py b/awx/main/tests/functional/tasks/test_adoption_clock.py
index efd7956732..fbcd78cde2 100644
--- a/awx/main/tests/functional/tasks/test_adoption_clock.py
+++ b/awx/main/tests/functional/tasks/test_adoption_clock.py
@@ -63,15 +63,17 @@ def test_peer_is_lost_only_after_the_grace_period(heartbeat_settings):
     assert lost_hostnames(clock) == ['ctrl-1']
 
 
+@pytest.mark.xfail(strict=True, reason='AAP-96827: a controller with a fast clock marks a healthy peer lost')
 @pytest.mark.django_db
-def test_judge_with_a_fast_clock_finds_a_healthy_peer_lost(heartbeat_settings):
-    # ctrl-0's clock runs five minutes fast, so its own last_seen is five minutes ahead.
-    # ctrl-1 heartbeated just now by the correct clock and is healthy.
+def test_judge_with_a_fast_clock_does_not_find_a_healthy_peer_lost(heartbeat_settings):
+    # 2.8 Clock skew. ctrl-0's clock runs five minutes fast, so its own last_seen is five
+    # minutes ahead. ctrl-1 heartbeated just now by the correct clock and is healthy.
+    # Today ctrl-0 judges it by its own clock and finds it lost.
     t = now()
     control_node('ctrl-0', t + timedelta(minutes=5))
     control_node('ctrl-1', t)
 
-    assert lost_hostnames(FakeClock(t + timedelta(minutes=5))) == ['ctrl-1']
+    assert lost_hostnames(FakeClock(t + timedelta(minutes=5))) == []
 
 
 @pytest.mark.django_db
diff --git a/awx/main/tests/functional/tasks/test_scenarios_adoption.py b/awx/main/tests/functional/tasks/test_scenarios_adoption.py
new file mode 100644
index 0000000000..467089d77a
--- /dev/null
+++ b/awx/main/tests/functional/tasks/test_scenarios_adoption.py
@@ -0,0 +1,118 @@
+"""Failure scenarios in the adoption lifecycle: claim, stream, finalize (write-up section 3).
+
+Tests marked xfail reproduce a defect found on devel and assert the behavior it should
+have. The others guard behavior devel already gets right.
+"""
+
+from unittest import mock
+
+import pytest
+from django.db import DatabaseError
+from django.utils.timezone import now, timedelta
+
+from awx.main.models import Job, JobEvent
+from awx.main.tasks import jobs
+from awx.main.tasks.system import adopt_job
+from awx.main.tests.adoption_cluster import UNIT, adoption_env
+from awx.main.tests.fake_clock import FakeClock
+from awx.main.tests.fake_receptor import FakeReceptorWork, FakeWorkUnit, RecordingDispatcher, make_events
+
+TIMEOUT = 300  # HADR_JOB_ADOPTION_TIMEOUT
+
+
+@pytest.fixture(autouse=True)
+def env(settings, tmp_path):
+    settings.HADR_JOB_ADOPTION_TIMEOUT = TIMEOUT
+    with adoption_env(settings, tmp_path):
+        yield
+
+
+@pytest.fixture
+def orphan(me_inst):
+    """A running job this controller lost track of, started a minute ago."""
+    return Job.objects.create(controller_node=me_inst.hostname, status='running', work_unit_id=UNIT, started=now() - timedelta(seconds=60))
+
+
+def save_events(job, counters, created=None):
+    for counter in counters:
+        JobEvent.objects.create(job=job, counter=counter, event='runner_on_ok', job_created=job.created)
+    if created:
+        JobEvent.objects.filter(job=job).update(created=created)
+
+
+@pytest.mark.xfail(strict=True, reason='AAP-96828: an error while finalizing a resumed job releases its unit and strands the job')
+@pytest.mark.django_db
+def test_finalize_error_is_retried_on_the_next_adoption(orphan):
+    # 3.5 Finalize strand. The first adoption streams the whole unit, then finalization
+    # raises. Today the unit is released anyway, so the next adoption finds nothing and the
+    # job stays running until the adoption timeout fails it.
+    work = FakeReceptorWork({UNIT: FakeWorkUnit(make_events(10))})
+    real_finalize = jobs._finalize_job_run
+    attempts = []
+
+    def finalize_fails_once(*args, **kwargs):
+        attempts.append(1)
+        if len(attempts) == 1:
+            raise DatabaseError('connection lost during finalize')
+        return real_finalize(*args, **kwargs)
+
+    with mock.patch('awx.main.tasks.jobs._finalize_job_run', side_effect=finalize_fails_once):
+        adopt_job(orphan.id, open_work=lambda: work)
+        adopt_job(orphan.id, open_work=lambda: work)
+
+    orphan.refresh_from_db()
+    assert orphan.status == 'successful'
+
+
+@pytest.mark.xfail(strict=True, reason='AAP-96829: a restart fails a healthy job with a long quiet task while its playbook keeps running')
+@pytest.mark.django_db
+def test_quiet_job_whose_unit_is_still_running_is_not_failed(orphan):
+    # 3.6 Quiet deadline. The playbook is in a long task with no output, so its last event
+    # is older than HADR_JOB_ADOPTION_TIMEOUT, but receptor reports the unit running.
+    # Today adoption measures the deadline from the last event and fails the job, while
+    # the playbook keeps running on the execution node.
+    clock = FakeClock()
+    save_events(orphan, range(1, 6), created=clock() - timedelta(seconds=TIMEOUT + 60))
+    work = FakeReceptorWork({UNIT: FakeWorkUnit(make_events(10), state='Running')})
+
+    adopt_job(orphan.id, open_work=lambda: work, clock=clock)
+
+    orphan.refresh_from_db()
+    assert orphan.status == 'running'
+
+
+@pytest.mark.django_db
+def test_adoption_waits_through_repeated_restarts_and_replays_each_event_once(orphan):
+    # 3.7 Three restarts during one job. While the unit runs, each adoption defers; once
+    # it finishes, the replay sends only the events not yet saved, each once.
+    clock = FakeClock()
+    save_events(orphan, range(1, 13), created=clock())
+    unit = FakeWorkUnit(make_events(30), state='Running')
+    work = FakeReceptorWork({UNIT: unit})
+
+    for _ in range(3):
+        adopt_job(orphan.id, open_work=lambda: work, clock=clock)
+        orphan.refresh_from_db()
+        assert orphan.status == 'running'
+        assert RecordingDispatcher.dispatched == []
+        clock.advance(60)
+
+    unit.state = 'Succeeded'
+    adopt_job(orphan.id, open_work=lambda: work, clock=clock)
+
+    orphan.refresh_from_db()
+    assert orphan.status == 'successful'
+    assert RecordingDispatcher.counters() == list(range(13, 31))
+
+
+@pytest.mark.django_db
+def test_large_replay_sends_every_event_once(orphan):
+    # 5.8 Large replay. A 20,000 event unit with the first 500 already saved.
+    save_events(orphan, range(1, 501))
+    work = FakeReceptorWork({UNIT: FakeWorkUnit(make_events(20000))})
+
+    adopt_job(orphan.id, open_work=lambda: work)
+
+    orphan.refresh_from_db()
+    assert orphan.status == 'successful'
+    assert RecordingDispatcher.counters() == list(range(501, 20001))
diff --git a/awx/main/tests/functional/tasks/test_scenarios_data.py b/awx/main/tests/functional/tasks/test_scenarios_data.py
new file mode 100644
index 0000000000..53a728a6e7
--- /dev/null
+++ b/awx/main/tests/functional/tasks/test_scenarios_data.py
@@ -0,0 +1,139 @@
+"""Failure scenarios about the data a job leaves behind after adoption or a reap (write-up section 5).
+
+Tests marked xfail reproduce a defect found on devel and assert the behavior it should have.
+"""
+
+import json
+import os
+from unittest import mock
+
+import pytest
+
+from awx.main.models import ActivityStream, Job
+from awx.main.tasks.system import adopt_job, awx_receptor_workunit_reaper
+from awx.main.tests.adoption_cluster import EXEC, PEER, UNIT, adoption_env, configure, make_cluster, mesh, run_lost_instance_handling
+from awx.main.tests.fake_receptor import FakeReceptorWork, FakeWorkUnit, RecordingDispatcher, make_events
+
+
+@pytest.fixture(autouse=True)
+def env(settings, tmp_path):
+    configure(settings)
+    with adoption_env(settings, tmp_path):
+        yield
+
+
+def xfail(bug, summary):
+    return pytest.mark.xfail(strict=True, reason=f'{bug}: {summary}')
+
+
+# ── A reaped job's own result ─────────────────────────────────────────────────
+
+
+@pytest.mark.parametrize(
+    'bug',
+    [
+        pytest.param('AAP-96839', id='AAP-96839-relaunch-on-failed-hosts', marks=xfail('AAP-96839', 'a reaped job cannot be relaunched on its failed hosts')),
+        pytest.param('AAP-96846', id='AAP-96846-host-metrics', marks=xfail('AAP-96846', 'a reaped job records no host metrics')),
+        pytest.param('AAP-96847', id='AAP-96847-host-pointers', marks=xfail('AAP-96847', 'a reaped job leaves hosts showing the previous job')),
+    ],
+)
+@pytest.mark.django_db
+def test_stats_of_a_reaped_job_are_saved_when_its_unit_finishes(me_inst, settings, bug):
+    # 4.20, 5.10, 5.12. The job was marked failed while its playbook ran on, and the
+    # playbook then finished. Its stats event, which writes host summaries, host metrics
+    # and each host's last job, is never read: the work unit reaper cancels and releases
+    # the finished unit. So relaunch on failed hosts has no summaries to use, the hosts
+    # count nothing, and they keep pointing at the previous job.
+    settings.RECEPTOR_RELEASE_WORK = True
+    Job.objects.create(
+        controller_node=me_inst.hostname, execution_node=EXEC, status='failed', work_unit_id=UNIT, job_explanation='Job reaped due to instance shutdown'
+    )
+    work = FakeReceptorWork({UNIT: FakeWorkUnit(make_events(10))})
+
+    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 'playbook_on_stats' in [event.get('event') for event in RecordingDispatcher.dispatched]
+
+
+# ── Host ids on replayed events ───────────────────────────────────────────────
+
+
+def host_events(hostnames):
+    """Runner events for one task on each host, then the stats event."""
+    events = make_events(len(hostnames) + 2)
+    for event, hostname in zip(events[1:-1], hostnames):
+        event['event_data'] = {'host': hostname}
+    return events
+
+
+@pytest.mark.xfail(strict=True, reason='AAP-96843: replayed events link to the wrong hosts after an inventory change')
+@pytest.mark.django_db
+def test_replayed_events_keep_the_host_ids_the_job_started_with(me_inst, inventory):
+    # 5.2 Host map after an inventory change. A normal run maps each host name to the id
+    # it had when the job started. Adoption rebuilds the map from the inventory as it is
+    # now, so after a change while the job was orphaned, replayed events lose the deleted
+    # and disabled hosts and point a recreated host at its new id.
+    hosts = {name: inventory.hosts.create(name=name) for name in ('fp-h1', 'fp-h2', 'fp-h3', 'fp-h4', 'fp-h5')}
+    ids_at_start = {name: host.id for name, host in hosts.items()}
+    job = Job.objects.create(controller_node=me_inst.hostname, status='running', work_unit_id=UNIT, inventory=inventory)
+
+    # While the job is orphaned: delete fp-h2, recreate fp-h3, disable fp-h4.
+    hosts['fp-h2'].delete()
+    hosts['fp-h3'].delete()
+    inventory.hosts.create(name='fp-h3')
+    hosts['fp-h4'].enabled = False
+    hosts['fp-h4'].save()
+
+    work = FakeReceptorWork({UNIT: FakeWorkUnit(host_events(list(hosts)))})
+    adopt_job(job.id, open_work=lambda: work)
+
+    replayed = {event['host_name']: event.get('host_id') for event in RecordingDispatcher.dispatched if event.get('host_name')}
+    assert replayed == ids_at_start
+
+
+# ── What records a reap ───────────────────────────────────────────────────────
+
+
+@pytest.mark.xfail(strict=True, reason='AAP-96849: the activity stream does not record that a job was reaped')
+@pytest.mark.django_db
+def test_reap_is_recorded_in_the_activity_stream():
+    # 5.16 Activity stream. Nothing records that a lost controller's job was marked
+    # failed, so an auditor cannot tell a reaped job from one whose playbook failed.
+    make_cluster()
+    job = Job.objects.create(controller_node=PEER, execution_node=EXEC, status='running', work_unit_id=UNIT)
+    entries_before = ActivityStream.objects.filter(job=job).count()
+
+    run_lost_instance_handling(mesh(EXEC))
+
+    job.refresh_from_db()
+    assert job.status == 'failed'
+    assert ActivityStream.objects.filter(job=job).count() > entries_before
+
+
+# ── Files left on disk ────────────────────────────────────────────────────────
+
+
+@pytest.mark.xfail(strict=True, reason='AAP-96852: a dead controller\'s job files stay on disk after it returns')
+@pytest.mark.django_db
+def test_resumed_job_removes_its_original_private_data_dir(me_inst, settings):
+    # 5.20 Credentials left on disk. The job's private data directory from before the
+    # restart holds its inventory and extra variables. Adoption works in a new directory
+    # and removes only that one, so the original stays until periodic cleanup's grace
+    # period (60 minutes by default) has passed.
+    job = Job.objects.create(controller_node=me_inst.hostname, status='running', work_unit_id=UNIT)
+    original = os.path.join(settings.AWX_ISOLATION_BASE_PATH, f'awx_{job.pk}_before_restart')
+    os.makedirs(os.path.join(original, 'env'))
+    with open(os.path.join(original, 'env', 'extravars'), 'w') as f:
+        json.dump({'survey_password': 'secret'}, f)
+    work = FakeReceptorWork({UNIT: FakeWorkUnit(make_events(10))})
+
+    adopt_job(job.id, open_work=lambda: work)
+
+    job.refresh_from_db()
+    assert job.status == 'successful'
+    assert not os.path.exists(original), f'{original} is still on disk'
diff --git a/awx/main/tests/functional/tasks/test_scenarios_job_control.py b/awx/main/tests/functional/tasks/test_scenarios_job_control.py
new file mode 100644
index 0000000000..7189af8a1f
--- /dev/null
+++ b/awx/main/tests/functional/tasks/test_scenarios_job_control.py
@@ -0,0 +1,177 @@
+"""Failure scenarios about canceling, capacity and launches around orphaned jobs (write-up section 4).
+
+Tests marked xfail reproduce a defect found on devel and assert the behavior it should
+have. The others guard behavior devel already gets right.
+
+Several devel bugs share one cause: when a lost controller's job is marked failed, its
+playbook keeps running on the execution node. Whatever depends on the job being active
+(launch blocking, relaunch, instance group limits, execution node capacity, workflow
+branching) then sees a finished job. test_reaped_job_does_not_leave_its_playbook_running
+covers each of those bugs with the job shape its scenario used.
+"""
+
+import pytest
+
+from awx.main.models import InstanceGroup, Job, JobTemplate, WorkflowJob, WorkflowJobNode
+from awx.main.tasks.system import adopt_job
+from awx.main.tests.adoption_cluster import (
+    EXEC,
+    PEER,
+    UNIT,
+    adoption_env,
+    assert_playbook_not_left_running,
+    configure,
+    make_cluster,
+    mesh,
+    run_lost_instance_handling,
+)
+from awx.main.tests.fake_receptor import FakeReceptorWork, FakeWorkUnit, make_events
+
+
+@pytest.fixture(autouse=True)
+def env(settings, tmp_path):
+    configure(settings)
+    with adoption_env(settings, tmp_path):
+        yield
+
+
+def peer_job(**fields):
+    """A job controlled by ctrl-1 and running on receptor-1."""
+    fields.setdefault('work_unit_id', UNIT)
+    return Job.objects.create(controller_node=PEER, execution_node=EXEC, status='running', celery_task_id='peer-task', **fields)
+
+
+def running_unit():
+    return FakeReceptorWork({UNIT: FakeWorkUnit(make_events(10), state='Running')})
+
+
+# ── Canceling an orphaned job ─────────────────────────────────────────────────
+
+
+@pytest.mark.xfail(strict=True, reason='AAP-96830: canceling a job whose controller is down is dropped; the job never ends canceled')
+@pytest.mark.django_db
+def test_cancel_requested_while_orphaned_cancels_the_unit(me_inst):
+    # 4.1 Cancel while orphaned. The user canceled while the controller was down, so the
+    # cancel reached no one. Today adoption never reads cancel_flag: it waits for the unit
+    # and finishes the job, successful, with cancel_flag set.
+    job = Job.objects.create(controller_node=me_inst.hostname, status='running', work_unit_id=UNIT, cancel_flag=True)
+    work = running_unit()
+
+    adopt_job(job.id, open_work=lambda: work)
+
+    job.refresh_from_db()
+    assert ('cancel', UNIT) in work.calls
+    assert job.status == 'canceled'
+
+
+@pytest.mark.xfail(strict=True, reason='AAP-96835: canceling a job while it is finalized after a resume leaves it running forever')
+@pytest.mark.django_db
+def test_canceled_job_whose_unit_is_gone_is_finalized_canceled(me_inst):
+    # 4.14 Cancel during the adopter's finalize. The cancel stopped the adoption after the
+    # unit was released but before the status was saved. Today every later adoption finds
+    # no unit and defers, so the job stays running forever.
+    job = Job.objects.create(controller_node=me_inst.hostname, status='running', work_unit_id=UNIT, cancel_flag=True)
+
+    adopt_job(job.id, open_work=lambda: FakeReceptorWork())
+
+    job.refresh_from_db()
+    assert job.status == 'canceled'
+
+
+@pytest.mark.xfail(strict=True, reason='AAP-96836: canceling many orphaned jobs marks them failed one by one instead of canceled')
+@pytest.mark.django_db
+def test_canceled_orphans_of_a_lost_controller_end_canceled():
+    # 4.15 Canceling many orphans at once. Twenty jobs of a dead controller were canceled
+    # while it was down. Today they are marked failed, "Job reaped due to instance shutdown".
+    make_cluster()
+    canceled = [peer_job(work_unit_id=f'unit-{n}', cancel_flag=True) for n in range(20)]
+
+    run_lost_instance_handling(mesh(EXEC))
+
+    statuses = {job.status for job in Job.objects.filter(pk__in=[j.pk for j in canceled])}
+    assert statuses == {'canceled'}
+
+
+@pytest.mark.xfail(strict=True, reason='AAP-96840: canceling a sliced job leaves the orphaned slice failed with its playbook running')
+@pytest.mark.django_db
+def test_canceled_slice_of_a_lost_controller_ends_canceled_and_stops():
+    # 4.21 Canceling a sliced job. The parent was canceled, so the orphaned slice has
+    # cancel_flag set. Today it ends failed and its playbook runs to the end.
+    make_cluster()
+    job = peer_job(job_slice_number=1, job_slice_count=2, cancel_flag=True)
+    work = running_unit()
+
+    run_lost_instance_handling(mesh(EXEC), work=work)
+
+    job.refresh_from_db()
+    assert job.status == 'canceled'
+    assert ('cancel', UNIT) in work.calls
+
+
+# ── A reaped job whose playbook keeps running ─────────────────────────────────
+
+
+def job_of_template_without_simultaneous_runs():
+    return peer_job(job_template=JobTemplate.objects.create(name='same-hosts', allow_simultaneous=False))
+
+
+def job_in_workflow():
+    workflow = WorkflowJob.objects.create(name='outer', status='running')
+    job = peer_job()
+    WorkflowJobNode.objects.create(workflow_job=workflow, job=job)
+    return job
+
+
+def job_in_limited_instance_group():
+    return peer_job(instance_group=InstanceGroup.objects.create(name='limited', max_concurrent_jobs=2))
+
+
+def xfail(bug):
+    return pytest.mark.xfail(strict=True, reason=f'{bug}: a reaped job leaves its playbook running')
+
+
+@pytest.mark.parametrize(
+    'make_job',
+    [
+        pytest.param(job_of_template_without_simultaneous_runs, id='AAP-96832-same-hosts', marks=xfail('AAP-96832')),
+        pytest.param(job_in_workflow, id='AAP-96833-nested-workflow', marks=xfail('AAP-96833')),
+        pytest.param(job_of_template_without_simultaneous_runs, id='AAP-96834-relaunch', marks=xfail('AAP-96834')),
+        pytest.param(job_in_limited_instance_group, id='AAP-96837-instance-group-limit', marks=xfail('AAP-96837')),
+        pytest.param(peer_job, id='AAP-96841-execution-node-capacity', marks=xfail('AAP-96841')),
+        pytest.param(job_in_workflow, id='AAP-96844-workflow-branch', marks=xfail('AAP-96844')),
+        pytest.param(job_in_workflow, id='AAP-96845-set-stats-artifacts', marks=xfail('AAP-96845')),
+    ],
+)
+@pytest.mark.django_db
+def test_reaped_job_does_not_leave_its_playbook_running(make_job):
+    # ctrl-1 is lost while its job runs on receptor-1, which is still up. Today the job is
+    # marked failed and nothing cancels its work unit, so the playbook runs on unobserved:
+    # the next launch overlaps it (4.6, 4.12), it stops counting toward instance group and
+    # execution node limits (4.17, 4.23), and a workflow takes the failure branch for a
+    # playbook that succeeds (4.11, 5.5, 5.6). Either the unit is canceled when the job is
+    # marked failed, or the job stays active until the unit ends.
+    make_cluster()
+    job = make_job()
+    work = running_unit()
+
+    run_lost_instance_handling(mesh(EXEC), work=work)
+
+    assert_playbook_not_left_running(job, work)
+
+
+# ── Waiting jobs of a lost controller ─────────────────────────────────────────
+
+
+@pytest.mark.django_db
+def test_waiting_jobs_of_a_lost_controller_return_to_pending():
+    # 4.5 Jobs waiting when the controller dies. They had not started, so they go back to
+    # the task manager and run once, elsewhere.
+    make_cluster()
+    waiting = Job.objects.create(controller_node=PEER, execution_node=EXEC, status='waiting')
+
+    run_lost_instance_handling(mesh(EXEC))
+
+    waiting.refresh_from_db()
+    assert waiting.status == 'pending'
+    assert waiting.controller_node == ''
+    assert waiting.execution_node == ''
diff --git a/awx/main/tests/functional/tasks/test_scenarios_liveness.py b/awx/main/tests/functional/tasks/test_scenarios_liveness.py
new file mode 100644
index 0000000000..4b55c44c89
--- /dev/null
+++ b/awx/main/tests/functional/tasks/test_scenarios_liveness.py
@@ -0,0 +1,63 @@
+"""Failure scenarios about when a controller is really gone (write-up section 2).
+
+Tests marked xfail reproduce a defect found on devel and assert the behavior it should
+have. Each names its Jira bug; when the bug is fixed the test passes, strict xfail fails
+the run, and the marker comes off.
+"""
+
+from unittest import mock
+
+import pytest
+
+from awx.main.models import Job
+from awx.main.tests.adoption_cluster import EXEC, PEER, UNIT, configure, make_cluster, mesh, run_lost_instance_handling
+
+
+@pytest.fixture(autouse=True)
+def cluster_settings(settings):
+    configure(settings)
+
+
+@pytest.fixture
+def peer_job():
+    """A job controlled by ctrl-1 and running on receptor-1."""
+    return Job.objects.create(controller_node=PEER, execution_node=EXEC, status='running', work_unit_id=UNIT, celery_task_id='peer-task')
+
+
+@pytest.mark.xfail(strict=True, reason='AAP-96824: a controller that misses heartbeats is judged lost while it still runs its job')
+@pytest.mark.django_db
+def test_slow_controller_still_on_the_mesh_keeps_its_job(peer_job):
+    # 2.1 Slow controller. ctrl-1 has missed two heartbeats but is still connected on the
+    # receptor mesh and still streaming its job. Today ctrl-0 judges it lost from last_seen
+    # alone and marks the live job failed.
+    make_cluster()
+
+    run_lost_instance_handling(mesh(PEER, EXEC))
+
+    peer_job.refresh_from_db()
+    assert peer_job.status == 'running'
+
+
+@pytest.mark.django_db
+def test_controller_gone_from_the_mesh_has_its_job_reaped(peer_job):
+    # The case the slow-controller test must not break: a controller that is silent and
+    # gone from the mesh is lost, and its job is no longer reported as running.
+    make_cluster()
+
+    assert run_lost_instance_handling(mesh(EXEC)) == [PEER]
+
+    peer_job.refresh_from_db()
+    assert peer_job.status == 'failed'
+
+
+@pytest.mark.xfail(strict=True, reason='AAP-96848: a slow controller\'s job first notifies failed with no hosts, then successful')
+@pytest.mark.django_db
+def test_slow_controller_job_sends_no_failed_notification(peer_job):
+    # 5.15 Notification content. The reap of a live controller's job sends a failed
+    # notification with no hosts; the job later ends successful and notifies again.
+    make_cluster()
+
+    with mock.patch.object(Job, 'send_notification_templates') as send:
+        run_lost_instance_handling(mesh(PEER, EXEC))
+
+    assert mock.call('failed') not in send.call_args_list
diff --git a/awx/main/tests/unit/tasks/test_adoption_decisions.py b/awx/main/tests/unit/tasks/test_adoption_decisions.py
index 0cd9d8b44a..bf899c9d60 100644
--- a/awx/main/tests/unit/tasks/test_adoption_decisions.py
+++ b/awx/main/tests/unit/tasks/test_adoption_decisions.py
@@ -62,10 +62,11 @@ class TestFindLostInstances:
             pytest.param(ME, 'control', 10_000, 0, False, id='this-instance-is-never-lost-here'),
             pytest.param(EXEC, 'execution', 150, 0, False, id='execution-node-gets-advertisement-period-added'),
             # A controller whose heartbeat is delayed past the grace period looks exactly like
-            # a dead one. The decision has only last_seen to go on.
+            # a dead one. The decision has only last_seen to go on (AAP-96824; the expected
+            # behavior is in test_scenarios_liveness.py).
             pytest.param(PEER, 'control', 125, 0, True, id='slow-controller-is-judged-lost-while-still-running'),
             # The peer heartbeated just now by its own clock, but this instance's clock runs
-            # five minutes ahead.
+            # five minutes ahead (AAP-96827; the expected behavior is in test_adoption_clock.py).
             pytest.param(PEER, 'control', 0, 300, True, id='clock-skew-on-the-judge-makes-a-healthy-peer-look-lost'),
         ],
     )
@@ -178,7 +179,8 @@ class TestAdoptionDeadline:
             pytest.param(None, 299, False, id='no-events-and-recent-start-is-within-deadline'),
             pytest.param(None, None, False, id='job-with-no-events-or-start-is-never-past-deadline'),
             # A playbook task that runs silently longer than the timeout (a long pause, a slow
-            # package install) looks the same as a dead one.
+            # package install) looks the same as a dead one (AAP-96829; the expected behavior
+            # is in test_scenarios_adoption.py).
             pytest.param(TIMEOUT + 1, TIMEOUT + 60, True, id='quiet-job-is-past-deadline-even-while-healthy'),
         ],
     )