Add invocable activity carrier reference handler to Python SDK

durable-workflow-ops · durable-workflow-ops · commit 817e0d79eb07 · 2026-04-22T04:06:03.000Z
Provide a Python adapter for activity-grade invocable carriers that
consumes the stable external-task input envelope and returns the stable
external-task result envelope. The handler deliberately rejects
workflow-task inputs so HTTP or serverless adapters cannot become hidden
workflow runtimes, and maps cancellation, deadline, decode, and
non-retryable errors onto the existing failure shape.
diff --git a/README.md b/README.md
@@ -420,6 +420,18 @@ decisions for success, retryability, malformed output, cancellation, deadline
 exceeded, handler crash, decode failure, and unsupported payload
 codec/reference states without treating stderr as a machine signal.
 
+Invocable activity carriers can use `InvocableActivityHandler` as a reference
+adapter for HTTP or serverless runtimes. It accepts the same external-task input
+envelope, invokes a registered activity handler, and returns the same
+external-task result envelope while rejecting workflow-task inputs:
+
+```python
+from durable_workflow import InvocableActivityHandler
+
+adapter = InvocableActivityHandler({"billing.charge-card": charge_card})
+result_envelope = await adapter.handle(request_json)
+```
+
 Bridge adapters can hand bounded webhook ingress into the server through
 `Client.send_webhook_bridge_event()`. The method returns the server's typed
 bridge outcome for accepted, duplicate, and rejected events, including
diff --git a/src/durable_workflow/__init__.py b/src/durable_workflow/__init__.py
@@ -98,6 +98,7 @@
     WorkflowTaskHandler,
     WorkflowTaskInterceptorContext,
 )
+from .invocable import InvocableActivityHandler, handle_invocable_activity
 from .metrics import (
     InMemoryMetrics,
     MetricsRecorder,
@@ -192,6 +193,7 @@
     "parse_external_task_result_artifact",
     "InvalidArgument",
     "InMemoryMetrics",
+    "InvocableActivityHandler",
     "LocalFilesystemExternalStorage",
     "MetricsRecorder",
     "NamespaceNotFound",
@@ -223,6 +225,7 @@
     "parse_external_task_input",
     "parse_external_task_input_artifact",
     "parse_external_task_result",
+    "handle_invocable_activity",
     "to_avro_payload_value",
     "to_avro_payload_values",
     "replay",
diff --git a/src/durable_workflow/invocable.py b/src/durable_workflow/invocable.py
@@ -0,0 +1,242 @@
+from __future__ import annotations
+
+import asyncio
+import time
+import traceback
+from collections.abc import Awaitable, Callable, Mapping
+from typing import Any
+
+from . import serializer
+from .errors import ActivityCancelled, NonRetryableError
+from .external_storage import ExternalPayloadCache, ExternalStorageDriver
+from .external_task_input import ExternalTaskInput, parse_external_task_input
+from .external_task_result import EXTERNAL_TASK_RESULT_SCHEMA, EXTERNAL_TASK_RESULT_VERSION
+
+InvocableActivityCallable = Callable[..., Any | Awaitable[Any]]
+
+
+class InvocableActivityHandler:
+    """Reference adapter for activity-grade invocable carriers.
+
+    The handler consumes the stable external-task input envelope and returns
+    the stable external-task result envelope. It deliberately rejects workflow
+    tasks so lightweight HTTP/serverless carriers cannot become hidden workflow
+    runtimes.
+    """
+
+    def __init__(
+        self,
+        handlers: Mapping[str, InvocableActivityCallable],
+        *,
+        carrier: str = "python-invocable",
+        result_codec: str = serializer.AVRO_CODEC,
+        external_storage: ExternalStorageDriver | None = None,
+        external_storage_cache: ExternalPayloadCache | None = None,
+    ) -> None:
+        if result_codec not in serializer.SUPPORTED_CODECS:
+            raise ValueError(f"unsupported invocable result codec {result_codec!r}")
+        self.handlers = dict(handlers)
+        self.carrier = carrier
+        self.result_codec = result_codec
+        self.external_storage = external_storage
+        self.external_storage_cache = external_storage_cache
+
+    async def handle(self, envelope: Mapping[str, Any]) -> dict[str, Any]:
+        started = time.monotonic()
+        task_input = parse_external_task_input(envelope)
+
+        if not task_input.is_activity_task:
+            return self._failure(
+                task_input,
+                started,
+                kind="application",
+                classification="application_error",
+                message="invocable activity handlers only accept activity_task inputs",
+                failure_type="UnsupportedExternalTaskKind",
+                retryable=False,
+            )
+
+        handler_name = task_input.task.handler
+        handler = self.handlers.get(handler_name or "")
+        if handler is None:
+            return self._failure(
+                task_input,
+                started,
+                kind="application",
+                classification="application_error",
+                message=f"no invocable activity handler registered for {handler_name!r}",
+                failure_type="UnknownActivityHandler",
+                retryable=False,
+            )
+
+        try:
+            args = self._decode_arguments(task_input)
+            result = handler(*args)
+            if asyncio.iscoroutine(result):
+                result = await result
+        except ActivityCancelled:
+            return self._failure(
+                task_input,
+                started,
+                kind="cancellation",
+                classification="cancelled",
+                message="Activity was cancelled before the handler completed.",
+                failure_type="ActivityCancelled",
+                retryable=False,
+                cancelled=True,
+            )
+        except NonRetryableError as exc:
+            return self._failure(
+                task_input,
+                started,
+                kind="application",
+                classification="application_error",
+                message=str(exc),
+                failure_type=type(exc).__name__,
+                stack_trace=traceback.format_exc(),
+                retryable=False,
+            )
+        except TimeoutError as exc:
+            return self._failure(
+                task_input,
+                started,
+                kind="timeout",
+                classification="deadline_exceeded",
+                message=str(exc) or "Invocable activity handler exceeded its deadline.",
+                failure_type=type(exc).__name__,
+                stack_trace=traceback.format_exc(),
+                timeout_type="deadline_exceeded",
+                retryable=True,
+            )
+        except (TypeError, ValueError) as exc:
+            return self._failure(
+                task_input,
+                started,
+                kind="decode_failure",
+                classification="decode_failure",
+                message=f"Carrier could not decode or encode the activity payload: {exc}",
+                failure_type=type(exc).__name__,
+                stack_trace=traceback.format_exc(),
+                retryable=False,
+                details={"codec": self._input_codec(task_input)},
+            )
+        except Exception as exc:
+            return self._failure(
+                task_input,
+                started,
+                kind="application",
+                classification="application_error",
+                message=str(exc),
+                failure_type=type(exc).__name__,
+                stack_trace=traceback.format_exc(),
+                retryable=True,
+            )
+
+        return self._success(task_input, started, result)
+
+    def _decode_arguments(self, task_input: ExternalTaskInput) -> list[Any]:
+        raw_args = task_input.payloads["arguments"]
+        if raw_args is None:
+            return []
+
+        decoded = serializer.decode_envelope(
+            raw_args,
+            codec=self._input_codec(task_input),
+            external_storage=self.external_storage,
+            external_storage_cache=self.external_storage_cache,
+        )
+        if decoded is None:
+            return []
+        if isinstance(decoded, list):
+            return decoded
+        return [decoded]
+
+    @staticmethod
+    def _input_codec(task_input: ExternalTaskInput) -> str:
+        raw_args = task_input.payloads.get("arguments")
+        if isinstance(raw_args, Mapping):
+            codec = raw_args.get("codec")
+            if isinstance(codec, str):
+                return codec
+        return serializer.JSON_CODEC
+
+    def _success(self, task_input: ExternalTaskInput, started: float, result: Any) -> dict[str, Any]:
+        return {
+            "schema": EXTERNAL_TASK_RESULT_SCHEMA,
+            "version": EXTERNAL_TASK_RESULT_VERSION,
+            "outcome": {
+                "status": "succeeded",
+                "recorded": True,
+            },
+            "task": self._task_identity(task_input),
+            "result": {
+                "payload": serializer.envelope(result, codec=self.result_codec),
+                "metadata": {
+                    "content_type": "application/vnd.durable-workflow.result+json",
+                },
+            },
+            "metadata": self._metadata(task_input, started),
+        }
+
+    def _failure(
+        self,
+        task_input: ExternalTaskInput,
+        started: float,
+        *,
+        kind: str,
+        classification: str,
+        message: str,
+        failure_type: str,
+        retryable: bool,
+        stack_trace: str | None = None,
+        timeout_type: str | None = None,
+        cancelled: bool = False,
+        details: Mapping[str, Any] | None = None,
+    ) -> dict[str, Any]:
+        return {
+            "schema": EXTERNAL_TASK_RESULT_SCHEMA,
+            "version": EXTERNAL_TASK_RESULT_VERSION,
+            "outcome": {
+                "status": "failed",
+                "retryable": retryable,
+                "recorded": True,
+            },
+            "task": self._task_identity(task_input),
+            "failure": {
+                "kind": kind,
+                "classification": classification,
+                "message": message,
+                "type": failure_type,
+                "stack_trace": stack_trace,
+                "timeout_type": timeout_type,
+                "cancelled": cancelled,
+                "details": details,
+            },
+            "metadata": self._metadata(task_input, started),
+        }
+
+    @staticmethod
+    def _task_identity(task_input: ExternalTaskInput) -> dict[str, Any]:
+        return {
+            "id": task_input.task.id,
+            "kind": task_input.task.kind,
+            "attempt": task_input.task.attempt,
+            "idempotency_key": task_input.task.idempotency_key,
+        }
+
+    def _metadata(self, task_input: ExternalTaskInput, started: float) -> dict[str, Any]:
+        return {
+            "handler": task_input.task.handler,
+            "carrier": self.carrier,
+            "duration_ms": max(0, int((time.monotonic() - started) * 1000)),
+        }
+
+
+async def handle_invocable_activity(
+    envelope: Mapping[str, Any],
+    handlers: Mapping[str, InvocableActivityCallable],
+    **options: Any,
+) -> dict[str, Any]:
+    """Handle one invocable activity task with a temporary adapter instance."""
+
+    return await InvocableActivityHandler(handlers, **options).handle(envelope)
diff --git a/tests/test_invocable.py b/tests/test_invocable.py
@@ -0,0 +1,108 @@
+import copy
+import json
+from pathlib import Path
+from typing import Any
+
+import pytest
+
+from durable_workflow import serializer
+from durable_workflow.errors import NonRetryableError
+from durable_workflow.external_task_result import parse_external_task_result
+from durable_workflow.invocable import InvocableActivityHandler, handle_invocable_activity
+
+FIXTURES = Path(__file__).parent / "fixtures" / "external-task-input"
+
+
+def load_fixture(name: str) -> dict[str, Any]:
+    return json.loads((FIXTURES / name).read_text())
+
+
+def activity_input(*args: Any) -> dict[str, Any]:
+    envelope = copy.deepcopy(load_fixture("activity-task.v1.json"))
+    envelope["payloads"]["arguments"] = serializer.envelope(list(args), codec=serializer.JSON_CODEC)
+    return envelope
+
+
+async def test_invocable_activity_handler_returns_success_result_envelope() -> None:
+    handler = InvocableActivityHandler(
+        {"billing.charge-card": lambda amount, currency: {"approved": True, "amount": amount, "currency": currency}},
+        carrier="lambda-adapter",
+        result_codec=serializer.JSON_CODEC,
+    )
+
+    output = await handler.handle(activity_input(4200, "USD"))
+    result = parse_external_task_result(output)
+
+    assert result.succeeded is True
+    assert result.task.id == "acttask_01HV7D3G3G61TAH2YB5RK45XJS"
+    assert result.metadata["carrier"] == "lambda-adapter"
+    assert result.result is not None
+    assert serializer.decode_envelope(result.result["payload"]) == {
+        "approved": True,
+        "amount": 4200,
+        "currency": "USD",
+    }
+
+
+async def test_invocable_activity_handler_awaits_async_handlers() -> None:
+    async def charge(amount: int) -> dict[str, int]:
+        return {"amount": amount}
+
+    output = await handle_invocable_activity(
+        activity_input(42),
+        {"billing.charge-card": charge},
+        result_codec=serializer.JSON_CODEC,
+    )
+    result = parse_external_task_result(output)
+
+    assert result.succeeded is True
+    assert result.result is not None
+    assert serializer.decode_envelope(result.result["payload"]) == {"amount": 42}
+
+
+async def test_invocable_activity_handler_fails_closed_for_unknown_handler() -> None:
+    output = await InvocableActivityHandler({}, result_codec=serializer.JSON_CODEC).handle(activity_input("x"))
+    result = parse_external_task_result(output)
+
+    assert result.failed is True
+    assert result.retryable is False
+    assert result.failure_kind == "application"
+    assert result.failure_classification == "application_error"
+    assert result.failure is not None
+    assert "no invocable activity handler registered" in result.failure.message
+
+
+async def test_invocable_activity_handler_maps_non_retryable_errors() -> None:
+    def handler() -> None:
+        raise NonRetryableError("card rejected")
+
+    output = await InvocableActivityHandler(
+        {"billing.charge-card": handler},
+        result_codec=serializer.JSON_CODEC,
+    ).handle(activity_input())
+    result = parse_external_task_result(output)
+
+    assert result.failed is True
+    assert result.retryable is False
+    assert result.failure_kind == "application"
+    assert result.failure is not None
+    assert result.failure.message == "card rejected"
+
+
+async def test_invocable_activity_handler_rejects_workflow_task_inputs() -> None:
+    output = await InvocableActivityHandler(
+        {"billing.invoice.workflow": lambda: {"ignored": True}},
+        result_codec=serializer.JSON_CODEC,
+    ).handle(load_fixture("workflow-task.v1.json"))
+    result = parse_external_task_result(output)
+
+    assert result.failed is True
+    assert result.retryable is False
+    assert result.task.kind == "workflow_task"
+    assert result.failure is not None
+    assert "only accept activity_task" in result.failure.message
+
+
+async def test_invocable_activity_handler_validates_result_codec() -> None:
+    with pytest.raises(ValueError, match="unsupported invocable result codec"):
+        InvocableActivityHandler({}, result_codec="protobuf")
