Feat: Expose step run errors on context

examples/on_failure/worker.py

@@ -25,6 +25,10 @@ def step1(self, context: Context) -> None:
     def on_failure(self, context: Context) -> dict[str, str]:
         # 👀 we can do things like perform cleanup logic
         # or notify a user here
+
+        # 👀 Fetch the errors from upstream step runs from the context
+        print(context.step_run_errors())
+
         return {"status": "success"}
 
 

hatchet_sdk/context/context.py

@@ -376,6 +376,16 @@ def child_key(self) -> str | None:
     def parent_workflow_run_id(self) -> str | None:
         return self.action.parent_workflow_run_id
 
+    def step_run_errors(self) -> dict[str, str]:
+        errors = cast(dict[str, str], self.data.get("step_run_errors", {}))
+
+        if not errors:
+            logger.error(
+                "No step run errors found. `context.step_run_errors` is intended to be run in an on-failure step, and will only work on engine versions more recent than v0.53.10"
+            )
+
+        return errors
+
     def fetch_run_failures(self) -> list[dict[str, StrictStr]]:
         data = self.rest_client.workflow_run_get(self.action.workflow_run_id)
         other_job_runs = [

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "hatchet-sdk"
-version = "0.43.3"
+version = "0.44.0"
 description = ""
 authors = ["Alexander Belanger <alexander@hatchet.run>"]
 readme = "README.md"