Implement v1 show_history status compatibility

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.

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.

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]:

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

azure-functions-durable/azure/durable_functions/internal/compat/history_projection.py

@@ -0,0 +1,189 @@
+# 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"
+    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"