From 9e7b50343682bef33c3a00c953f868c5b7dad229 Mon Sep 17 00:00:00 2001
From: kartikeya <67143288+kartikeya-27@users.noreply.github.com>
Date: Fri, 24 Jul 2026 00:56:36 +0530
Subject: [PATCH 01/11] feat(java): take a WorkflowCondition on the step
 builder

Adds a WorkflowCondition enum and a Step.Builder.condition overload; the String overload stays.
---
 .../org/byteveda/taskito/workflows/Step.java  |  9 +++++
 .../taskito/workflows/WorkflowCondition.java  | 34 +++++++++++++++++++
 .../workflows/WorkflowConditionTest.java      | 12 +++++++
 3 files changed, 55 insertions(+)
 create mode 100644 sdks/java/src/main/java/org/byteveda/taskito/workflows/WorkflowCondition.java

diff --git a/sdks/java/src/main/java/org/byteveda/taskito/workflows/Step.java b/sdks/java/src/main/java/org/byteveda/taskito/workflows/Step.java
index 8c6b9e68..c1bd48ec 100644
--- a/sdks/java/src/main/java/org/byteveda/taskito/workflows/Step.java
+++ b/sdks/java/src/main/java/org/byteveda/taskito/workflows/Step.java
@@ -160,6 +160,15 @@ public Builder condition(String condition) {
             return this;
         }
 
+        /**
+         * Type-safe variant of {@link #condition(String)} — run this step only when
+         * {@code condition} holds. Prefer this over the string overload.
+         */
+        public Builder condition(WorkflowCondition condition) {
+            this.condition = condition == null ? null : condition.wire();
+            return this;
+        }
+
         /** Run this step only if every predecessor completed (the default). */
         public Builder onSuccess() {
             return condition("on_success");
diff --git a/sdks/java/src/main/java/org/byteveda/taskito/workflows/WorkflowCondition.java b/sdks/java/src/main/java/org/byteveda/taskito/workflows/WorkflowCondition.java
new file mode 100644
index 00000000..c593bba0
--- /dev/null
+++ b/sdks/java/src/main/java/org/byteveda/taskito/workflows/WorkflowCondition.java
@@ -0,0 +1,34 @@
+package org.byteveda.taskito.workflows;
+
+import java.util.Locale;
+import org.byteveda.taskito.errors.SerializationException;
+
+/**
+ * When a step runs, based on its predecessors' outcomes. The wire form is the lowercase
+ * snake_case name, shared across SDKs.
+ */
+public enum WorkflowCondition {
+    /** Every predecessor completed — the default. */
+    ON_SUCCESS,
+    /** At least one predecessor failed — an error-handler branch. */
+    ON_FAILURE,
+    /** Regardless of predecessor outcomes, once they settle. */
+    ALWAYS;
+
+    /** Lowercase snake_case wire form ({@code "on_success"}/{@code "on_failure"}/{@code "always"}). */
+    public String wire() {
+        return name().toLowerCase(Locale.ROOT);
+    }
+
+    /** Parse a wire form ({@code "on_success"}/{@code "on_failure"}/{@code "always"}). */
+    public static WorkflowCondition fromWire(String wire) {
+        if (wire == null) {
+            throw new SerializationException("workflow condition is null");
+        }
+        try {
+            return valueOf(wire.toUpperCase(Locale.ROOT));
+        } catch (IllegalArgumentException e) {
+            throw new SerializationException("unknown workflow condition: " + wire, e);
+        }
+    }
+}
diff --git a/sdks/java/src/test/java/org/byteveda/taskito/workflows/WorkflowConditionTest.java b/sdks/java/src/test/java/org/byteveda/taskito/workflows/WorkflowConditionTest.java
index 7365085e..66197862 100644
--- a/sdks/java/src/test/java/org/byteveda/taskito/workflows/WorkflowConditionTest.java
+++ b/sdks/java/src/test/java/org/byteveda/taskito/workflows/WorkflowConditionTest.java
@@ -111,6 +111,18 @@ void alwaysRunsAfterFailure(@TempDir Path dir) throws Exception {
         }
     }
 
+    @Test
+    void conditionEnumOverloadMatchesStringForm() {
+        assertEquals("on_success", WorkflowCondition.ON_SUCCESS.wire());
+        assertEquals(WorkflowCondition.ALWAYS, WorkflowCondition.fromWire("always"));
+        Step viaEnum = Step.of("s", RECOVER, "x")
+                .condition(WorkflowCondition.ON_FAILURE)
+                .build();
+        Step viaString = Step.of("s", RECOVER, "x").onFailure().build();
+        assertEquals(viaString.condition, viaEnum.condition);
+        assertEquals("on_failure", viaEnum.condition);
+    }
+
     @Test
     @Timeout(30)
     void callableConditionRunsWhenTrue(@TempDir Path dir) throws Exception {

From 0cfac36bb9057b90dcf4bcca3109796f78ef04c4 Mon Sep 17 00:00:00 2001
From: kartikeya <67143288+kartikeya-27@users.noreply.github.com>
Date: Fri, 24 Jul 2026 00:56:53 +0530
Subject: [PATCH 02/11] feat(java): take a StorageBackend on the client builder

Adds a StorageBackend enum and a Builder.backend overload; the String overload stays.
---
 .../java/org/byteveda/taskito/Taskito.java    |  9 +++++
 .../taskito/model/StorageBackend.java         | 34 +++++++++++++++++++
 .../taskito/model/ClosedSetEnumTest.java      |  9 +++++
 3 files changed, 52 insertions(+)
 create mode 100644 sdks/java/src/main/java/org/byteveda/taskito/model/StorageBackend.java

diff --git a/sdks/java/src/main/java/org/byteveda/taskito/Taskito.java b/sdks/java/src/main/java/org/byteveda/taskito/Taskito.java
index c1d59b01..0fc73aaf 100644
--- a/sdks/java/src/main/java/org/byteveda/taskito/Taskito.java
+++ b/sdks/java/src/main/java/org/byteveda/taskito/Taskito.java
@@ -34,6 +34,7 @@
 import org.byteveda.taskito.model.PeriodicInfo;
 import org.byteveda.taskito.model.QueueStats;
 import org.byteveda.taskito.model.ReplayEntry;
+import org.byteveda.taskito.model.StorageBackend;
 import org.byteveda.taskito.model.Subscription;
 import org.byteveda.taskito.model.TaskLog;
 import org.byteveda.taskito.model.TaskLogLevel;
@@ -659,6 +660,14 @@ public Builder backend(String backend) {
             return this;
         }
 
+        /**
+         * Type-safe variant of {@link #backend(String)}. Prefer this over the string overload.
+         */
+        public Builder backend(StorageBackend backend) {
+            options.put("backend", backend == null ? null : backend.wire());
+            return this;
+        }
+
         /** Connection string: a file path for SQLite, a URL for Postgres/Redis. */
         public Builder url(String dsn) {
             options.put("dsn", dsn);
diff --git a/sdks/java/src/main/java/org/byteveda/taskito/model/StorageBackend.java b/sdks/java/src/main/java/org/byteveda/taskito/model/StorageBackend.java
new file mode 100644
index 00000000..3b60ad24
--- /dev/null
+++ b/sdks/java/src/main/java/org/byteveda/taskito/model/StorageBackend.java
@@ -0,0 +1,34 @@
+package org.byteveda.taskito.model;
+
+import java.util.Locale;
+import org.byteveda.taskito.errors.SerializationException;
+
+/**
+ * Storage backend a {@link org.byteveda.taskito.Taskito} client opens. The wire form is the
+ * lowercase name, shared across SDKs.
+ */
+public enum StorageBackend {
+    /** Brokerless SQLite file store — the default. */
+    SQLITE,
+    /** PostgreSQL. */
+    POSTGRES,
+    /** Redis. */
+    REDIS;
+
+    /** Lowercase wire form passed to the native layer. */
+    public String wire() {
+        return name().toLowerCase(Locale.ROOT);
+    }
+
+    /** Parse a wire form ({@code "sqlite"}/{@code "postgres"}/{@code "redis"}). */
+    public static StorageBackend fromWire(String wire) {
+        if (wire == null) {
+            throw new SerializationException("storage backend is null");
+        }
+        try {
+            return valueOf(wire.toUpperCase(Locale.ROOT));
+        } catch (IllegalArgumentException e) {
+            throw new SerializationException("unknown storage backend: " + wire, e);
+        }
+    }
+}
diff --git a/sdks/java/src/test/java/org/byteveda/taskito/model/ClosedSetEnumTest.java b/sdks/java/src/test/java/org/byteveda/taskito/model/ClosedSetEnumTest.java
index fce7ee3c..deb5614a 100644
--- a/sdks/java/src/test/java/org/byteveda/taskito/model/ClosedSetEnumTest.java
+++ b/sdks/java/src/test/java/org/byteveda/taskito/model/ClosedSetEnumTest.java
@@ -24,10 +24,19 @@ void taskLogLevelRoundTripsItsWireForm() {
         assertEquals(TaskLogLevel.CRITICAL, TaskLogLevel.fromWire("critical"));
     }
 
+    @Test
+    void storageBackendRoundTripsItsWireForm() {
+        assertEquals("sqlite", StorageBackend.SQLITE.wire());
+        assertEquals("postgres", StorageBackend.POSTGRES.wire());
+        assertEquals(StorageBackend.REDIS, StorageBackend.fromWire("redis"));
+        assertEquals(StorageBackend.SQLITE, StorageBackend.fromWire("SQLITE"));
+    }
+
     @Test
     void unknownWireFormIsRejected() {
         assertThrows(SerializationException.class, () -> DispatchOrder.fromWire("sideways"));
         assertThrows(SerializationException.class, () -> TaskLogLevel.fromWire("verbose"));
         assertThrows(SerializationException.class, () -> TaskLogLevel.fromWire(null));
+        assertThrows(SerializationException.class, () -> StorageBackend.fromWire("mysql"));
     }
 }

From 3576cce6ed21c8b46822cb0f3076aa77284a3324 Mon Sep 17 00:00:00 2001
From: kartikeya <67143288+kartikeya-27@users.noreply.github.com>
Date: Fri, 24 Jul 2026 00:57:03 +0530
Subject: [PATCH 03/11] feat(python): type workflow condition/status compares

Adds WorkflowCondition; the builder coerces it, and the tracker/saga compare node and run status against the enums instead of string literals.
---
 sdks/python/taskito/workflows/__init__.py     |  2 +
 sdks/python/taskito/workflows/builder.py      | 38 +++++++++----------
 .../taskito/workflows/saga/orchestrator.py    |  3 +-
 sdks/python/taskito/workflows/tracker/dag.py  | 30 +++++++++------
 sdks/python/taskito/workflows/types.py        | 13 +++++++
 5 files changed, 53 insertions(+), 33 deletions(-)

diff --git a/sdks/python/taskito/workflows/__init__.py b/sdks/python/taskito/workflows/__init__.py
index e00eaaf4..df813e41 100644
--- a/sdks/python/taskito/workflows/__init__.py
+++ b/sdks/python/taskito/workflows/__init__.py
@@ -16,6 +16,7 @@
     GateAction,
     NodeSnapshot,
     NodeStatus,
+    WorkflowCondition,
     WorkflowState,
     WorkflowStatus,
 )
@@ -28,6 +29,7 @@
     "NodeSnapshot",
     "NodeStatus",
     "Workflow",
+    "WorkflowCondition",
     "WorkflowContext",
     "WorkflowProxy",
     "WorkflowRun",
diff --git a/sdks/python/taskito/workflows/builder.py b/sdks/python/taskito/workflows/builder.py
index 12889b49..7da80222 100644
--- a/sdks/python/taskito/workflows/builder.py
+++ b/sdks/python/taskito/workflows/builder.py
@@ -14,7 +14,7 @@
 from taskito.enums import coerce_enum
 
 from . import analysis as _analysis
-from .types import DiagramFormat, FanStrategy, GateAction
+from .types import DiagramFormat, FanStrategy, GateAction, WorkflowCondition
 from .visualization import nodes_and_edges_from_steps, render_dot, render_mermaid
 
 if TYPE_CHECKING:
@@ -33,7 +33,6 @@ class HasTaskName(Protocol):
     def _task_name(self) -> str: ...
 
 
-_VALID_CONDITIONS = frozenset({"on_success", "on_failure", "always"})
 _VALID_ON_FAILURE = frozenset({"fail_fast", "continue"})
 
 
@@ -79,7 +78,7 @@ class _Step:
     priority: int | None = None
     fan_out: FanStrategy | None = None
     fan_in: FanStrategy | None = None
-    condition: str | Callable | None = None
+    condition: WorkflowCondition | Callable | None = None
     gate_config: GateConfig | None = None
     sub_workflow: SubWorkflowRef | None = None
     # ``None`` = no compensation for this step; a string = the compensation
@@ -145,7 +144,7 @@ def step(
         priority: int | None = None,
         fan_out: FanStrategy | str | None = None,
         fan_in: FanStrategy | str | None = None,
-        condition: str | Callable | None = None,
+        condition: WorkflowCondition | str | Callable | None = None,
         compensates: HasTaskName | str | None | _InheritCompensator = INHERIT_COMPENSATOR,
     ) -> Workflow:
         """Add a step to the workflow.
@@ -164,10 +163,11 @@ def step(
                 return value into one job per element.
             fan_in: Fan-in strategy. ``"all"`` collects all fan-out children's
                 results into a list passed to this step.
-            condition: Optional gate expression evaluated against upstream
-                results. Either a string DSL expression or a callable
-                receiving the workflow context; when it returns falsy the
-                step (and its descendants) are skipped.
+            condition: Optional gate on upstream results. A
+                :class:`~taskito.workflows.types.WorkflowCondition` (or its
+                string — ``"on_success"``, ``"on_failure"``, ``"always"``), or a
+                callable receiving the workflow context; when it returns falsy
+                the step (and its descendants) are skipped.
 
         Returns:
             ``self`` for chaining.
@@ -225,15 +225,9 @@ def step(
                     f"step '{name}': fan_in predecessor '{predecessors[0]}' must have fan_out set"
                 )
 
-        is_invalid_condition = (
-            condition is not None
-            and not callable(condition)
-            and condition not in _VALID_CONDITIONS
-        )
-        if is_invalid_condition:
-            raise ValueError(
-                f"step '{name}': condition must be one of "
-                f"{sorted(_VALID_CONDITIONS)} or a callable, got '{condition}'"
+        if condition is not None and not callable(condition):
+            condition = coerce_enum(
+                WorkflowCondition, condition, param=f"step '{name}': condition"
             )
 
         sub_wf = task if isinstance(task, SubWorkflowRef) else None
@@ -284,7 +278,7 @@ def gate(
         name: str,
         *,
         after: str | list[str] | None = None,
-        condition: str | Callable | None = None,
+        condition: WorkflowCondition | str | Callable | None = None,
         timeout: float | None = None,
         on_timeout: GateAction | str = GateAction.REJECT,
         message: str | Callable | None = None,
@@ -310,6 +304,10 @@ def gate(
         if "[" in name:
             raise ValueError(f"step '{name}': names must not contain '['")
         timeout_action = coerce_enum(GateAction, on_timeout, param=f"gate '{name}': on_timeout")
+        if condition is not None and not callable(condition):
+            condition = coerce_enum(
+                WorkflowCondition, condition, param=f"gate '{name}': condition"
+            )
 
         if after is None:
             predecessors: list[str] = []
@@ -461,8 +459,8 @@ def _compile(
 
         for step in self._steps.values():
             str_condition: str | None = None
-            if isinstance(step.condition, str):
-                str_condition = step.condition
+            if isinstance(step.condition, WorkflowCondition):
+                str_condition = step.condition.value
             elif callable(step.condition):
                 callable_conditions[step.name] = step.condition
                 str_condition = "callable"
diff --git a/sdks/python/taskito/workflows/saga/orchestrator.py b/sdks/python/taskito/workflows/saga/orchestrator.py
index 3c686c79..31c635bb 100644
--- a/sdks/python/taskito/workflows/saga/orchestrator.py
+++ b/sdks/python/taskito/workflows/saga/orchestrator.py
@@ -45,6 +45,7 @@
 from taskito.events import EventType
 from taskito.workflows.analysis import topological_levels
 from taskito.workflows.saga.context import CompensationContext
+from taskito.workflows.types import NodeStatus
 
 if TYPE_CHECKING:
     from taskito.app import Queue
@@ -258,7 +259,7 @@ def _fetch_compensable_nodes(
                 status = entry[1]
             except (IndexError, TypeError):
                 continue
-            if status not in ("completed", "cache_hit"):
+            if status not in (NodeStatus.COMPLETED.value, NodeStatus.CACHE_HIT.value):
                 continue
             if name in compensation_map:
                 compensable.add(name)
diff --git a/sdks/python/taskito/workflows/tracker/dag.py b/sdks/python/taskito/workflows/tracker/dag.py
index dbe16adf..d7bb5c2a 100644
--- a/sdks/python/taskito/workflows/tracker/dag.py
+++ b/sdks/python/taskito/workflows/tracker/dag.py
@@ -14,6 +14,7 @@
 
 from taskito.events import EventType
 from taskito.workflows.context import WorkflowContext
+from taskito.workflows.types import NodeStatus, WorkflowCondition, WorkflowState
 
 if TYPE_CHECKING:
     from taskito.workflows.tracker.tracker import WorkflowTracker
@@ -22,6 +23,11 @@
 
 logger = logging.getLogger("taskito.workflows")
 
+# A predecessor gates its successors once it reaches one of these node statuses.
+_PREDECESSOR_TERMINAL = frozenset(
+    {NodeStatus.COMPLETED.value, NodeStatus.FAILED.value, NodeStatus.SKIPPED.value}
+)
+
 
 def build_dag_maps(
     dag_bytes: bytes | list[int],
@@ -46,13 +52,13 @@ def int_or(value: Any, default: int) -> int:
 
 
 def final_state_to_event(state: str) -> EventType | None:
-    if state == "completed":
+    if state == WorkflowState.COMPLETED.value:
         return EventType.WORKFLOW_COMPLETED
-    if state == "completed_with_failures":
+    if state == WorkflowState.COMPLETED_WITH_FAILURES.value:
         return EventType.WORKFLOW_COMPLETED_WITH_FAILURES
-    if state == "failed":
+    if state == WorkflowState.FAILED.value:
         return EventType.WORKFLOW_FAILED
-    if state == "cancelled":
+    if state == WorkflowState.CANCELLED.value:
         return EventType.WORKFLOW_CANCELLED
     return None
 
@@ -68,7 +74,7 @@ def all_predecessors_terminal(
         if info is None:
             return False
         status = info["status"]
-        if status not in ("completed", "failed", "skipped"):
+        if status not in _PREDECESSOR_TERMINAL:
             return False
     return True
 
@@ -101,12 +107,12 @@ def build_workflow_context(
     for name, info in node_statuses.items():
         status = info["status"]
         statuses[name] = status
-        if status == "completed":
+        if status == NodeStatus.COMPLETED.value:
             success_count += 1
             jid = info.get("job_id")
             if jid:
                 results[name] = tracker._fetch_result(jid)
-        elif status == "failed":
+        elif status == NodeStatus.FAILED.value:
             failure_count += 1
 
     return WorkflowContext(
@@ -136,11 +142,11 @@ def should_execute(
     condition = meta.get("condition")
     pred_statuses = get_predecessor_statuses(tracker, run_id, node_name, config)
 
-    if condition is None or condition == "on_success":
-        return all(s == "completed" for s in pred_statuses.values())
-    if condition == "on_failure":
-        return any(s == "failed" for s in pred_statuses.values())
-    return bool(condition == "always")
+    if condition is None or condition == WorkflowCondition.ON_SUCCESS.value:
+        return all(s == NodeStatus.COMPLETED.value for s in pred_statuses.values())
+    if condition == WorkflowCondition.ON_FAILURE.value:
+        return any(s == NodeStatus.FAILED.value for s in pred_statuses.values())
+    return bool(condition == WorkflowCondition.ALWAYS.value)
 
 
 def skip_and_propagate(
diff --git a/sdks/python/taskito/workflows/types.py b/sdks/python/taskito/workflows/types.py
index 1fd33c57..c78c31a9 100644
--- a/sdks/python/taskito/workflows/types.py
+++ b/sdks/python/taskito/workflows/types.py
@@ -33,6 +33,19 @@ class DiagramFormat(str, enum.Enum):
     DOT = "dot"
 
 
+class WorkflowCondition(str, enum.Enum):
+    """When a step runs, based on its predecessors' outcomes."""
+
+    ON_SUCCESS = "on_success"
+    """Every predecessor completed (the default)."""
+
+    ON_FAILURE = "on_failure"
+    """At least one predecessor failed — an error-handler branch."""
+
+    ALWAYS = "always"
+    """Regardless of predecessor outcomes, once they settle."""
+
+
 class WorkflowState(str, enum.Enum):
     """Terminal and intermediate states of a workflow run."""
 

From 6efde5dbbe285acf8707788256ea80c2e6bc3af6 Mon Sep 17 00:00:00 2001
From: kartikeya <67143288+kartikeya-27@users.noreply.github.com>
Date: Fri, 24 Jul 2026 00:57:17 +0530
Subject: [PATCH 04/11] feat(python): accept a Strategy enum in register_type

---
 sdks/python/taskito/mixins/runtime_config.py | 12 +++++++-----
 1 file changed, 7 insertions(+), 5 deletions(-)

diff --git a/sdks/python/taskito/mixins/runtime_config.py b/sdks/python/taskito/mixins/runtime_config.py
index 1165e10d..8c66a677 100644
--- a/sdks/python/taskito/mixins/runtime_config.py
+++ b/sdks/python/taskito/mixins/runtime_config.py
@@ -15,7 +15,8 @@
 from collections.abc import Callable
 from typing import TYPE_CHECKING, Any
 
-from taskito.interception.strategy import Strategy as S
+from taskito.enums import coerce_enum
+from taskito.interception.strategy import Strategy
 
 if TYPE_CHECKING:
     from taskito.interception import ArgumentInterceptor
@@ -31,7 +32,7 @@ class QueueRuntimeConfigMixin:
     def register_type(
         self,
         python_type: type,
-        strategy: str,
+        strategy: Strategy | str,
         *,
         resource: str | None = None,
         message: str | None = None,
@@ -43,8 +44,9 @@ def register_type(
 
         Args:
             python_type: The type to register.
-            strategy: One of ``"pass"``, ``"convert"``, ``"redirect"``,
-                ``"reject"``, or ``"proxy"``.
+            strategy: A :class:`~taskito.interception.strategy.Strategy` or its
+                string: ``"pass"``, ``"convert"``, ``"redirect"``, ``"reject"``,
+                or ``"proxy"``.
             resource: Resource name for ``"redirect"`` strategy.
             message: Rejection reason for ``"reject"`` strategy.
             converter: Converter callable for ``"convert"`` strategy.
@@ -56,7 +58,7 @@ def register_type(
                 "Interception is disabled; set interception='strict' or "
                 "'lenient' to use register_type()"
             )
-        strat = S(strategy)
+        strat = coerce_enum(Strategy, strategy, param="strategy")
         self._interceptor._registry.register(
             python_type,
             strat,

From dc829e7835f52a2413a04f75186f0fdecd8279e6 Mon Sep 17 00:00:00 2001
From: kartikeya <67143288+kartikeya-27@users.noreply.github.com>
Date: Fri, 24 Jul 2026 00:57:30 +0530
Subject: [PATCH 05/11] feat(python): type the storage backend as
 StorageBackend

---
 sdks/python/taskito/__init__.py |  2 ++
 sdks/python/taskito/app.py      | 13 ++++++++++---
 sdks/python/taskito/enums.py    | 10 +++++++++-
 3 files changed, 21 insertions(+), 4 deletions(-)

diff --git a/sdks/python/taskito/__init__.py b/sdks/python/taskito/__init__.py
index 9d89e8d0..123f1271 100644
--- a/sdks/python/taskito/__init__.py
+++ b/sdks/python/taskito/__init__.py
@@ -15,6 +15,7 @@
     PayloadCodec,
 )
 from taskito.context import LogLevel, current_job
+from taskito.enums import StorageBackend
 from taskito.events import EventType
 from taskito.exceptions import (
     CircuitBreakerOpenError,
@@ -120,6 +121,7 @@
     "SignedSerializer",
     "SmartSerializer",
     "SoftTimeoutError",
+    "StorageBackend",
     "TaskCancelledError",
     "TaskFailedError",
     "TaskMiddleware",
diff --git a/sdks/python/taskito/app.py b/sdks/python/taskito/app.py
index d201b5cf..910fe4d2 100644
--- a/sdks/python/taskito/app.py
+++ b/sdks/python/taskito/app.py
@@ -30,7 +30,7 @@
 from taskito.async_support.mixins import AsyncQueueMixin
 from taskito.batching import BatchAccumulator, BatchConfig
 from taskito.codecs import CodecSerializer, PayloadCodec
-from taskito.enums import coerce_enum
+from taskito.enums import StorageBackend, coerce_enum
 from taskito.events import EventBus, EventType
 from taskito.exceptions import QueueFullError, SerializationError
 from taskito.interception import ArgumentInterceptor, InterceptionMode
@@ -134,7 +134,7 @@ def __init__(
         codec: PayloadCodec | Sequence[PayloadCodec] | None = None,
         codecs: dict[str, PayloadCodec] | None = None,
         middleware: list[TaskMiddleware] | None = None,
-        backend: str = "sqlite",
+        backend: StorageBackend | str = StorageBackend.SQLITE,
         db_url: str | None = None,
         schema: str = "taskito",
         pool_size: int | None = None,
@@ -188,7 +188,9 @@ def __init__(
                 via ``@queue.task(codecs=["name", ...])``; applies to task
                 payloads only (results stay on the queue serializer).
             middleware: List of global middleware instances applied to all tasks.
-            backend: Storage backend — ``"sqlite"`` (default) or ``"postgres"``.
+            backend: Storage backend — a
+                :class:`~taskito.enums.StorageBackend` or its string:
+                ``"sqlite"`` (default), ``"postgres"``, or ``"redis"``.
             db_url: PostgreSQL connection URL (required when backend is ``"postgres"``).
                 Example: ``"postgresql://user:pass@localhost/taskito"``.
             schema: PostgreSQL schema name for all taskito tables. Defaults to
@@ -243,6 +245,11 @@ def __init__(
                 guarantee as the rate limiter. Also settable at runtime via
                 ``set_queue_max_pending``.
         """
+        # Unwrap the enum to its wire string; a raw string still passes through
+        # so the native layer's aliases (e.g. "postgresql") keep working.
+        if isinstance(backend, StorageBackend):
+            backend = backend.value
+
         if backend == "sqlite":
             # Ensure parent directory exists for SQLite
             db_dir = os.path.dirname(db_path)
diff --git a/sdks/python/taskito/enums.py b/sdks/python/taskito/enums.py
index b1193a81..6433fc45 100644
--- a/sdks/python/taskito/enums.py
+++ b/sdks/python/taskito/enums.py
@@ -1,4 +1,4 @@
-"""Coercion helper for the closed-set enums on public entry points."""
+"""Cross-cutting closed-set enums plus the coercion helper for entry points."""
 
 from __future__ import annotations
 
@@ -8,6 +8,14 @@
 E = TypeVar("E", bound=enum.Enum)
 
 
+class StorageBackend(str, enum.Enum):
+    """Storage backend for a :class:`~taskito.app.Queue`."""
+
+    SQLITE = "sqlite"
+    POSTGRES = "postgres"
+    REDIS = "redis"
+
+
 def coerce_enum(enum_cls: type[E], value: E | str, *, param: str) -> E:
     """Accept an enum member or its wire string, else raise naming the valid set.
 

From 5b4ff89b84b04dc7b8e0ffd14fd7f8012b0e53d7 Mon Sep 17 00:00:00 2001
From: kartikeya <67143288+kartikeya-27@users.noreply.github.com>
Date: Fri, 24 Jul 2026 00:59:08 +0530
Subject: [PATCH 06/11] feat(python): type the webhook delivery status

---
 .../taskito/dashboard/delivery_store.py       | 34 +++++++++++++++----
 .../dashboard/handlers/webhook_deliveries.py  | 19 ++++++++---
 sdks/python/taskito/webhooks.py               | 12 +++----
 .../dashboard/test_webhook_deliveries.py      | 24 ++++++++-----
 4 files changed, 63 insertions(+), 26 deletions(-)

diff --git a/sdks/python/taskito/dashboard/delivery_store.py b/sdks/python/taskito/dashboard/delivery_store.py
index 0efc8d98..c9235e84 100644
--- a/sdks/python/taskito/dashboard/delivery_store.py
+++ b/sdks/python/taskito/dashboard/delivery_store.py
@@ -16,7 +16,7 @@
         "task_name": "send_email" | null,
         "job_id": "abc123" | null,
         "payload": {...},
-        "status": "delivered" | "failed" | "dead",
+        "status": "pending" | "delivered" | "failed" | "dead",
         "attempts": 3,
         "response_code": 200 | null,
         "response_body": "..." | null,
@@ -33,6 +33,7 @@
 
 from __future__ import annotations
 
+import enum
 import json
 import logging
 import time
@@ -44,6 +45,27 @@
     from taskito.app import Queue
 
 
+class DeliveryStatus(str, enum.Enum):
+    """Terminal (or pending) state of a webhook delivery attempt.
+
+    ``PENDING`` is a Python-only starting state — a record exists before its
+    first attempt settles. Other SDKs record only settled attempts.
+    """
+
+    PENDING = "pending"
+    DELIVERED = "delivered"
+    FAILED = "failed"
+    DEAD = "dead"
+
+
+def _parse_status(value: Any) -> DeliveryStatus:
+    """Read a stored status, falling back to ``PENDING`` on an unknown value."""
+    try:
+        return DeliveryStatus(str(value))
+    except ValueError:
+        return DeliveryStatus.PENDING
+
+
 DELIVERY_PREFIX = "webhooks:deliveries:"
 DEFAULT_MAX_PER_WEBHOOK = 200
 RESPONSE_BODY_MAX_BYTES = 2048
@@ -61,7 +83,7 @@ class DeliveryRecord:
     payload: dict[str, Any]
     task_name: str | None = None
     job_id: str | None = None
-    status: str = "pending"  # "delivered" | "failed" | "dead" | "pending"
+    status: DeliveryStatus = DeliveryStatus.PENDING
     attempts: int = 0
     response_code: int | None = None
     response_body: str | None = None
@@ -79,7 +101,7 @@ def from_row(cls, row: dict[str, Any]) -> DeliveryRecord:
             payload=dict(row.get("payload") or {}),
             task_name=row.get("task_name"),
             job_id=row.get("job_id"),
-            status=str(row.get("status", "pending")),
+            status=_parse_status(row.get("status")),
             attempts=int(row.get("attempts", 0)),
             response_code=row.get("response_code"),
             response_body=row.get("response_body"),
@@ -144,7 +166,7 @@ def record_attempt(
         event: str,
         payload: dict[str, Any],
         *,
-        status: str,
+        status: DeliveryStatus,
         attempts: int,
         response_code: int | None = None,
         response_body: str | None = None,
@@ -169,7 +191,7 @@ def record_attempt(
             latency_ms=latency_ms,
             error=error,
             created_at=now,
-            completed_at=now if status != "pending" else None,
+            completed_at=now if status is not DeliveryStatus.PENDING else None,
         )
         rows = self._load(subscription_id)
         rows.append(asdict(record))
@@ -182,7 +204,7 @@ def list_for(
         self,
         subscription_id: str,
         *,
-        status: str | None = None,
+        status: DeliveryStatus | str | None = None,
         event: str | None = None,
         limit: int = 50,
         offset: int = 0,
diff --git a/sdks/python/taskito/dashboard/handlers/webhook_deliveries.py b/sdks/python/taskito/dashboard/handlers/webhook_deliveries.py
index aa5bbe49..347cf275 100644
--- a/sdks/python/taskito/dashboard/handlers/webhook_deliveries.py
+++ b/sdks/python/taskito/dashboard/handlers/webhook_deliveries.py
@@ -5,7 +5,7 @@
 from dataclasses import asdict
 from typing import TYPE_CHECKING, Any
 
-from taskito.dashboard.delivery_store import DeliveryRecord, DeliveryStore
+from taskito.dashboard.delivery_store import DeliveryRecord, DeliveryStatus, DeliveryStore
 from taskito.dashboard.errors import _BadRequest, _NotFound
 from taskito.dashboard.webhook_store import WebhookSubscriptionStore
 
@@ -44,9 +44,14 @@ def handle_list_deliveries(queue: Queue, qs: dict, subscription_id: str) -> dict
     ``event``, ``limit``, and ``offset`` query parameters."""
     _ensure_subscription(queue, subscription_id)
 
-    status = qs.get("status", [None])[0]
-    if status is not None and status not in {"delivered", "failed", "dead", "pending"}:
-        raise _BadRequest("status must be one of: delivered, failed, dead, pending")
+    status_raw = qs.get("status", [None])[0]
+    status: DeliveryStatus | None = None
+    if status_raw:
+        try:
+            status = DeliveryStatus(status_raw)
+        except ValueError:
+            valid = ", ".join(s.value for s in DeliveryStatus)
+            raise _BadRequest(f"status must be one of: {valid}") from None
     event = qs.get("event", [None])[0]
 
     limit = min(_parse_int_param(qs, "limit", 50, minimum=1), _MAX_PAGE_SIZE)
@@ -98,7 +103,11 @@ def handle_replay_delivery(queue: Queue, sub_and_delivery_id: tuple[str, str]) -
         subscription_id,
         event=str(payload.get("event", record.event)),
         payload=payload,
-        status="delivered" if status is not None and status < 400 else "failed",
+        status=(
+            DeliveryStatus.DELIVERED
+            if status is not None and status < 400
+            else DeliveryStatus.FAILED
+        ),
         attempts=1,
         response_code=status,
         task_name=record.task_name,
diff --git a/sdks/python/taskito/webhooks.py b/sdks/python/taskito/webhooks.py
index 92cad79d..e79abbc4 100644
--- a/sdks/python/taskito/webhooks.py
+++ b/sdks/python/taskito/webhooks.py
@@ -27,7 +27,7 @@
 import urllib.request
 from typing import TYPE_CHECKING, Any
 
-from taskito.dashboard.delivery_store import DeliveryStore
+from taskito.dashboard.delivery_store import DeliveryStatus, DeliveryStore
 from taskito.dashboard.url_safety import UnsafeWebhookUrl, validate_webhook_url
 from taskito.events import EventType
 
@@ -243,7 +243,7 @@ def _send(
                         self._record(
                             wh,
                             payload,
-                            status="delivered",
+                            status=DeliveryStatus.DELIVERED,
                             attempts=attempt_count,
                             response_code=last_status,
                             response_body=last_response_body,
@@ -268,7 +268,7 @@ def _send(
                     self._record(
                         wh,
                         payload,
-                        status="failed",
+                        status=DeliveryStatus.FAILED,
                         attempts=attempt_count,
                         response_code=last_status,
                         response_body=last_response_body,
@@ -287,7 +287,7 @@ def _send(
                 self._record(
                     wh,
                     payload,
-                    status="failed",
+                    status=DeliveryStatus.FAILED,
                     attempts=attempt_count,
                     response_code=None,
                     response_body=None,
@@ -319,7 +319,7 @@ def _send(
         self._record(
             wh,
             payload,
-            status="dead",
+            status=DeliveryStatus.DEAD,
             attempts=attempt_count,
             response_code=last_status,
             response_body=last_response_body,
@@ -334,7 +334,7 @@ def _record(
         wh: dict[str, Any],
         payload: dict[str, Any],
         *,
-        status: str,
+        status: DeliveryStatus,
         attempts: int,
         response_code: int | None = None,
         response_body: str | None = None,
diff --git a/sdks/python/tests/dashboard/test_webhook_deliveries.py b/sdks/python/tests/dashboard/test_webhook_deliveries.py
index 16f5e83c..98736151 100644
--- a/sdks/python/tests/dashboard/test_webhook_deliveries.py
+++ b/sdks/python/tests/dashboard/test_webhook_deliveries.py
@@ -16,7 +16,7 @@
 from taskito import Queue
 from taskito.dashboard import _make_handler
 from taskito.dashboard._testing import AuthedClient, seed_admin_and_session
-from taskito.dashboard.delivery_store import DeliveryStore
+from taskito.dashboard.delivery_store import DeliveryStatus, DeliveryStore
 from taskito.events import EventType
 
 
@@ -100,7 +100,7 @@ def test_record_attempt_appends(queue: Queue) -> None:
         "sub1",
         event="job.completed",
         payload={"job_id": "x"},
-        status="delivered",
+        status=DeliveryStatus.DELIVERED,
         attempts=1,
         response_code=200,
         latency_ms=10,
@@ -120,7 +120,7 @@ def test_record_attempt_caps_history(queue: Queue) -> None:
             "sub1",
             event="job.completed",
             payload={"job_id": str(i)},
-            status="delivered",
+            status=DeliveryStatus.DELIVERED,
             attempts=1,
         )
     items = store.list_for("sub1")
@@ -137,7 +137,7 @@ def test_record_attempt_truncates_response_body(queue: Queue) -> None:
         "sub1",
         event="job.completed",
         payload={},
-        status="failed",
+        status=DeliveryStatus.FAILED,
         attempts=1,
         response_body=big,
     )
@@ -147,13 +147,19 @@ def test_record_attempt_truncates_response_body(queue: Queue) -> None:
 
 def test_list_for_filters_by_status_and_event(queue: Queue) -> None:
     store = DeliveryStore(queue)
-    store.record_attempt("sub1", event="job.completed", payload={}, status="delivered", attempts=1)
-    store.record_attempt("sub1", event="job.failed", payload={}, status="failed", attempts=1)
-    store.record_attempt("sub1", event="job.completed", payload={}, status="failed", attempts=1)
+    store.record_attempt(
+        "sub1", event="job.completed", payload={}, status=DeliveryStatus.DELIVERED, attempts=1
+    )
+    store.record_attempt(
+        "sub1", event="job.failed", payload={}, status=DeliveryStatus.FAILED, attempts=1
+    )
+    store.record_attempt(
+        "sub1", event="job.completed", payload={}, status=DeliveryStatus.FAILED, attempts=1
+    )
 
-    delivered = store.list_for("sub1", status="delivered")
+    delivered = store.list_for("sub1", status=DeliveryStatus.DELIVERED)
     assert len(delivered) == 1
-    failed = store.list_for("sub1", status="failed")
+    failed = store.list_for("sub1", status=DeliveryStatus.FAILED)
     assert len(failed) == 2
     completed_event = store.list_for("sub1", event="job.completed")
     assert len(completed_event) == 2

From 83da7179f3d65ed30e3e5c1d0859b3c1de7589c1 Mon Sep 17 00:00:00 2001
From: kartikeya <67143288+kartikeya-27@users.noreply.github.com>
Date: Fri, 24 Jul 2026 00:59:21 +0530
Subject: [PATCH 07/11] test(python): cover the remaining closed-set enums

---
 .../tests/python/test_closed_set_enums.py     | 95 ++++++++++++++++++-
 1 file changed, 93 insertions(+), 2 deletions(-)

diff --git a/sdks/python/tests/python/test_closed_set_enums.py b/sdks/python/tests/python/test_closed_set_enums.py
index 913b83fe..780a32f2 100644
--- a/sdks/python/tests/python/test_closed_set_enums.py
+++ b/sdks/python/tests/python/test_closed_set_enums.py
@@ -6,8 +6,14 @@
 
 import pytest
 
-from taskito import InterceptionMode, PredicateAction, Queue
+from taskito import InterceptionMode, PredicateAction, Queue, StorageBackend
+from taskito.dashboard.delivery_store import DeliveryStatus, DeliveryStore
+from taskito.dashboard.errors import _BadRequest
+from taskito.dashboard.handlers.webhook_deliveries import handle_list_deliveries
+from taskito.dashboard.webhook_store import WebhookSubscriptionStore
+from taskito.interception.strategy import Strategy
 from taskito.mixins._log_consumer import ConsumerErrorAction
+from taskito.workflows import Workflow, WorkflowCondition
 from taskito.workflows.types import DiagramFormat, FanStrategy, GateAction
 
 
@@ -19,6 +25,10 @@ def test_wire_values_are_the_cross_sdk_contract() -> None:
     assert GateAction.APPROVE.value == "approve"
     assert FanStrategy.EACH.value == "each"
     assert DiagramFormat.DOT.value == "dot"
+    assert WorkflowCondition.ON_SUCCESS.value == "on_success"
+    assert StorageBackend.POSTGRES.value == "postgres"
+    assert Strategy.PROXY.value == "proxy"
+    assert DeliveryStatus.DEAD.value == "dead"
     # A (str, Enum) member carries its value as its string content, which is what
     # the native layer reads — str() would render "FanStrategy.EACH" instead.
     assert FanStrategy.EACH.encode() == b"each"
@@ -74,7 +84,6 @@ def bad(message: object) -> None:
 
 def test_unknown_visualize_format_raises_instead_of_defaulting(queue: Queue) -> None:
     """Previously anything that wasn't 'dot' silently rendered Mermaid."""
-    from taskito.workflows import Workflow
 
     @queue.task()
     def step() -> None:
@@ -85,3 +94,85 @@ def step() -> None:
     assert "graph" in workflow.visualize("mermaid")
     with pytest.raises(ValueError, match="'mermaid', 'dot'"):
         workflow.visualize("graphviz")
+
+
+def test_step_condition_accepts_enum_and_string(queue: Queue) -> None:
+    @queue.task()
+    def step() -> None:
+        """A node to gate."""
+
+    wf = Workflow("wf").step("a", step)
+    wf.step("b", step, after="a", condition=WorkflowCondition.ON_FAILURE)
+    wf.step("c", step, after="a", condition="always")
+    assert wf._steps["b"].condition is WorkflowCondition.ON_FAILURE
+    assert wf._steps["c"].condition is WorkflowCondition.ALWAYS
+
+
+def test_step_condition_typo_raises(queue: Queue) -> None:
+    @queue.task()
+    def step() -> None:
+        """A node that never gets a valid condition."""
+
+    wf = Workflow("wf").step("a", step)
+    with pytest.raises(ValueError, match="'on_success', 'on_failure', 'always'"):
+        wf.step("b", step, after="a", condition="on_succes")
+
+
+def test_gate_condition_typo_raises(queue: Queue) -> None:
+    @queue.task()
+    def step() -> None:
+        """Predecessor for the gate."""
+
+    wf = Workflow("wf").step("a", step)
+    with pytest.raises(ValueError, match="'on_success', 'on_failure', 'always'"):
+        wf.gate("g", after="a", condition="whenever")
+
+
+def test_backend_accepts_enum_and_string(tmp_path: Path) -> None:
+    enum_queue = Queue(db_path=str(tmp_path / "a.db"), backend=StorageBackend.SQLITE)
+    string_queue = Queue(db_path=str(tmp_path / "b.db"), backend="sqlite")
+    try:
+        # Normalized to the wire string so the display / native layer see "sqlite".
+        assert enum_queue._backend == "sqlite"
+        assert string_queue._backend == "sqlite"
+    finally:
+        enum_queue.close()
+        string_queue.close()
+
+
+def test_register_type_accepts_enum_and_string(tmp_path: Path) -> None:
+    queue = Queue(db_path=str(tmp_path / "a.db"), interception="lenient")
+    try:
+        queue.register_type(complex, Strategy.PASS)
+        queue.register_type(bytes, "pass")
+    finally:
+        queue.close()
+
+
+def test_register_type_typo_raises(tmp_path: Path) -> None:
+    queue = Queue(db_path=str(tmp_path / "a.db"), interception="lenient")
+    try:
+        with pytest.raises(ValueError, match="'pass', 'convert'"):
+            queue.register_type(complex, "passs")
+    finally:
+        queue.close()
+
+
+def test_delivery_status_filter_round_trips_and_rejects_typo(queue: Queue) -> None:
+    sub = WebhookSubscriptionStore(queue).create(url="https://example.test/hook", events=["*"])
+    store = DeliveryStore(queue)
+    store.record_attempt(
+        sub.id, event="job.completed", payload={}, status=DeliveryStatus.DELIVERED, attempts=1
+    )
+    store.record_attempt(
+        sub.id, event="job.failed", payload={}, status=DeliveryStatus.DEAD, attempts=3
+    )
+
+    only_dead = store.list_for(sub.id, status=DeliveryStatus.DEAD)
+    assert [r.status for r in only_dead] == [DeliveryStatus.DEAD]
+
+    body = handle_list_deliveries(queue, {"status": ["delivered"]}, sub.id)
+    assert [item["status"] for item in body["items"]] == ["delivered"]
+
+    with pytest.raises(_BadRequest, match="delivered, failed, dead"):
+        handle_list_deliveries(queue, {"status": ["gone"]}, sub.id)

From 1481ef924088b803e52d71412e175b8e65b24d50 Mon Sep 17 00:00:00 2001
From: kartikeya <67143288+kartikeya-27@users.noreply.github.com>
Date: Fri, 24 Jul 2026 00:59:26 +0530
Subject: [PATCH 08/11] docs: note the remaining closed-set enum typing

---
 CHANGELOG.md | 6 ++++++
 1 file changed, 6 insertions(+)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 8118cceb..b8ee24a5 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -22,6 +22,12 @@ underlying Rust crates are released together, in lock-step.
   workflow run state, dashboard role, subscription mode, and worker status. Wire and stored
   values are unchanged, and existing string callers keep working (Java keeps its `String`
   overloads, deprecated).
+- More closed-set surfaces typed, finishing the sweep above: workflow step `condition`
+  (`WorkflowCondition`, Python + Java), storage `backend` (`StorageBackend`, Python + Java),
+  interception `register_type(strategy=…)` (`Strategy`, Python), and the webhook delivery-log
+  `status` filter (`DeliveryStatus`, Python). Same non-breaking contract — a string still works.
+  Internally, Python's workflow tracker and saga orchestrator now compare node/run status against
+  the enums rather than string literals.
 
 ### Added
 

From 0707a436ad519ce1961f917bc54a108a0c428d5a Mon Sep 17 00:00:00 2001
From: kartikeya <67143288+kartikeya-27@users.noreply.github.com>
Date: Fri, 24 Jul 2026 01:17:15 +0530
Subject: [PATCH 09/11] fix(python): put None last in builder unions (RUF036)

ruff 0.16 flags None not at the end of a type union; reorder the compensates annotations. Annotation-only, no behavior change.
---
 sdks/python/taskito/workflows/builder.py | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

diff --git a/sdks/python/taskito/workflows/builder.py b/sdks/python/taskito/workflows/builder.py
index 7da80222..f7497261 100644
--- a/sdks/python/taskito/workflows/builder.py
+++ b/sdks/python/taskito/workflows/builder.py
@@ -85,7 +85,7 @@ class _Step:
     # task name; the ``INHERIT_COMPENSATOR`` sentinel = look up the
     # decorator-level ``@queue.task(compensates=...)`` default when the
     # workflow is compiled.
-    compensates: str | None | _InheritCompensator = field(
+    compensates: str | _InheritCompensator | None = field(
         default_factory=lambda: INHERIT_COMPENSATOR
     )
 
@@ -145,7 +145,7 @@ def step(
         fan_out: FanStrategy | str | None = None,
         fan_in: FanStrategy | str | None = None,
         condition: WorkflowCondition | str | Callable | None = None,
-        compensates: HasTaskName | str | None | _InheritCompensator = INHERIT_COMPENSATOR,
+        compensates: HasTaskName | str | _InheritCompensator | None = INHERIT_COMPENSATOR,
     ) -> Workflow:
         """Add a step to the workflow.
 
@@ -234,7 +234,7 @@ def step(
 
         # Resolve the compensates= argument to either a task-name string,
         # ``None`` (disabled), or the INHERIT sentinel (look up at compile).
-        resolved_compensates: str | None | _InheritCompensator
+        resolved_compensates: str | _InheritCompensator | None
         if isinstance(compensates, _InheritCompensator):
             resolved_compensates = INHERIT_COMPENSATOR
         elif compensates is None:

From 46d8c87662ab15725937360f05c2e8615664ae1a Mon Sep 17 00:00:00 2001
From: kartikeya <67143288+kartikeya-27@users.noreply.github.com>
Date: Fri, 24 Jul 2026 01:38:39 +0530
Subject: [PATCH 10/11] fix(java): mention WorkflowCondition in condition error

---
 .../src/main/java/org/byteveda/taskito/workflows/Step.java     | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/sdks/java/src/main/java/org/byteveda/taskito/workflows/Step.java b/sdks/java/src/main/java/org/byteveda/taskito/workflows/Step.java
index c1bd48ec..6d7a0e90 100644
--- a/sdks/java/src/main/java/org/byteveda/taskito/workflows/Step.java
+++ b/sdks/java/src/main/java/org/byteveda/taskito/workflows/Step.java
@@ -154,7 +154,8 @@ public Builder condition(String condition) {
                     && !"on_failure".equals(condition)
                     && !"always".equals(condition)) {
                 throw new IllegalArgumentException("unknown condition '" + condition
-                        + "'; use on_success, on_failure, always, or condition(Condition)");
+                        + "'; use on_success, on_failure, always, a WorkflowCondition,"
+                        + " or condition(Condition)");
             }
             this.condition = condition;
             return this;

From 07241590ee808281473594d1f9bd73e9600ec975 Mon Sep 17 00:00:00 2001
From: kartikeya <67143288+kartikeya-27@users.noreply.github.com>
Date: Fri, 24 Jul 2026 01:38:39 +0530
Subject: [PATCH 11/11] fix(python): resolve compensates via _task_name too

---
 sdks/python/taskito/workflows/builder.py | 7 +++++--
 1 file changed, 5 insertions(+), 2 deletions(-)

diff --git a/sdks/python/taskito/workflows/builder.py b/sdks/python/taskito/workflows/builder.py
index f7497261..ccd82182 100644
--- a/sdks/python/taskito/workflows/builder.py
+++ b/sdks/python/taskito/workflows/builder.py
@@ -241,8 +241,11 @@ def step(
             resolved_compensates = None
         elif isinstance(compensates, str):
             resolved_compensates = compensates
-        elif hasattr(compensates, "name"):  # TaskWrapper / HasTaskName
-            resolved_compensates = compensates.name
+        elif getattr(compensates, "_task_name", None) or getattr(compensates, "name", None):
+            # HasTaskName exposes `_task_name`; a TaskWrapper also exposes `.name`.
+            resolved_compensates = getattr(compensates, "_task_name", None) or getattr(
+                compensates, "name", None
+            )
         else:
             raise TypeError("compensates= must be a TaskWrapper, task-name string, or None")