diff --git a/tron/core/action.py b/tron/core/action.py
index 03b9e6419..33b18c355 100644
--- a/tron/core/action.py
+++ b/tron/core/action.py
@@ -115,6 +115,13 @@ def serialize_namedtuple(obj):
             return obj
 
         try:
+            # NOTE: you'll notice that there's a lot of get() accesses of state_data for
+            # pretty common fields - this is because ActionCommandConfig is used by more
+            # than one type of ActionRun (Kubernetes, Mesos, SSH) and these generally look
+            # different. Alternatively, some of these fields are used by KubernetesActionRun
+            # but are relatively new and older runs do not have data for them.
+            # Once we get rid of the SSH and Mesos code as well as older runs in DynamoDB,
+            # we'll likely be able to clean this up.
             return json.dumps(
                 {
                     "command": state_data["command"],
@@ -124,30 +131,35 @@ def serialize_namedtuple(obj):
                     "cap_add": state_data["cap_add"],
                     "cap_drop": state_data["cap_drop"],
                     "constraints": [
-                        serialize_namedtuple(constraint) for constraint in state_data["constraints"]
+                        serialize_namedtuple(constraint) for constraint in state_data.get("constraints", [])
                     ],  # convert each ConfigConstraint to dictionary, so it would be a list of dicts
                     "docker_image": state_data["docker_image"],
                     "docker_parameters": [
-                        serialize_namedtuple(parameter) for parameter in state_data["docker_parameters"]
+                        serialize_namedtuple(parameter) for parameter in state_data.get("docker_parameters", [])
                     ],
-                    "env": state_data["env"],
-                    "secret_env": {key: serialize_namedtuple(val) for key, val in state_data["secret_env"].items()},
-                    "secret_volumes": [serialize_namedtuple(volume) for volume in state_data["secret_volumes"]],
+                    "env": state_data.get("env", {}),
+                    "secret_env": {
+                        key: serialize_namedtuple(val) for key, val in state_data.get("secret_env", {}).items()
+                    },
+                    "secret_volumes": [serialize_namedtuple(volume) for volume in state_data.get("secret_volumes", [])],
                     "projected_sa_volumes": [
-                        serialize_namedtuple(volume) for volume in state_data["projected_sa_volumes"]
+                        serialize_namedtuple(volume) for volume in state_data.get("projected_sa_volumes", [])
                     ],
                     "field_selector_env": {
-                        key: serialize_namedtuple(val) for key, val in state_data["field_selector_env"].items()
+                        key: serialize_namedtuple(val) for key, val in state_data.get("field_selector_env", {}).items()
                     },
-                    "extra_volumes": [serialize_namedtuple(volume) for volume in state_data["extra_volumes"]],
-                    "node_selectors": state_data["node_selectors"],
-                    "node_affinities": [serialize_namedtuple(affinity) for affinity in state_data["node_affinities"]],
-                    "labels": state_data["labels"],
-                    "annotations": state_data["annotations"],
-                    "service_account_name": state_data["service_account_name"],
-                    "ports": state_data["ports"],
+                    "extra_volumes": [serialize_namedtuple(volume) for volume in state_data.get("extra_volumes", [])],
+                    "node_selectors": state_data.get("node_selectors", {}),
+                    "node_affinities": [
+                        serialize_namedtuple(affinity) for affinity in state_data.get("node_affinities", [])
+                    ],
+                    "labels": state_data.get("labels", {}),
+                    "annotations": state_data.get("annotations", {}),
+                    "service_account_name": state_data.get("service_account_name"),
+                    "ports": state_data.get("ports", []),
                     "topology_spread_constraints": [
-                        serialize_namedtuple(constraint) for constraint in state_data["topology_spread_constraints"]
+                        serialize_namedtuple(constraint)
+                        for constraint in state_data.get("topology_spread_constraints", [])
                     ],
                 }
             )
diff --git a/tron/core/actionrun.py b/tron/core/actionrun.py
index a37a83b15..efec9bcdf 100644
--- a/tron/core/actionrun.py
+++ b/tron/core/actionrun.py
@@ -185,8 +185,13 @@ def to_json(state_data: dict) -> Optional[str]:
                     "end_time": state_data["end_time"].isoformat() if state_data["end_time"] else None,
                     "rendered_command": state_data["rendered_command"],
                     "exit_status": state_data["exit_status"],
-                    "mesos_task_id": state_data["mesos_task_id"],
-                    "kubernetes_task_id": state_data["kubernetes_task_id"],
+                    # NOTE: mesos_task_id can be deleted once we delete all Mesos
+                    # code and run data - and kubernetes_task_id can then be
+                    # accessed unconditionally :)
+                    # (see note in ActionCommandConfig::to_json() for more
+                    # information about why we do this)
+                    "mesos_task_id": state_data.get("mesos_task_id"),
+                    "kubernetes_task_id": state_data.get("kubernetes_task_id"),
                 }
             )
         except KeyError:
@@ -811,12 +816,12 @@ def to_json(state_data: dict) -> Optional[str]:
                     "job_run_id": state_data["job_run_id"],
                     "action_name": state_data["action_name"],
                     "state": state_data["state"],
-                    "original_command": state_data["original_command"],
+                    "original_command": state_data.get("original_command"),
                     "start_time": state_data["start_time"].isoformat() if state_data["start_time"] else None,
                     "end_time": state_data["end_time"].isoformat() if state_data["end_time"] else None,
                     "node_name": state_data["node_name"],
                     "exit_status": state_data["exit_status"],
-                    "attempts": [ActionRunAttempt.to_json(attempt) for attempt in state_data["attempts"]],
+                    "attempts": [ActionRunAttempt.to_json(attempt) for attempt in state_data.get("attempts", [])],
                     "retries_remaining": state_data["retries_remaining"],
                     "retries_delay": (
                         state_data["retries_delay"].total_seconds() if state_data["retries_delay"] is not None else None