feat: add execution history event generation and pagination

- Add event_conversion module to transform operations into history events
- Implement operation_to_started_event and operation_to_finished_event
- Add generate_execution_events for complete execution history
- Update executor get_execution_history with proper cursor-based pagination
- Support reverse_order and include_execution_data options
- Add comprehensive test coverage for event conversion and pagination

Author: Rares Polenciuc

src/aws_durable_execution_sdk_python_testing/event_conversion.py

@@ -0,0 +1,360 @@
+"""Event conversion utilities for durable execution history.
+
+This module provides functions to convert operation objects into history events
+that can be consumed by the execution history API.
+"""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+from typing import TYPE_CHECKING
+
+from aws_durable_execution_sdk_python.execution import InvocationStatus
+from aws_durable_execution_sdk_python.lambda_service import (
+    OperationStatus,
+    OperationType,
+)
+
+from aws_durable_execution_sdk_python_testing.model import Event as HistoryEvent
+
+
+if TYPE_CHECKING:
+    from aws_durable_execution_sdk_python.lambda_service import Operation
+
+    from aws_durable_execution_sdk_python_testing.execution import Execution
+
+
+def _iso(ts, *, default_now: bool = False) -> str:
+    """Convert timestamp to ISO format string.
+
+    Args:
+        ts: Timestamp to convert
+        default_now: If True, use current time when ts is None
+
+    Returns:
+        ISO format timestamp string
+
+    Raises:
+        ValueError: If ts is None and default_now is False
+    """
+    if ts is None:
+        if default_now:
+            return datetime.now(UTC).isoformat()
+        msg = "timestamp is required"
+        raise ValueError(msg)
+    if ts.tzinfo is None:
+        ts = ts.replace(tzinfo=UTC)
+    return ts.isoformat()
+
+
+def _payload_envelope(payload, include: bool):  # noqa: FBT001
+    """Create payload envelope for event data.
+
+    Args:
+        payload: Payload data to wrap
+        include: Whether to include the actual payload data
+
+    Returns:
+        Payload envelope dict or None if payload is None
+    """
+    if payload is None:
+        return None
+    return {"Payload": payload if include else None, "Truncated": False}
+
+
+def _error_envelope(err):
+    """Create error envelope for event data.
+
+    Args:
+        err: Error object to wrap
+
+    Returns:
+        Error envelope dict or None if err is None
+    """
+    if err is None:
+        return None
+    return {
+        "Payload": err.to_dict() if hasattr(err, "to_dict") else err,
+        "Truncated": False,
+    }
+
+
+def _wait_seconds_from(op: Operation) -> int | None:
+    """Extract wait duration from operation.
+
+    Args:
+        op: Operation to extract wait duration from
+
+    Returns:
+        Wait duration in seconds or None if not available
+    """
+    if hasattr(op, "wait_options") and op.wait_options:
+        v = op.wait_options.get("wait_seconds")
+        if v is not None:
+            return int(v)
+    if op.wait_details and hasattr(op.wait_details, "wait_seconds"):
+        v = op.wait_details.wait_seconds
+        if v is not None:
+            return int(v)
+    return None
+
+
+def _scheduled_end_iso(op: Operation) -> str | None:
+    """Extract scheduled end timestamp from operation.
+
+    Args:
+        op: Operation to extract timestamp from
+
+    Returns:
+        ISO format timestamp string or None if not available
+    """
+    wd = op.wait_details
+    ts = getattr(wd, "scheduled_timestamp", None) if wd else None
+    return _iso(ts) if ts else None
+
+
+def _with_event_id(d: dict, eid: int) -> dict:
+    """Add event ID to event dictionary.
+
+    Args:
+        d: Event dictionary
+        eid: Event ID to add
+
+    Returns:
+        Updated dictionary with event ID
+    """
+    out = dict(d)
+    out["EventId"] = eid
+    return out
+
+
+def operation_to_started_event(
+    operation: Operation,
+    execution=None,
+    include_execution_data: bool = False,  # noqa: FBT001, FBT002
+) -> HistoryEvent:
+    """Convert operation to started event.
+
+    Args:
+        operation: Operation to convert
+        execution: Optional execution context
+        include_execution_data: Whether to include execution data in the event
+
+    Returns:
+        History event for operation start
+
+    Raises:
+        ValueError: If operation type is unknown
+    """
+    base = {
+        "EventTimestamp": _iso(operation.start_timestamp, default_now=True),
+        "Id": operation.operation_id,
+        "Name": operation.name,
+        "ParentId": operation.parent_id,
+    }
+
+    match operation.operation_type:
+        case OperationType.EXECUTION:
+            input_payload = (
+                operation.execution_details.input_payload
+                if operation.execution_details
+                else None
+            )
+            details = {
+                "Input": _payload_envelope(input_payload, include_execution_data),
+                "ExecutionTimeout": (
+                    getattr(
+                        getattr(execution, "start_input", None),
+                        "execution_timeout_seconds",
+                        None,
+                    )
+                    if execution
+                    else None
+                ),
+            }
+            return HistoryEvent.from_dict(
+                {
+                    **base,
+                    "EventType": "ExecutionStarted",
+                    "ExecutionStartedDetails": details,
+                }
+            )
+
+        case OperationType.STEP:
+            return HistoryEvent.from_dict(
+                {
+                    **base,
+                    "EventType": "StepStarted",
+                    "StepStartedDetails": {"Present": True},
+                }
+            )
+
+        case OperationType.WAIT:
+            details = {
+                "Duration": _wait_seconds_from(operation),
+                "ScheduledEndTimestamp": _scheduled_end_iso(operation),
+            }
+            return HistoryEvent.from_dict(
+                {
+                    **base,
+                    "EventType": "WaitStarted",
+                    "WaitStartedDetails": details,
+                }
+            )
+
+        case OperationType.CALLBACK:
+            cb = getattr(operation, "callback_details", None)
+            details = {
+                "CallbackId": getattr(cb, "callback_id", None),
+            }
+            return HistoryEvent.from_dict(
+                {
+                    **base,
+                    "EventType": "CallbackStarted",
+                    "CallbackStartedDetails": details,
+                }
+            )
+
+        case OperationType.INVOKE:
+            return HistoryEvent.from_dict(
+                {
+                    **base,
+                    "EventType": "InvokeStarted",
+                    "InvokeStartedDetails": {
+                        "Input": {"Payload": None, "Truncated": False}
+                    },
+                }
+            )
+
+        case _:
+            msg = f"Unknown operation type: {operation.operation_type}"
+            raise ValueError(msg)
+
+
+def operation_to_finished_event(
+    operation: Operation,
+    execution=None,
+    include_execution_data: bool = False,  # noqa: FBT001, FBT002
+) -> HistoryEvent | None:
+    """Convert operation to finished event.
+
+    Args:
+        operation: Operation to convert
+        execution: Optional execution context
+        include_execution_data: Whether to include execution data in the event
+
+    Returns:
+        History event for operation completion or None if operation not finished
+    """
+    if not operation.end_timestamp:
+        return None
+
+    base = {
+        "EventTimestamp": _iso(operation.end_timestamp),
+        "Id": operation.operation_id,
+        "Name": operation.name,
+        "ParentId": operation.parent_id,
+    }
+
+    match operation.operation_type:
+        case OperationType.EXECUTION if (
+            execution
+            and getattr(execution, "is_complete", False)
+            and getattr(execution, "result", None)
+        ):
+            res = execution.result
+            if res.status == InvocationStatus.SUCCEEDED:
+                details = {
+                    "Result": _payload_envelope(
+                        getattr(res, "result", None), include_execution_data
+                    )
+                }
+                return HistoryEvent.from_dict(
+                    {
+                        **base,
+                        "EventType": "ExecutionSucceeded",
+                        "ExecutionSucceededDetails": details,
+                    }
+                )
+            details = {"Error": _error_envelope(getattr(res, "error", None))}
+            return HistoryEvent.from_dict(
+                {
+                    **base,
+                    "EventType": "ExecutionFailed",
+                    "ExecutionFailedDetails": details,
+                }
+            )
+
+        case OperationType.WAIT if operation.status == OperationStatus.SUCCEEDED:
+            details = {"Duration": _wait_seconds_from(operation)}
+            return HistoryEvent.from_dict(
+                {**base, "EventType": "WaitSucceeded", "WaitSucceededDetails": details}
+            )
+
+        case OperationType.STEP:
+            sd = operation.step_details
+            if operation.status == OperationStatus.SUCCEEDED:
+                result = getattr(sd, "result", None) if sd else None
+                details = (
+                    {"Result": _payload_envelope(result, include_execution_data)}
+                    if result is not None
+                    else {"Result": None}
+                )
+                return HistoryEvent.from_dict(
+                    {
+                        **base,
+                        "EventType": "StepSucceeded",
+                        "StepSucceededDetails": details,
+                    }
+                )
+            if operation.status == OperationStatus.FAILED:
+                err = getattr(sd, "error", None) if sd else None
+                details = (
+                    {"Error": _error_envelope(err)}
+                    if err is not None
+                    else {"Error": None}
+                )
+                return HistoryEvent.from_dict(
+                    {**base, "EventType": "StepFailed", "StepFailedDetails": details}
+                )
+            return None
+
+        case OperationType.CALLBACK:
+            return None
+        case OperationType.INVOKE:
+            return None
+
+        case _:
+            return None
+
+
+def generate_execution_events(
+    execution: Execution,
+    include_execution_data: bool = False,  # noqa: FBT001, FBT002
+) -> list[HistoryEvent]:
+    """Generate history events for an execution.
+
+    Walk all operations on an execution and emit start + (if present) finish events.
+    EventIds are strictly increasing in emission order.
+
+    Args:
+        execution: Execution to generate events for
+        include_execution_data: Whether to include execution data in events
+
+    Returns:
+        List of history events in chronological order
+    """
+    events: list[HistoryEvent] = []
+    eid = 1
+    for op in getattr(execution, "operations", []):
+        started = operation_to_started_event(op, execution, include_execution_data)
+        started = HistoryEvent.from_dict(_with_event_id(started.to_dict(), eid))
+        events.append(started)
+        eid += 1
+
+        finished = operation_to_finished_event(op, execution, include_execution_data)
+        if finished is not None:
+            finished = HistoryEvent.from_dict(_with_event_id(finished.to_dict(), eid))
+            events.append(finished)
+            eid += 1
+    return events