diff --git a/lib/ansible/executor/task_executor.py b/lib/ansible/executor/task_executor.py
index ebef9cbfd15..ce8a4fcf1db 100644
--- a/lib/ansible/executor/task_executor.py
+++ b/lib/ansible/executor/task_executor.py
@@ -866,6 +866,7 @@ class TaskExecutor:
             task_vars = self._job_vars
 
         async_jid = result.get('ansible_job_id')
+        results_file = result.get('results_file')
         if async_jid is None:
             return dict(failed=True, msg="No job id was returned by the async task")
 
@@ -875,7 +876,7 @@ class TaskExecutor:
 
         async_task = Task.load(dict(
             action='async_status',
-            args={'jid': async_jid},
+            args={'jid': results_file or async_jid},
             check_mode=self._task.check_mode,
             environment=self._task.environment,
         ))
@@ -947,7 +948,7 @@ class TaskExecutor:
             cleanup_task = Task.load(
                 {
                     'async_status': {
-                        'jid': async_jid,
+                        'jid': results_file or async_jid,
                         'mode': 'cleanup',
                     },
                     'check_mode': self._task.check_mode,
diff --git a/lib/ansible/modules/async_status.py b/lib/ansible/modules/async_status.py
index 0a4eeb53ac2..5f47b66096b 100644
--- a/lib/ansible/modules/async_status.py
+++ b/lib/ansible/modules/async_status.py
@@ -18,6 +18,8 @@ options:
   jid:
     description:
     - Job or task identifier
+    - May be the exact value of C(ansible_job_id) or the full path to the results file provided by C(results_file)
+    - Support for this value taking a path was added in 2.19.
     type: str
     required: true
   mode:
@@ -73,7 +75,7 @@ EXAMPLES = r"""
 
 - name: Clean up async file
   ansible.builtin.async_status:
-    jid: '{{ dnf_sleeper.ansible_job_id }}'
+    jid: '{{ dnf_sleeper.results_file }}'
     mode: cleanup
 """
 
@@ -88,6 +90,12 @@ finished:
   returned: always
   type: int
   sample: 1
+results_file:
+  description: Path to the async results file, which is a concatenation of the
+               async dir and the C(ansible_job_id)
+  returned: always
+  type: str
+  sample: '/home/user/.async_status/360874038559.4169'
 started:
   description: Whether the asynchronous job has started (V(1)) or not (V(0))
   returned: always
@@ -132,7 +140,10 @@ def main():
     async_dir = module.params['_async_dir']
 
     # setup logging directory
-    log_path = os.path.join(async_dir, jid)
+    if os.path.isabs(jid):
+        log_path = jid
+    else:
+        log_path = os.path.join(async_dir, jid)
 
     if not os.path.exists(log_path):
         module.fail_json(msg="could not find job", ansible_job_id=jid, started=1, finished=1)
diff --git a/lib/ansible/plugins/action/async_status.py b/lib/ansible/plugins/action/async_status.py
index a0fe11eb59d..e4882fdced0 100644
--- a/lib/ansible/plugins/action/async_status.py
+++ b/lib/ansible/plugins/action/async_status.py
@@ -3,6 +3,8 @@
 
 from __future__ import annotations
 
+import os
+
 from ansible.plugins.action import ActionBase
 from ansible.utils.vars import merge_hash
 
@@ -36,8 +38,13 @@ class ActionModule(ActionBase):
         mode = new_module_args["mode"]
 
         results['ansible_job_id'] = jid
-        async_dir = self._get_async_dir()
-        log_path = self._connection._shell.join_path(async_dir, jid)
+        if os.path.isabs(jid):
+            log_path = jid
+            new_module_args['_async_dir'] = ''
+        else:
+            async_dir = self._get_async_dir()
+            log_path = self._connection._shell.join_path(async_dir, jid)
+            new_module_args['_async_dir'] = async_dir
 
         if mode == 'cleanup':
             results['erased'] = log_path
@@ -45,7 +52,6 @@ class ActionModule(ActionBase):
             results['results_file'] = log_path
             results['started'] = 1
 
-        new_module_args['_async_dir'] = async_dir
         results = merge_hash(results, self._execute_module(module_name='ansible.legacy.async_status', task_vars=task_vars, module_args=new_module_args))
 
         return results
diff --git a/lib/ansible/plugins/action/gather_facts.py b/lib/ansible/plugins/action/gather_facts.py
index 31210ec724d..192843b6f41 100644
--- a/lib/ansible/plugins/action/gather_facts.py
+++ b/lib/ansible/plugins/action/gather_facts.py
@@ -135,7 +135,12 @@ class ActionModule(ActionBase):
 
             while jobs:
                 for module in jobs:
-                    poll_args = {'jid': jobs[module]['ansible_job_id'], '_async_dir': os.path.dirname(jobs[module]['results_file'])}
+                    jid = jobs[module]['ansible_job_id']
+                    if os.path.isabs(jid):
+                        async_dir = ''
+                    else:
+                        async_dir = os.path.dirname(jobs[module]['results_file'])
+                    poll_args = {'jid': jid, '_async_dir': async_dir}
                     res = self._execute_module(module_name='ansible.legacy.async_status', module_args=poll_args, task_vars=task_vars, wrap_async=False)
                     if res.get('finished', 0) == 1:
                         if res.get('failed', False):
diff --git a/test/integration/targets/async/tasks/main.yml b/test/integration/targets/async/tasks/main.yml
index 65182070553..0493e43df1e 100644
--- a/test/integration/targets/async/tasks/main.yml
+++ b/test/integration/targets/async/tasks/main.yml
@@ -122,7 +122,7 @@
 - name: assert task failed correctly
   assert:
     that:
-    - async_result.ansible_job_id is match('j\d+\.\d+')
+    - async_result.ansible_job_id|basename is match('j\d+\.\d+')
     - async_result.finished == 1
     - async_result is finished
     - async_result is not changed
@@ -140,7 +140,7 @@
 - name: validate response
   assert:
     that:
-    - async_result.ansible_job_id is match('j\d+\.\d+')
+    - async_result.ansible_job_id|basename is match('j\d+\.\d+')
     - async_result.finished == 1
     - async_result is finished
     - async_result.changed == false
@@ -159,7 +159,7 @@
 - name: validate response
   assert:
     that:
-    - async_result.ansible_job_id is match('j\d+\.\d+')
+    - async_result.ansible_job_id|basename is match('j\d+\.\d+')
     - async_result.finished == 1
     - async_result is finished
     - async_result.changed == true
@@ -176,7 +176,7 @@
 - name: validate response
   assert:
     that:
-    - async_result.ansible_job_id is match('j\d+\.\d+')
+    - async_result.ansible_job_id|basename is match('j\d+\.\d+')
     - async_result.finished == 1
     - async_result is finished
     - async_result.changed == true
diff --git a/test/integration/targets/async_fail/tasks/main.yml b/test/integration/targets/async_fail/tasks/main.yml
index 24cea1d581d..c2ff89c7cbd 100644
--- a/test/integration/targets/async_fail/tasks/main.yml
+++ b/test/integration/targets/async_fail/tasks/main.yml
@@ -28,7 +28,7 @@
 - name: validate that by the end of the retry interval, we succeeded
   assert:
     that:
-    - async_result.ansible_job_id is match('j\d+\.\d+')
+    - async_result.ansible_job_id|basename is match('j\d+\.\d+')
     - async_result.finished == 1
     - async_result is finished
     - async_result is changed
diff --git a/test/integration/targets/callback_default/runme.sh b/test/integration/targets/callback_default/runme.sh
index 4dab4f40ae8..114748739e1 100755
--- a/test/integration/targets/callback_default/runme.sh
+++ b/test/integration/targets/callback_default/runme.sh
@@ -136,8 +136,8 @@ set -e
 # Check for async output
 # NOTE: regex to match 1 or more digits works for both BSD and GNU grep
 ansible-playbook -i inventory test_async.yml 2>&1 | tee async_test.out
-grep "ASYNC OK .* jid=j[0-9]\{1,\}" async_test.out
-grep "ASYNC FAILED .* jid=j[0-9]\{1,\}" async_test.out
+grep "ASYNC OK .* jid=.*j[0-9]\{1,\}" async_test.out
+grep "ASYNC FAILED .* jid=.*j[0-9]\{1,\}" async_test.out
 rm -f async_test.out
 
 # Hide skipped