feat: type the remaining closed-set params as enums

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
 

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);

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);
+        }
+    }
+}

sdks/java/src/main/java/org/byteveda/taskito/workflows/Step.java

@@ -154,12 +154,22 @@ 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;
         }
 
+        /**
+         * 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");

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);
+        }
+    }
+}

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"));
     }
 }

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 {

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",

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)

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,

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,

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.