diff --git a/changelogs/fragments/65931-json-callback-non-lockstep-output.yml b/changelogs/fragments/65931-json-callback-non-lockstep-output.yml
new file mode 100644
index 0000000..c6854ae
--- /dev/null
+++ b/changelogs/fragments/65931-json-callback-non-lockstep-output.yml
@@ -0,0 +1,4 @@
+bugfixes:
+- json callback - Fix host result to task references in the resultant JSON
+  output for non-lockstep strategy plugins such as free
+  (https://github.com/ansible/ansible/issues/65931)
diff --git a/plugins/callback/json.py b/plugins/callback/json.py
index 0009ac0..a69f38f 100644
--- a/plugins/callback/json.py
+++ b/plugins/callback/json.py
@@ -25,6 +25,11 @@ DOCUMENTATION = '''
           - key: show_custom_stats
             section: defaults
         type: bool
+    notes:
+      - When using a strategy such as free, host_pinned, or a custom strategy, host results will
+        be added to new task results in ``.plays[].tasks[]``. As such, there will exist duplicate
+        task objects indicated by duplicate task IDs at ``.plays[].tasks[].task.id``, each with an
+        individual host result for the task.
 '''
 
 import datetime
@@ -33,10 +38,14 @@ import json
 from functools import partial
 
 from ansible.inventory.host import Host
+from ansible.module_utils._text import to_text
 from ansible.parsing.ajson import AnsibleJSONEncoder
 from ansible.plugins.callback import CallbackBase
 
 
+LOCKSTEP_CALLBACKS = frozenset(('linear', 'debug'))
+
+
 def current_time():
     return '%sZ' % datetime.datetime.utcnow().isoformat()
 
@@ -49,12 +58,15 @@ class CallbackModule(CallbackBase):
     def __init__(self, display=None):
         super(CallbackModule, self).__init__(display)
         self.results = []
+        self._task_map = {}
+        self._is_lockstep = False
 
     def _new_play(self, play):
+        self._is_lockstep = play.strategy in LOCKSTEP_CALLBACKS
         return {
             'play': {
                 'name': play.get_name(),
-                'id': str(play._uuid),
+                'id': to_text(play._uuid),
                 'duration': {
                     'start': current_time()
                 }
@@ -66,7 +78,7 @@ class CallbackModule(CallbackBase):
         return {
             'task': {
                 'name': task.get_name(),
-                'id': str(task._uuid),
+                'id': to_text(task._uuid),
                 'duration': {
                     'start': current_time()
                 }
@@ -74,13 +86,32 @@ class CallbackModule(CallbackBase):
             'hosts': {}
         }
 
+    def _find_result_task(self, host, task):
+        key = (host.get_name(), task._uuid)
+        return self._task_map.get(
+            key,
+            self.results[-1]['tasks'][-1]
+        )
+
     def v2_playbook_on_play_start(self, play):
         self.results.append(self._new_play(play))
 
+    def v2_runner_on_start(self, host, task):
+        if self._is_lockstep:
+            return
+        key = (host.get_name(), task._uuid)
+        task_result = self._new_task(task)
+        self._task_map[key] = task_result
+        self.results[-1]['tasks'].append(task_result)
+
     def v2_playbook_on_task_start(self, task, is_conditional):
+        if not self._is_lockstep:
+            return
         self.results[-1]['tasks'].append(self._new_task(task))
 
     def v2_playbook_on_handler_task_start(self, task):
+        if not self._is_lockstep:
+            return
         self.results[-1]['tasks'].append(self._new_task(task))
 
     def _convert_host_to_name(self, key):
@@ -118,14 +149,22 @@ class CallbackModule(CallbackBase):
         """This function is used as a partial to add failed/skipped info in a single method"""
         host = result._host
         task = result._task
-        task_result = result._result.copy()
-        task_result.update(on_info)
-        task_result['action'] = task.action
-        self.results[-1]['tasks'][-1]['hosts'][host.name] = task_result
+
+        result_copy = result._result.copy()
+        result_copy.update(on_info)
+        result_copy['action'] = task.action
+
+        task_result = self._find_result_task(host, task)
+
+        task_result['hosts'][host.name] = result_copy
         end_time = current_time()
-        self.results[-1]['tasks'][-1]['task']['duration']['end'] = end_time
+        task_result['task']['duration']['end'] = end_time
         self.results[-1]['play']['duration']['end'] = end_time
 
+        if not self._is_lockstep:
+            key = (host.get_name(), task._uuid)
+            del self._task_map[key]
+
     def __getattribute__(self, name):
         """Return ``_record_task_result`` partial with a dict containing skipped/failed if necessary"""
         if name not in ('v2_runner_on_ok', 'v2_runner_on_failed', 'v2_runner_on_unreachable', 'v2_runner_on_skipped'):