[Python, TS] Fix: Cancellation Propagation (#2865)

* feat: don't send fail / success event on cancelled

* feat: basic bounded dict to handle cancellations without leaking memory

* chore: changelog

* fix: rm redundant line

* chore: ts ver

* chore: changelogs

* fix: add default

* fix: early return

Author: mrkaye97

sdks/python/CHANGELOG.md

@@ -5,6 +5,12 @@ All notable changes to Hatchet's Python SDK will be documented in this changelog
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [1.22.11] - 2026-01-27
+
+### Changed
+
+- Improves handling of cancellations for tasks to limit how often tasks receive a cancellation but then are marked as succeeded anyways.
+
 ## [1.22.10] - 2026-01-26
 
 ### Added

sdks/python/hatchet_sdk/utils/cache.py

@@ -0,0 +1,20 @@
+from collections import OrderedDict
+from typing import TypeVar
+
+K = TypeVar("K")
+V = TypeVar("V")
+
+
+class BoundedDict(OrderedDict[K, V]):
+    def __init__(self, maxsize: int):
+        super().__init__()
+        self.maxsize = maxsize
+
+    def __setitem__(self, key: K, value: V) -> None:
+        if key in self:
+            self.move_to_end(key)
+
+        super().__setitem__(key, value)
+
+        if len(self) > self.maxsize:
+            self.popitem(last=False)

sdks/python/hatchet_sdk/worker/runner/runner.py

@@ -49,6 +49,7 @@
 from hatchet_sdk.runnables.task import Task
 from hatchet_sdk.runnables.types import R, TWorkflowInput
 from hatchet_sdk.serde import HATCHET_PYDANTIC_SENTINEL
+from hatchet_sdk.utils.cache import BoundedDict
 from hatchet_sdk.utils.serde import remove_null_unicode_character
 from hatchet_sdk.utils.typing import DataclassInstance
 from hatchet_sdk.worker.action_listener_process import ActionEvent
@@ -86,6 +87,7 @@ def __init__(
         self.slots = slots
         self.tasks: dict[ActionKey, asyncio.Task[Any]] = {}  # Store run ids and futures
         self.contexts: dict[ActionKey, Context] = {}  # Store run ids and contexts
+        self.cancellations = BoundedDict[str, bool](maxsize=1000)
         self.action_registry = action_registry or {}
 
         self.event_queue = event_queue
@@ -156,8 +158,9 @@ def step_run_callback(
     ) -> Callable[[asyncio.Task[Any]], None]:
         def inner_callback(task: asyncio.Task[Any]) -> None:
             self.cleanup_run_id(action.key)
+            was_cancelled = self.cancellations.pop(action.key, False)
 
-            if task.cancelled():
+            if was_cancelled or task.cancelled():
                 return
 
             try:
@@ -348,6 +351,9 @@ def cleanup_run_id(self, key: ActionKey) -> None:
             del self.threads[key]
 
         if key in self.contexts:
+            if self.contexts[key].exit_flag:
+                self.cancellations[key] = True
+
             del self.contexts[key]
 
     @overload
@@ -467,6 +473,7 @@ async def handle_cancel_action(self, action: Action) -> None:
             # call cancel to signal the context to stop
             if key in self.contexts:
                 self.contexts[key]._set_cancellation_flag()
+                self.cancellations[key] = True
 
             await asyncio.sleep(1)
 

sdks/python/pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "hatchet-sdk"
-version = "1.22.10"
+version = "1.22.11"
 description = "This is the official Python SDK for Hatchet, a distributed, fault-tolerant task queue. The SDK allows you to easily integrate Hatchet's task scheduling and workflow orchestration capabilities into your Python applications."
 authors = [
   "Alexander Belanger <[email protected]>",

sdks/typescript/CHANGELOG.md

@@ -4,3 +4,9 @@ All notable changes to Hatchet's TypeScript SDK will be documented in this chang
 
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
+
+## [1.10.6] - 2026-01-27
+
+### Changed
+
+- Improves handling of cancellations for tasks to limit how often tasks receive a cancellation but then are marked as succeeded anyways.

sdks/typescript/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@hatchet-dev/typescript-sdk",
-  "version": "1.10.5",
+  "version": "1.10.6",
   "description": "Background task orchestration & visibility for developers",
   "types": "dist/index.d.ts",
   "files": [

sdks/typescript/src/v1/client/worker/worker-internal.ts

@@ -550,9 +550,13 @@ export class V1Worker {
       };
 
       const success = async (result: any) => {
-        this.logger.info(`Step run ${action.stepRunId} succeeded`);
-
         try {
+          if (context.cancelled) {
+            return;
+          }
+
+          this.logger.info(`Step run ${action.stepRunId} succeeded`);
+
           // Send the action event to the dispatcher
           const event = this.getStepActionEvent(
             action,
@@ -595,15 +599,19 @@ export class V1Worker {
       };
 
       const failure = async (error: any) => {
-        this.logger.error(`Step run ${action.stepRunId} failed: ${error.message}`);
-
-        if (error.stack) {
-          this.logger.error(error.stack);
-        }
-
         const shouldNotRetry = error instanceof NonRetryableError;
 
         try {
+          if (context.cancelled) {
+            return;
+          }
+
+          this.logger.error(`Step run ${action.stepRunId} failed: ${error.message}`);
+
+          if (error.stack) {
+            this.logger.error(error.stack);
+          }
+
           // Send the action event to the dispatcher
           const event = this.getStepActionEvent(
             action,