From fc3b04834e3647b4e20f0f38a2fe8c1872fdb852 Mon Sep 17 00:00:00 2001 From: Andy Staples Date: Wed, 29 Jul 2026 14:21:16 -0600 Subject: [PATCH 1/4] Add v1 status history compatibility Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: a28a3fa8-303a-4a74-8fbc-8e062e39d97f --- azure-functions-durable/CHANGELOG.md | 7 +- azure-functions-durable/MIGRATION_GUIDE.md | 2 - .../azure/durable_functions/client.py | 21 +- .../compat/durable_orchestration_status.py | 22 +- .../internal/compat/history_projection.py | 191 ++++++++++++++++++ .../test_client_compat.py | 165 ++++++++++++++- ...est_durable_orchestration_status_compat.py | 23 ++- 7 files changed, 408 insertions(+), 23 deletions(-) create mode 100644 azure-functions-durable/azure/durable_functions/internal/compat/history_projection.py diff --git a/azure-functions-durable/CHANGELOG.md b/azure-functions-durable/CHANGELOG.md index e9c022e6..804ae4d1 100644 --- a/azure-functions-durable/CHANGELOG.md +++ b/azure-functions-durable/CHANGELOG.md @@ -32,6 +32,9 @@ avoiding repeated allocation of unused worker resources. FIXED +- The v1-compatible `get_status()` API now supports `show_history` and + `show_history_output`, including compacted `historyEvents` output without an + additional history request when history is not requested. - HTTP management payloads now preserve the host-provided management URL templates and configured HTTP base paths, include `rewindPostUri`, encode instance IDs, and use forwarded request origins when enabled by the host. @@ -206,7 +209,3 @@ code: - Orchestration history is not exposed on the context; `DurableOrchestrationContext.histories` raises `NotImplementedError`. Use the client's `get_orchestration_history(...)` instead. -- The client status methods accept the v1 `show_history` / - `show_history_output` flags for signature compatibility but ignore them, so - the returned status has no `historyEvents`. Use - `get_orchestration_history(...)` to retrieve history. diff --git a/azure-functions-durable/MIGRATION_GUIDE.md b/azure-functions-durable/MIGRATION_GUIDE.md index e43165f4..3a718750 100644 --- a/azure-functions-durable/MIGRATION_GUIDE.md +++ b/azure-functions-durable/MIGRATION_GUIDE.md @@ -269,8 +269,6 @@ instead of calling `context.set_result`. - `DurableOrchestrationContext.histories` is unavailable. Retrieve history with `client.get_orchestration_history(instance_id)`. -- The compatibility `show_history` and `show_history_output` status flags are - accepted but ignored. - Distributed tracing is not yet wired through the Python provider. - Continuous history export is not supported by Azure Functions. - Unusual callable signatures can be misclassified by the compatibility layer. diff --git a/azure-functions-durable/azure/durable_functions/client.py b/azure-functions-durable/azure/durable_functions/client.py index aef06339..512b6f55 100644 --- a/azure-functions-durable/azure/durable_functions/client.py +++ b/azure-functions-durable/azure/durable_functions/client.py @@ -28,6 +28,7 @@ from .http.http_management_payload import HttpManagementPayload, replace_url_origin from .internal.compat.durable_orchestration_status import DurableOrchestrationStatus from .internal.compat.entity_state_response import EntityStateResponse +from .internal.compat.history_projection import project_history from .internal.compat.orchestration_runtime_status import OrchestrationRuntimeStatus, to_durabletask_statuses from .internal.compat.purge_history_result import PurgeHistoryResult @@ -339,15 +340,27 @@ async def get_status( ``OrchestrationState`` for v1 back-compat. When the instance does not exist, a falsy status is returned rather than ``None``. - The ``show_history`` and ``show_history_output`` flags have no - equivalent in durabletask and are ignored. Payloads are fetched to + When ``show_history`` is true, history is fetched and projected into + the v1 status-query shape. ``show_history_output`` controls whether + output-bearing history fields are included. Payloads are fetched to preserve the v1 output, custom-status, and failure-detail fields; ``show_input`` controls whether the compatibility wrapper exposes the - input. + orchestration and history inputs. """ state = await self.get_orchestration_state(instance_id, fetch_payloads=True) + projected_history = None + if (show_history + and state is not None + and state.runtime_status != OrchestrationStatus.PENDING): + history = await self.get_orchestration_history(instance_id) + projected_history = project_history( + history, + show_input=show_input, + show_history_output=show_history_output) return DurableOrchestrationStatus.from_orchestration_state( - state, include_input=show_input) + state, + include_input=show_input, + history=projected_history) @deprecated("get_status_all is deprecated; use get_all_orchestration_states instead.") async def get_status_all(self) -> list[DurableOrchestrationStatus]: diff --git a/azure-functions-durable/azure/durable_functions/internal/compat/durable_orchestration_status.py b/azure-functions-durable/azure/durable_functions/internal/compat/durable_orchestration_status.py index cc6ea3e0..17f6b2b5 100644 --- a/azure-functions-durable/azure/durable_functions/internal/compat/durable_orchestration_status.py +++ b/azure-functions-durable/azure/durable_functions/internal/compat/durable_orchestration_status.py @@ -27,18 +27,23 @@ class DurableOrchestrationStatus: the v1 behaviour where ``get_status`` never returned ``None``. """ - def __init__(self, state: Optional[OrchestrationState] = None): + def __init__( + self, + state: Optional[OrchestrationState] = None, + history: Optional[list[Any]] = None): self._state = state self._include_input = True + self._history = history @classmethod def from_orchestration_state( cls, state: Optional[OrchestrationState], *, - include_input: bool = True) -> "DurableOrchestrationStatus": + include_input: bool = True, + history: Optional[list[Any]] = None) -> "DurableOrchestrationStatus": """Wrap a durabletask ``OrchestrationState`` (or ``None``).""" - status = cls(state) + status = cls(state, history) status._include_input = include_input return status @@ -77,7 +82,8 @@ def _reserialize(value: Any) -> Optional[str]: serialized_custom_status=_reserialize(data.get("customStatus")), failure_details=None, ) - return cls(state) + history = data.get("historyEvents", data.get("history")) + return cls(state, history) def __bool__(self) -> bool: return self._state is not None @@ -145,10 +151,10 @@ def custom_status(self) -> Any: def history(self) -> Optional[list[Any]]: """Get the execution history. - History is not available through this compatibility path and is always - ``None``; use ``get_orchestration_history`` on the client instead. + The history is populated only when requested by passing + ``show_history=True`` to the client's compatibility ``get_status`` API. """ - return None + return self._history def to_json(self) -> dict[str, Any]: """Convert this status into a v1-compatible JSON dictionary. @@ -179,6 +185,8 @@ def to_json(self) -> dict[str, Any]: self._state.serialized_custom_status if self._state is not None else None) if custom_status is not None: result["customStatus"] = custom_status + if self.history is not None: + result["historyEvents"] = self.history return result @staticmethod diff --git a/azure-functions-durable/azure/durable_functions/internal/compat/history_projection.py b/azure-functions-durable/azure/durable_functions/internal/compat/history_projection.py new file mode 100644 index 00000000..f333b8e0 --- /dev/null +++ b/azure-functions-durable/azure/durable_functions/internal/compat/history_projection.py @@ -0,0 +1,191 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT License. + +import json +from dataclasses import fields, is_dataclass +from datetime import datetime, timezone +from typing import Any, cast + +from durabletask import history, task +from durabletask.client import OrchestrationStatus + +from .orchestration_runtime_status import from_durabletask_status + + +def project_history( + events: list[history.HistoryEvent], + *, + show_input: bool, + show_history_output: bool) -> list[dict[str, Any]]: + """Project Durable Task history events into the v1 status-query shape.""" + projected: list[dict[str, Any]] = [] + scheduled_tasks: dict[int, tuple[int, history.TaskScheduledEvent]] = {} + scheduled_sub_orchestrations: dict[ + int, tuple[int, history.SubOrchestrationInstanceCreatedEvent]] = {} + removed_indexes: set[int] = set() + + for event in events: + if isinstance( + event, + (history.OrchestratorStartedEvent, + history.OrchestratorCompletedEvent)): + continue + + item = _project_event( + event, + show_input=show_input, + show_history_output=show_history_output) + projected_index = len(projected) + + if isinstance(event, history.TaskScheduledEvent): + scheduled_tasks[event.event_id] = (projected_index, event) + elif isinstance( + event, + (history.TaskCompletedEvent, history.TaskFailedEvent)): + _add_scheduled_event_data( + item, + event.task_scheduled_id, + scheduled_tasks, + removed_indexes, + show_input) + elif isinstance(event, history.SubOrchestrationInstanceCreatedEvent): + scheduled_sub_orchestrations[event.event_id] = ( + projected_index, event) + elif isinstance( + event, + (history.SubOrchestrationInstanceCompletedEvent, + history.SubOrchestrationInstanceFailedEvent)): + _add_scheduled_event_data( + item, + event.task_scheduled_id, + scheduled_sub_orchestrations, + removed_indexes, + show_input) + + projected.append(item) + + return [ + event for index, event in enumerate(projected) + if index not in removed_indexes + ] + + +def _project_event( + event: history.HistoryEvent, + *, + show_input: bool, + show_history_output: bool) -> dict[str, Any]: + item = { + "EventType": type(event).__name__.removesuffix("Event"), + "Timestamp": _format_datetime(event.timestamp), + } + for field in fields(event): + if field.name in {"event_id", "timestamp"}: + continue + if field.name == "input" and not show_input: + continue + if field.name in {"result", "output"} and not show_history_output: + continue + + value = getattr(event, field.name) + if value is None: + continue + if field.name in {"result", "output"}: + value = _raw_payload(value) + item[_field_name(field.name)] = _serialize(value) + + if isinstance( + event, + (history.TaskScheduledEvent, + history.SubOrchestrationInstanceCreatedEvent)): + item.pop("Version", None) + elif isinstance(event, history.ExecutionStartedEvent): + item["FunctionName"] = item.pop("Name") + for field_name in ( + "OrchestrationInstance", + "ParentInstance", + "Version", + "Tags"): + item.pop(field_name, None) + elif isinstance( + event, + (history.TaskCompletedEvent, + history.TaskFailedEvent, + history.SubOrchestrationInstanceCompletedEvent, + history.SubOrchestrationInstanceFailedEvent)): + item.pop("TaskScheduledId", None) + elif isinstance(event, history.ExecutionCompletedEvent): + try: + status = OrchestrationStatus(event.orchestration_status) + except ValueError: + pass + else: + item["OrchestrationStatus"] = from_durabletask_status(status).name + elif isinstance(event, history.TimerFiredEvent): + item.pop("TimerId", None) + + return item + + +def _add_scheduled_event_data( + item: dict[str, Any], + scheduled_id: int, + scheduled_events: dict[int, tuple[int, Any]], + removed_indexes: set[int], + show_input: bool) -> None: + scheduled = scheduled_events.get(scheduled_id) + if scheduled is None: + return + + scheduled_index, scheduled_event = scheduled + item["ScheduledTime"] = _format_datetime(scheduled_event.timestamp) + item["FunctionName"] = scheduled_event.name + if show_input and scheduled_event.input is not None: + item["Input"] = scheduled_event.input + removed_indexes.add(scheduled_index) + + +def _serialize(value: Any) -> Any: + if isinstance(value, datetime): + return _format_datetime(value) + if isinstance(value, task.FailureDetails): + result = { + "ErrorMessage": value.message, + "ErrorType": value.error_type, + } + if value.stack_trace is not None: + result["StackTrace"] = value.stack_trace + return result + if is_dataclass(value) and not isinstance(value, type): + return { + _field_name(field.name): _serialize(getattr(value, field.name)) + for field in fields(value) + if getattr(value, field.name) is not None + } + if isinstance(value, list): + return [_serialize(item) for item in cast(list[Any], value)] + if isinstance(value, dict): + mapping = cast(dict[Any, Any], value) + return {key: _serialize(item) for key, item in mapping.items()} + return value + + +def _field_name(name: str) -> str: + if name == "scheduled_start_timestamp": + return "ScheduledStartTime" + if name in {"orchestration_span_id", "span_id"}: + return "".join(part.title() for part in name.split("_"))[:-2] + "ID" + return "".join(part.title() for part in name.split("_")) + + +def _raw_payload(serialized: str) -> Any: + try: + return json.loads(serialized) + except (TypeError, ValueError): + return serialized + + +def _format_datetime(value: datetime) -> str: + if value.tzinfo is not None: + value = value.astimezone(timezone.utc).replace(tzinfo=None) + return value.strftime("%Y-%m-%dT%H:%M:%S.%f") + "Z" diff --git a/tests/azure-functions-durable/test_client_compat.py b/tests/azure-functions-durable/test_client_compat.py index fd48bd2f..3a372113 100644 --- a/tests/azure-functions-durable/test_client_compat.py +++ b/tests/azure-functions-durable/test_client_compat.py @@ -19,6 +19,7 @@ from azure.durable_functions.http.http_management_payload import ( replace_url_origin, ) +from durabletask import history as dt_history from durabletask.client import AsyncTaskHubGrpcClient, OrchestrationStatus from durabletask.entities import EntityInstanceId from durabletask.task import RetryPolicy @@ -738,10 +739,14 @@ async def test_get_status_suppresses_only_input_by_default(): client = _make_client() try: state = _fake_state(runtime_status=OrchestrationStatus.COMPLETED) + history_mock = AsyncMock() with patch.object(client, "get_orchestration_state", - new=AsyncMock(return_value=state)): + new=AsyncMock(return_value=state)), \ + patch.object(client, "get_orchestration_history", + new=history_mock): with pytest.warns(DeprecationWarning): status = await client.get_status("abc") + history_mock.assert_not_awaited() assert bool(status) is True assert status.name == "orch" assert status.instance_id == "abc" @@ -796,10 +801,14 @@ async def test_get_status_failed_preserves_failure_output(): async def test_get_status_missing_instance_is_falsy(): client = _make_client() try: + history_mock = AsyncMock() with patch.object(client, "get_orchestration_state", - new=AsyncMock(return_value=None)): + new=AsyncMock(return_value=None)), \ + patch.object(client, "get_orchestration_history", + new=history_mock): with pytest.warns(DeprecationWarning): - status = await client.get_status("missing") + status = await client.get_status("missing", show_history=True) + history_mock.assert_not_awaited() assert bool(status) is False assert status.runtime_status is None assert status.output is None @@ -807,6 +816,156 @@ async def test_get_status_missing_instance_is_falsy(): await client.close() +def _fake_history(): + started_at = datetime(2026, 1, 1, tzinfo=timezone.utc) + return [ + dt_history.OrchestratorStartedEvent( + event_id=-1, timestamp=started_at), + dt_history.ExecutionStartedEvent( + event_id=0, + timestamp=started_at, + name="orch", + version="v1", + input='{"root": true}', + scheduled_start_timestamp=started_at + timedelta(minutes=1)), + dt_history.TaskScheduledEvent( + event_id=1, + timestamp=started_at + timedelta(seconds=1), + name="activity", + version="v1", + input='{"value": 1}'), + dt_history.OrchestratorCompletedEvent( + event_id=-1, + timestamp=started_at + timedelta(seconds=2)), + dt_history.TaskCompletedEvent( + event_id=2, + timestamp=started_at + timedelta(seconds=3), + task_scheduled_id=1, + result='{"result": 2}'), + dt_history.ExecutionCompletedEvent( + event_id=3, + timestamp=started_at + timedelta(seconds=4), + orchestration_status=OrchestrationStatus.COMPLETED.value, + result='{"done": true}'), + ] + + +@pytest.mark.parametrize("show_history_output", [False, True]) +async def test_get_status_projects_v1_history(show_history_output): + client = _make_client() + try: + history_mock = AsyncMock(return_value=_fake_history()) + with patch.object( + client, + "get_orchestration_state", + new=AsyncMock(return_value=_fake_state())), \ + patch.object( + client, + "get_orchestration_history", + new=history_mock): + with pytest.warns(DeprecationWarning): + status = await client.get_status( + "abc", + show_history=True, + show_history_output=show_history_output, + show_input=True) + + history_mock.assert_awaited_once_with("abc") + assert status.history is not None + assert [event["EventType"] for event in status.history] == [ + "ExecutionStarted", + "TaskCompleted", + "ExecutionCompleted", + ] + assert status.history[0]["FunctionName"] == "orch" + assert status.history[0]["Input"] == '{"root": true}' + assert status.history[0]["ScheduledStartTime"] == ( + "2026-01-01T00:01:00.000000Z") + assert "ScheduledStartTimestamp" not in status.history[0] + assert status.history[1]["FunctionName"] == "activity" + assert status.history[1]["Input"] == '{"value": 1}' + assert status.history[1]["ScheduledTime"] == ( + "2026-01-01T00:00:01.000000Z") + assert status.history[2]["OrchestrationStatus"] == "Completed" + if show_history_output: + assert status.history[1]["Result"] == {"result": 2} + assert status.history[2]["Result"] == {"done": True} + else: + assert "Result" not in status.history[1] + assert "Result" not in status.history[2] + assert status.to_json()["historyEvents"] == status.history + finally: + await client.close() + + +async def test_get_status_history_suppresses_inputs(): + client = _make_client() + try: + with patch.object( + client, + "get_orchestration_state", + new=AsyncMock(return_value=_fake_state())), \ + patch.object( + client, + "get_orchestration_history", + new=AsyncMock(return_value=_fake_history())): + with pytest.warns(DeprecationWarning): + status = await client.get_status("abc", show_history=True) + + assert status.history is not None + assert all("Input" not in event for event in status.history) + finally: + await client.close() + + +@pytest.mark.parametrize( + "runtime_status", + [OrchestrationStatus.PENDING, None], +) +async def test_get_status_show_history_skips_unavailable_history(runtime_status): + client = _make_client() + try: + state = None if runtime_status is None else _fake_state(runtime_status) + history_mock = AsyncMock() + with patch.object( + client, + "get_orchestration_state", + new=AsyncMock(return_value=state)), \ + patch.object( + client, + "get_orchestration_history", + new=history_mock): + with pytest.warns(DeprecationWarning): + status = await client.get_status("abc", show_history=True) + + history_mock.assert_not_awaited() + assert status.history is None + finally: + await client.close() + + +async def test_get_status_show_history_fetches_continued_as_new_history(): + client = _make_client() + try: + history_mock = AsyncMock(return_value=[]) + with patch.object( + client, + "get_orchestration_state", + new=AsyncMock(return_value=_fake_state( + OrchestrationStatus.CONTINUED_AS_NEW))), \ + patch.object( + client, + "get_orchestration_history", + new=history_mock): + with pytest.warns(DeprecationWarning): + status = await client.get_status("abc", show_history=True) + + history_mock.assert_awaited_once_with("abc") + assert status.history == [] + finally: + await client.close() + + async def test_get_status_all_returns_wrapped_list(): client = _make_client() try: diff --git a/tests/azure-functions-durable/test_durable_orchestration_status_compat.py b/tests/azure-functions-durable/test_durable_orchestration_status_compat.py index 3ba3c151..15f36810 100644 --- a/tests/azure-functions-durable/test_durable_orchestration_status_compat.py +++ b/tests/azure-functions-durable/test_durable_orchestration_status_compat.py @@ -88,9 +88,26 @@ def test_empty_status_is_falsy_with_none_attributes(): assert status.to_json() == {} -def test_history_is_always_none(): - status = DurableOrchestrationStatus.from_json(_sample_json()) - assert status.history is None +def test_history_events_round_trip(): + source = _sample_json() + source["historyEvents"] = [ + { + "EventType": "ExecutionCompleted", + "Result": {"done": True}, + }, + ] + status = DurableOrchestrationStatus.from_json(source) + assert status.history == source["historyEvents"] + assert status.to_json() == source + + +def test_legacy_history_field_is_accepted(): + status = DurableOrchestrationStatus.from_json({ + **_sample_json(), + "history": [{"EventType": "ExecutionStarted"}], + }) + assert status.history == [{"EventType": "ExecutionStarted"}] + assert status.to_json()["historyEvents"] == status.history # --------------------------------------------------------------------------- From 879f3b5153190f735bd5a4cecf414ef34f4a1707 Mon Sep 17 00:00:00 2001 From: Andy Staples Date: Wed, 29 Jul 2026 14:27:29 -0600 Subject: [PATCH 2/4] Use v1 trace field casing Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: a28a3fa8-303a-4a74-8fbc-8e062e39d97f --- .../internal/compat/history_projection.py | 2 -- tests/azure-functions-durable/test_client_compat.py | 9 ++++++++- 2 files changed, 8 insertions(+), 3 deletions(-) diff --git a/azure-functions-durable/azure/durable_functions/internal/compat/history_projection.py b/azure-functions-durable/azure/durable_functions/internal/compat/history_projection.py index f333b8e0..fcf6ea1d 100644 --- a/azure-functions-durable/azure/durable_functions/internal/compat/history_projection.py +++ b/azure-functions-durable/azure/durable_functions/internal/compat/history_projection.py @@ -173,8 +173,6 @@ def _serialize(value: Any) -> Any: def _field_name(name: str) -> str: if name == "scheduled_start_timestamp": return "ScheduledStartTime" - if name in {"orchestration_span_id", "span_id"}: - return "".join(part.title() for part in name.split("_"))[:-2] + "ID" return "".join(part.title() for part in name.split("_")) diff --git a/tests/azure-functions-durable/test_client_compat.py b/tests/azure-functions-durable/test_client_compat.py index 3a372113..060f8a21 100644 --- a/tests/azure-functions-durable/test_client_compat.py +++ b/tests/azure-functions-durable/test_client_compat.py @@ -827,7 +827,11 @@ def _fake_history(): name="orch", version="v1", input='{"root": true}', - scheduled_start_timestamp=started_at + timedelta(minutes=1)), + scheduled_start_timestamp=started_at + timedelta(minutes=1), + parent_trace_context=dt_history.TraceContext( + trace_parent="trace-parent", + span_id="span-id"), + orchestration_span_id="orchestration-span-id"), dt_history.TaskScheduledEvent( event_id=1, timestamp=started_at + timedelta(seconds=1), @@ -882,6 +886,9 @@ async def test_get_status_projects_v1_history(show_history_output): assert status.history[0]["ScheduledStartTime"] == ( "2026-01-01T00:01:00.000000Z") assert "ScheduledStartTimestamp" not in status.history[0] + assert status.history[0]["ParentTraceContext"]["SpanId"] == "span-id" + assert status.history[0]["OrchestrationSpanId"] == ( + "orchestration-span-id") assert status.history[1]["FunctionName"] == "activity" assert status.history[1]["Input"] == '{"value": 1}' assert status.history[1]["ScheduledTime"] == ( From 7c2ef97f1a411fe500062c4a76e55839d2a688a6 Mon Sep 17 00:00:00 2001 From: Andy Staples Date: Thu, 30 Jul 2026 10:05:50 -0600 Subject: [PATCH 3/4] Preserve v1 failed history fields Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: a28a3fa8-303a-4a74-8fbc-8e062e39d97f --- CHANGELOG.md | 2 + durabletask/history.py | 17 ++++- .../test_client_compat.py | 65 +++++++++++++++---- .../durabletask/test_history_serialization.py | 36 ++++++++++ 4 files changed, 107 insertions(+), 13 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 82584ac1..7a564358 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -41,6 +41,8 @@ import paths, `__all__`, `dir()`, and star-imports behave exactly as before. FIXED +- Failed task and sub-orchestration history events now retain the host-provided + failure `reason` and redacted `details` alongside structured failure details. - Fixed `AsyncTaskHubGrpcClient` failing during construction when no current event loop was set. SDK-owned async gRPC channels are now created on first use, binding them to the event loop that performs the RPC. diff --git a/durabletask/history.py b/durabletask/history.py index a147bd0f..7db84d31 100644 --- a/durabletask/history.py +++ b/durabletask/history.py @@ -94,6 +94,8 @@ class TaskCompletedEvent(HistoryEvent): class TaskFailedEvent(HistoryEvent): task_scheduled_id: int failure_details: task.FailureDetails | None = None + reason: str | None = None + details: str | None = None @dataclass(slots=True) @@ -116,6 +118,8 @@ class SubOrchestrationInstanceCompletedEvent(HistoryEvent): class SubOrchestrationInstanceFailedEvent(HistoryEvent): task_scheduled_id: int failure_details: task.FailureDetails | None = None + reason: str | None = None + details: str | None = None @dataclass(slots=True) @@ -286,6 +290,15 @@ def _failure_details(msg: Message, field_name: str) -> task.FailureDetails | Non ) +def _failure_event_kwargs(msg: Message) -> dict[str, Any]: + failure_details = _failure_details(msg, 'failureDetails') + return { + 'failure_details': failure_details, + 'reason': failure_details.message if failure_details is not None else None, + 'details': failure_details.stack_trace if failure_details is not None else None, + } + + def _trace_context(msg: Message, field_name: str) -> TraceContext | None: if not msg.HasField(field_name): return None @@ -490,7 +503,7 @@ def _to_serializable(value: Any) -> Any: 'taskFailed': lambda event: TaskFailedEvent( **_base_kwargs(event), task_scheduled_id=event.taskFailed.taskScheduledId, - failure_details=_failure_details(event.taskFailed, 'failureDetails'), + **_failure_event_kwargs(event.taskFailed), ), 'subOrchestrationInstanceCreated': lambda event: SubOrchestrationInstanceCreatedEvent( **_base_kwargs(event), @@ -509,7 +522,7 @@ def _to_serializable(value: Any) -> Any: 'subOrchestrationInstanceFailed': lambda event: SubOrchestrationInstanceFailedEvent( **_base_kwargs(event), task_scheduled_id=event.subOrchestrationInstanceFailed.taskScheduledId, - failure_details=_failure_details(event.subOrchestrationInstanceFailed, 'failureDetails'), + **_failure_event_kwargs(event.subOrchestrationInstanceFailed), ), 'timerCreated': lambda event: TimerCreatedEvent( **_base_kwargs(event), diff --git a/tests/azure-functions-durable/test_client_compat.py b/tests/azure-functions-durable/test_client_compat.py index 060f8a21..2a03a0a8 100644 --- a/tests/azure-functions-durable/test_client_compat.py +++ b/tests/azure-functions-durable/test_client_compat.py @@ -19,7 +19,7 @@ from azure.durable_functions.http.http_management_payload import ( replace_url_origin, ) -from durabletask import history as dt_history +from durabletask import history as dt_history, task as dt_task from durabletask.client import AsyncTaskHubGrpcClient, OrchestrationStatus from durabletask.entities import EntityInstanceId from durabletask.task import RetryPolicy @@ -846,9 +846,36 @@ def _fake_history(): timestamp=started_at + timedelta(seconds=3), task_scheduled_id=1, result='{"result": 2}'), - dt_history.ExecutionCompletedEvent( - event_id=3, + dt_history.TaskScheduledEvent( + event_id=4, timestamp=started_at + timedelta(seconds=4), + name="failed_activity", + input='{"value": 2}'), + dt_history.TaskFailedEvent( + event_id=5, + timestamp=started_at + timedelta(seconds=5), + task_scheduled_id=4, + failure_details=dt_task.FailureDetails( + "activity failed", "RuntimeError", "activity details"), + reason="activity failed", + details="activity details"), + dt_history.SubOrchestrationInstanceCreatedEvent( + event_id=6, + timestamp=started_at + timedelta(seconds=6), + instance_id="child", + name="failed_child", + input='{"child": 1}'), + dt_history.SubOrchestrationInstanceFailedEvent( + event_id=7, + timestamp=started_at + timedelta(seconds=7), + task_scheduled_id=6, + failure_details=dt_task.FailureDetails( + "child failed", "RuntimeError", "child details"), + reason="child failed", + details="child details"), + dt_history.ExecutionCompletedEvent( + event_id=8, + timestamp=started_at + timedelta(seconds=8), orchestration_status=OrchestrationStatus.COMPLETED.value, result='{"done": true}'), ] @@ -879,8 +906,13 @@ async def test_get_status_projects_v1_history(show_history_output): assert [event["EventType"] for event in status.history] == [ "ExecutionStarted", "TaskCompleted", + "TaskFailed", + "SubOrchestrationInstanceFailed", "ExecutionCompleted", ] + events_by_type = { + event["EventType"]: event for event in status.history + } assert status.history[0]["FunctionName"] == "orch" assert status.history[0]["Input"] == '{"root": true}' assert status.history[0]["ScheduledStartTime"] == ( @@ -889,17 +921,28 @@ async def test_get_status_projects_v1_history(show_history_output): assert status.history[0]["ParentTraceContext"]["SpanId"] == "span-id" assert status.history[0]["OrchestrationSpanId"] == ( "orchestration-span-id") - assert status.history[1]["FunctionName"] == "activity" - assert status.history[1]["Input"] == '{"value": 1}' - assert status.history[1]["ScheduledTime"] == ( + completed = events_by_type["TaskCompleted"] + assert completed["FunctionName"] == "activity" + assert completed["Input"] == '{"value": 1}' + assert completed["ScheduledTime"] == ( "2026-01-01T00:00:01.000000Z") - assert status.history[2]["OrchestrationStatus"] == "Completed" + task_failed = events_by_type["TaskFailed"] + assert task_failed["Reason"] == "activity failed" + assert task_failed["Details"] == "activity details" + assert task_failed["FailureDetails"]["ErrorMessage"] == ( + "activity failed") + child_failed = events_by_type["SubOrchestrationInstanceFailed"] + assert child_failed["Reason"] == "child failed" + assert child_failed["Details"] == "child details" + assert child_failed["FailureDetails"]["ErrorMessage"] == "child failed" + execution_completed = events_by_type["ExecutionCompleted"] + assert execution_completed["OrchestrationStatus"] == "Completed" if show_history_output: - assert status.history[1]["Result"] == {"result": 2} - assert status.history[2]["Result"] == {"done": True} + assert completed["Result"] == {"result": 2} + assert execution_completed["Result"] == {"done": True} else: - assert "Result" not in status.history[1] - assert "Result" not in status.history[2] + assert "Result" not in completed + assert "Result" not in execution_completed assert status.to_json()["historyEvents"] == status.history finally: await client.close() diff --git a/tests/durabletask/test_history_serialization.py b/tests/durabletask/test_history_serialization.py index 82c8a753..9ce7888a 100644 --- a/tests/durabletask/test_history_serialization.py +++ b/tests/durabletask/test_history_serialization.py @@ -19,8 +19,10 @@ from typing import Any, NamedTuple, cast import pytest +from google.protobuf import timestamp_pb2, wrappers_pb2 from durabletask import history, task +import durabletask.internal.orchestrator_service_pb2 as pb _TS = datetime(2025, 1, 2, 3, 4, 5, 678901, tzinfo=timezone.utc) _NAIVE_TS = datetime(2024, 12, 31, 23, 59, 58) @@ -50,6 +52,40 @@ def _failure(message: str = 'boom') -> task.FailureDetails: return task.FailureDetails(message, 'RuntimeError', 'Traceback...') +@pytest.mark.parametrize( + ("proto_field", "expected_type"), + [ + ("taskFailed", history.TaskFailedEvent), + ("subOrchestrationInstanceFailed", + history.SubOrchestrationInstanceFailedEvent), + ], +) +def test_failed_events_preserve_legacy_failure_fields_from_proto( + proto_field: str, + expected_type: type[history.HistoryEvent]): + timestamp = timestamp_pb2.Timestamp() + timestamp.FromDatetime(_TS) + event = pb.HistoryEvent(eventId=2, timestamp=timestamp) + failed_event = getattr(event, proto_field) + failed_event.taskScheduledId = 1 + failed_event.failureDetails.CopyFrom(pb.TaskFailureDetails( + errorMessage="activity failed", + errorType="RuntimeError", + stackTrace=wrappers_pb2.StringValue(value="redacted details"))) + + converted = history._from_protobuf(event) # pyright: ignore[reportPrivateUsage] + + assert type(converted) is expected_type + assert isinstance( + converted, + (history.TaskFailedEvent, + history.SubOrchestrationInstanceFailedEvent)) + assert converted.reason == "activity failed" + assert converted.details == "redacted details" + assert converted.failure_details == task.FailureDetails( + "activity failed", "RuntimeError", "redacted details") + + @dataclass class _Inner: """A dataclass the walker should convert wherever it is nested.""" From e88238126e97ff7969a46efd54e1d13246cfa76f Mon Sep 17 00:00:00 2001 From: Andy Staples Date: Thu, 30 Jul 2026 10:40:55 -0600 Subject: [PATCH 4/4] Document gRPC history compatibility boundary Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: a28a3fa8-303a-4a74-8fbc-8e062e39d97f --- CHANGELOG.md | 2 -- azure-functions-durable/CHANGELOG.md | 7 ++-- azure-functions-durable/MIGRATION_GUIDE.md | 4 +++ durabletask/history.py | 17 ++------- .../test_client_compat.py | 24 ++++++------- .../durabletask/test_history_serialization.py | 36 ------------------- 6 files changed, 21 insertions(+), 69 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 7a564358..82584ac1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -41,8 +41,6 @@ import paths, `__all__`, `dir()`, and star-imports behave exactly as before. FIXED -- Failed task and sub-orchestration history events now retain the host-provided - failure `reason` and redacted `details` alongside structured failure details. - Fixed `AsyncTaskHubGrpcClient` failing during construction when no current event loop was set. SDK-owned async gRPC channels are now created on first use, binding them to the event loop that performs the RPC. diff --git a/azure-functions-durable/CHANGELOG.md b/azure-functions-durable/CHANGELOG.md index fbfd8233..f745068e 100644 --- a/azure-functions-durable/CHANGELOG.md +++ b/azure-functions-durable/CHANGELOG.md @@ -33,8 +33,11 @@ avoiding repeated allocation of unused worker resources. FIXED - The v1-compatible `get_status()` API now supports `show_history` and - `show_history_output`, including compacted `historyEvents` output without an - additional history request when history is not requested. +`show_history_output`, including compacted `historyEvents` output without an +additional history request when history is not requested. Failed activity and +sub-orchestration events expose the structured `FailureDetails` supplied by +the gRPC host, but not the v1-only top-level `Reason` and `Details` fields, +which are not represented by the shared gRPC history protocol. - Check-status responses now include the standard `Retry-After: 10` polling header. Failed orchestrations return HTTP 200 from the wait helper by default; callers can request HTTP 500 responses through diff --git a/azure-functions-durable/MIGRATION_GUIDE.md b/azure-functions-durable/MIGRATION_GUIDE.md index 3a718750..74acae21 100644 --- a/azure-functions-durable/MIGRATION_GUIDE.md +++ b/azure-functions-durable/MIGRATION_GUIDE.md @@ -269,6 +269,10 @@ instead of calling `context.set_result`. - `DurableOrchestrationContext.histories` is unavailable. Retrieve history with `client.get_orchestration_history(instance_id)`. +- `get_status(show_history=True)` cannot reproduce the v1 top-level `Reason` + and `Details` fields for failed activities or sub-orchestrations because the + shared gRPC history protocol represents only structured `FailureDetails`. + Read `FailureDetails` when present. - Distributed tracing is not yet wired through the Python provider. - Continuous history export is not supported by Azure Functions. - Unusual callable signatures can be misclassified by the compatibility layer. diff --git a/durabletask/history.py b/durabletask/history.py index 7db84d31..a147bd0f 100644 --- a/durabletask/history.py +++ b/durabletask/history.py @@ -94,8 +94,6 @@ class TaskCompletedEvent(HistoryEvent): class TaskFailedEvent(HistoryEvent): task_scheduled_id: int failure_details: task.FailureDetails | None = None - reason: str | None = None - details: str | None = None @dataclass(slots=True) @@ -118,8 +116,6 @@ class SubOrchestrationInstanceCompletedEvent(HistoryEvent): class SubOrchestrationInstanceFailedEvent(HistoryEvent): task_scheduled_id: int failure_details: task.FailureDetails | None = None - reason: str | None = None - details: str | None = None @dataclass(slots=True) @@ -290,15 +286,6 @@ def _failure_details(msg: Message, field_name: str) -> task.FailureDetails | Non ) -def _failure_event_kwargs(msg: Message) -> dict[str, Any]: - failure_details = _failure_details(msg, 'failureDetails') - return { - 'failure_details': failure_details, - 'reason': failure_details.message if failure_details is not None else None, - 'details': failure_details.stack_trace if failure_details is not None else None, - } - - def _trace_context(msg: Message, field_name: str) -> TraceContext | None: if not msg.HasField(field_name): return None @@ -503,7 +490,7 @@ def _to_serializable(value: Any) -> Any: 'taskFailed': lambda event: TaskFailedEvent( **_base_kwargs(event), task_scheduled_id=event.taskFailed.taskScheduledId, - **_failure_event_kwargs(event.taskFailed), + failure_details=_failure_details(event.taskFailed, 'failureDetails'), ), 'subOrchestrationInstanceCreated': lambda event: SubOrchestrationInstanceCreatedEvent( **_base_kwargs(event), @@ -522,7 +509,7 @@ def _to_serializable(value: Any) -> Any: 'subOrchestrationInstanceFailed': lambda event: SubOrchestrationInstanceFailedEvent( **_base_kwargs(event), task_scheduled_id=event.subOrchestrationInstanceFailed.taskScheduledId, - **_failure_event_kwargs(event.subOrchestrationInstanceFailed), + failure_details=_failure_details(event.subOrchestrationInstanceFailed, 'failureDetails'), ), 'timerCreated': lambda event: TimerCreatedEvent( **_base_kwargs(event), diff --git a/tests/azure-functions-durable/test_client_compat.py b/tests/azure-functions-durable/test_client_compat.py index 45bf6f18..80e68f5d 100644 --- a/tests/azure-functions-durable/test_client_compat.py +++ b/tests/azure-functions-durable/test_client_compat.py @@ -968,9 +968,7 @@ def _fake_history(): timestamp=started_at + timedelta(seconds=5), task_scheduled_id=4, failure_details=dt_task.FailureDetails( - "activity failed", "RuntimeError", "activity details"), - reason="activity failed", - details="activity details"), + "activity failed", "RuntimeError", "activity stack trace")), dt_history.SubOrchestrationInstanceCreatedEvent( event_id=6, timestamp=started_at + timedelta(seconds=6), @@ -980,11 +978,7 @@ def _fake_history(): dt_history.SubOrchestrationInstanceFailedEvent( event_id=7, timestamp=started_at + timedelta(seconds=7), - task_scheduled_id=6, - failure_details=dt_task.FailureDetails( - "child failed", "RuntimeError", "child details"), - reason="child failed", - details="child details"), + task_scheduled_id=6), dt_history.ExecutionCompletedEvent( event_id=8, timestamp=started_at + timedelta(seconds=8), @@ -994,7 +988,7 @@ def _fake_history(): @pytest.mark.parametrize("show_history_output", [False, True]) -async def test_get_status_projects_v1_history(show_history_output): +async def test_get_status_projects_available_v1_history(show_history_output): client = _make_client() try: history_mock = AsyncMock(return_value=_fake_history()) @@ -1039,14 +1033,16 @@ async def test_get_status_projects_v1_history(show_history_output): assert completed["ScheduledTime"] == ( "2026-01-01T00:00:01.000000Z") task_failed = events_by_type["TaskFailed"] - assert task_failed["Reason"] == "activity failed" - assert task_failed["Details"] == "activity details" assert task_failed["FailureDetails"]["ErrorMessage"] == ( "activity failed") + assert task_failed["FailureDetails"]["StackTrace"] == ( + "activity stack trace") + assert "Reason" not in task_failed + assert "Details" not in task_failed child_failed = events_by_type["SubOrchestrationInstanceFailed"] - assert child_failed["Reason"] == "child failed" - assert child_failed["Details"] == "child details" - assert child_failed["FailureDetails"]["ErrorMessage"] == "child failed" + assert "Reason" not in child_failed + assert "Details" not in child_failed + assert "FailureDetails" not in child_failed execution_completed = events_by_type["ExecutionCompleted"] assert execution_completed["OrchestrationStatus"] == "Completed" if show_history_output: diff --git a/tests/durabletask/test_history_serialization.py b/tests/durabletask/test_history_serialization.py index 9ce7888a..82c8a753 100644 --- a/tests/durabletask/test_history_serialization.py +++ b/tests/durabletask/test_history_serialization.py @@ -19,10 +19,8 @@ from typing import Any, NamedTuple, cast import pytest -from google.protobuf import timestamp_pb2, wrappers_pb2 from durabletask import history, task -import durabletask.internal.orchestrator_service_pb2 as pb _TS = datetime(2025, 1, 2, 3, 4, 5, 678901, tzinfo=timezone.utc) _NAIVE_TS = datetime(2024, 12, 31, 23, 59, 58) @@ -52,40 +50,6 @@ def _failure(message: str = 'boom') -> task.FailureDetails: return task.FailureDetails(message, 'RuntimeError', 'Traceback...') -@pytest.mark.parametrize( - ("proto_field", "expected_type"), - [ - ("taskFailed", history.TaskFailedEvent), - ("subOrchestrationInstanceFailed", - history.SubOrchestrationInstanceFailedEvent), - ], -) -def test_failed_events_preserve_legacy_failure_fields_from_proto( - proto_field: str, - expected_type: type[history.HistoryEvent]): - timestamp = timestamp_pb2.Timestamp() - timestamp.FromDatetime(_TS) - event = pb.HistoryEvent(eventId=2, timestamp=timestamp) - failed_event = getattr(event, proto_field) - failed_event.taskScheduledId = 1 - failed_event.failureDetails.CopyFrom(pb.TaskFailureDetails( - errorMessage="activity failed", - errorType="RuntimeError", - stackTrace=wrappers_pb2.StringValue(value="redacted details"))) - - converted = history._from_protobuf(event) # pyright: ignore[reportPrivateUsage] - - assert type(converted) is expected_type - assert isinstance( - converted, - (history.TaskFailedEvent, - history.SubOrchestrationInstanceFailedEvent)) - assert converted.reason == "activity failed" - assert converted.details == "redacted details" - assert converted.failure_details == task.FailureDetails( - "activity failed", "RuntimeError", "redacted details") - - @dataclass class _Inner: """A dataclass the walker should convert wherever it is nested."""