Run complete Python SDK qualification on landed main commits (#236)

Author: durable-workflow-ops

src/durable_workflow/__init__.py

@@ -78,12 +78,6 @@
     WorkflowPayloadDecodeError,
     WorkflowTerminated,
 )
-from .nexus import (
-    NEXUS_CALLER_SDK_LANGUAGE,
-    NEXUS_OPERATION_RESULT_SCHEMA,
-    NEXUS_OPERATION_RESULT_VERSION,
-    NexusOperationResult,
-)
 from .external_storage import (
     EXTERNAL_PAYLOAD_REFERENCE_SCHEMA,
     AzureBlobExternalStorage,
@@ -125,6 +119,24 @@
     parse_external_task_result,
     parse_external_task_result_artifact,
 )
+from .history_bundle_verify import (
+    BUNDLE_SCHEMA as HISTORY_BUNDLE_SCHEMA,
+)
+from .history_bundle_verify import (
+    BUNDLE_SCHEMA_VERSION as HISTORY_BUNDLE_SCHEMA_VERSION,
+)
+from .history_bundle_verify import (
+    REPORT_SCHEMA as HISTORY_BUNDLE_VERIFICATION_REPORT_SCHEMA,
+)
+from .history_bundle_verify import (
+    REPORT_SCHEMA_VERSION as HISTORY_BUNDLE_VERIFICATION_REPORT_SCHEMA_VERSION,
+)
+from .history_bundle_verify import (
+    verify_bundle as verify_history_bundle,
+)
+from .history_bundle_verify import (
+    verify_bundle_json as verify_history_bundle_json,
+)
 from .interceptors import (
     ActivityHandler,
     ActivityInterceptorContext,
@@ -149,17 +161,16 @@
     NoopMetrics,
     PrometheusMetrics,
 )
-from .retry_policy import RetryPolicy, TransportRetryPolicy
-from .serializer import (
-    PayloadSizeWarningConfig,
-    PayloadSizeWarningContext,
-    external_storage_envelope,
-    to_avro_payload_value,
-    to_avro_payload_values,
+from .nexus import (
+    NEXUS_CALLER_SDK_LANGUAGE,
+    NEXUS_OPERATION_RESULT_SCHEMA,
+    NEXUS_OPERATION_RESULT_VERSION,
+    NexusOperationResult,
 )
-from .worker import Worker
 from .replay_verify import (
     CaseReport as ReplayCaseReport,
+)
+from .replay_verify import (
     GoldenHistoryReport,
     SimulationReport,
     aggregate_verdicts,
@@ -169,14 +180,15 @@
     verify_golden_history,
     verify_replay,
 )
-from .history_bundle_verify import (
-    BUNDLE_SCHEMA as HISTORY_BUNDLE_SCHEMA,
-    BUNDLE_SCHEMA_VERSION as HISTORY_BUNDLE_SCHEMA_VERSION,
-    REPORT_SCHEMA as HISTORY_BUNDLE_VERIFICATION_REPORT_SCHEMA,
-    REPORT_SCHEMA_VERSION as HISTORY_BUNDLE_VERIFICATION_REPORT_SCHEMA_VERSION,
-    verify_bundle as verify_history_bundle,
-    verify_bundle_json as verify_history_bundle_json,
+from .retry_policy import RetryPolicy, TransportRetryPolicy
+from .serializer import (
+    PayloadSizeWarningConfig,
+    PayloadSizeWarningContext,
+    external_storage_envelope,
+    to_avro_payload_value,
+    to_avro_payload_values,
 )
+from .worker import Worker
 from .workflow import (
     ActivityRetryPolicy,
     ChildWorkflowRetryPolicy,

src/durable_workflow/client.py

@@ -77,10 +77,12 @@ def _worker_poll_timeout_seconds(timeout: float | None) -> int | None:
 
 
 def _worker_poll_http_timeout(timeout: float | None) -> float | None:
-    timeout_seconds = _worker_poll_timeout_seconds(timeout)
-    if timeout_seconds is None:
+    if timeout is None:
         return None
 
+    timeout_seconds = _worker_poll_timeout_seconds(timeout)
+    assert timeout_seconds is not None
+
     if timeout_seconds == 0:
         return max(float(timeout), 1.0)
 

src/durable_workflow/history_bundle_verify.py

@@ -141,7 +141,8 @@ def _check_envelope(bundle: Mapping[str, Any], findings: list[dict[str, Any]]) -
             _finding(
                 "bundle.schema_version_unsupported",
                 SEVERITY_ERROR,
-                f"Bundle schema_version [{bundle['schema_version']!s}] is unsupported; expected {BUNDLE_SCHEMA_VERSION}.",
+                f"Bundle schema_version [{bundle['schema_version']!s}] is unsupported; "
+                f"expected {BUNDLE_SCHEMA_VERSION}.",
                 {"observed": bundle["schema_version"]},
             )
         )
@@ -325,7 +326,8 @@ def _check_commands(bundle: Mapping[str, Any], findings: list[dict[str, Any]]) -
     if not isinstance(commands, list):
         return
 
-    events = bundle.get("history_events") if isinstance(bundle.get("history_events"), list) else []
+    raw_events = bundle.get("history_events")
+    events: list[Any] = raw_events if isinstance(raw_events, list) else []
     event_command_ids: set[str] = set()
     for event in events:
         if isinstance(event, Mapping):

src/durable_workflow/python_conformance.py

@@ -16,7 +16,9 @@
 import sys
 from collections.abc import Iterable, Mapping
 from pathlib import Path
-from typing import Any
+from typing import Any, TypeVar
+
+_T = TypeVar("_T")
 
 SCHEMA = "durable-workflow.v2.python-sdk-parity.contract"
 VERSION = 1
@@ -761,7 +763,7 @@ def _trace_plane_entries(value: Any, plane: str) -> list[Any]:
     return _filtered_traces(_trace_entries(value), plane)
 
 
-def _filtered_traces(traces: Any, plane: str) -> Any:
+def _filtered_traces(traces: Any, plane: str) -> list[Any]:
     if not isinstance(traces, list):
         return []
     return [
@@ -820,7 +822,7 @@ def _copy_mapping(value: Any) -> dict[str, Any]:
     return dict(value) if isinstance(value, Mapping) else {}
 
 
-def _deepcopy_json_like(value: Any) -> Any:
+def _deepcopy_json_like(value: _T) -> _T:
     return copy.deepcopy(value)
 
 

src/durable_workflow/replay_verify.py

@@ -29,16 +29,16 @@
 import json
 import sys
 import traceback
-from collections.abc import Iterable, Mapping, Sequence
+from collections.abc import Callable, Iterable, Mapping, Sequence
 from dataclasses import dataclass, field
 from pathlib import Path
-from typing import Any, Callable
+from typing import Any
 
 from .errors import ChildWorkflowFailed, NonDeterministicReplayError, WorkflowFailed
 from .workflow import (
     CompleteWorkflow,
-    ReplayOutcome,
     Replayer,
+    ReplayOutcome,
     ScheduleActivity,
     StartChildWorkflow,
 )

src/durable_workflow/worker.py

@@ -301,10 +301,7 @@ def _query_history_with_export_signal_arguments(
         if isinstance(name, str) and name:
             signals_by_name.setdefault(name, []).append(raw_signal)
 
-    if not signals_by_id and not signals_by_command_id and not signals_by_name:
-        signals_available = False
-    else:
-        signals_available = True
+    signals_available = bool(signals_by_id or signals_by_command_id or signals_by_name)
 
     activity_results_by_sequence = _activity_result_by_sequence_from_export(history_export)
 
@@ -431,10 +428,7 @@ def _query_history_events(
     *,
     default_codec: str | None,
 ) -> Any:
-    if isinstance(history, list):
-        events = history
-    else:
-        events = []
+    events = history if isinstance(history, list) else []
 
     if isinstance(history_export, Mapping):
         export_events = history_export.get("history_events")
@@ -1956,10 +1950,8 @@ async def _heartbeat_loop(self) -> None:
         considered for task dispatch when they miss enough heartbeats.
         """
         while not self._stop.is_set():
-            try:
+            with contextlib.suppress(asyncio.TimeoutError):
                 await asyncio.wait_for(self._stop.wait(), timeout=self._heartbeat_interval)
-            except asyncio.TimeoutError:
-                pass
             if self._stop.is_set():
                 return
             try:

src/durable_workflow/workflow.py

@@ -392,12 +392,16 @@ def _payload_blobs_or_external_envelopes(
             payloads.append(blob if isinstance(blob, str) else str(blob))
         return payloads
 
-    return serializer.encode_many(
-        values,
-        codec=payload_codec,
-        size_warning=size_warning,
-        warning_context=warning_contexts,
+    encoded_payloads: list[str | dict[str, Any]] = []
+    encoded_payloads.extend(
+        serializer.encode_many(
+            values,
+            codec=payload_codec,
+            size_warning=size_warning,
+            warning_context=warning_contexts,
+        )
     )
+    return encoded_payloads
 
 
 @dataclass
@@ -2537,7 +2541,7 @@ def _receiver_condition_wait_bindings() -> dict[int, str | None]:
         prefix_receivers: list[int] = []
         prefix_can_bind_to_first_wait = True
         current_wait_id: str | None = None
-        receivers_since_wait: list[int | None] = []
+        receivers_since_wait: list[int] = []
 
         for index, event in enumerate(events):
             event_type = _history_event_type(event)
@@ -3013,11 +3017,10 @@ def _terminal_state(value: Any, *, include_pending: bool) -> _ReplayState:
                             prefix="wait_condition predicate raised",
                         )])
                     if resolution == "satisfied":
-                        if has_reopened_same_wait:
-                            if not satisfied:
-                                consumed_reopen_for_current_wait = True
-                                wait_yield_count = next_wait_index
-                                continue
+                        if has_reopened_same_wait and not satisfied:
+                            consumed_reopen_for_current_wait = True
+                            wait_yield_count = next_wait_index
+                            continue
 
                         terminal_condition_reopen_cmd = (
                             cmd

tests/test_release_docs_audit_workflow.py

@@ -1,7 +1,6 @@
 import re
 from pathlib import Path
 
-
 REPO_ROOT = Path(__file__).resolve().parents[1]
 
 

tests/test_replay_conformance.py

@@ -9,6 +9,8 @@
     RESULT_SCHEMA,
     RESULT_VERSION,
     compose_report,
+)
+from durable_workflow.replay_conformance import (
     main as replay_conformance_main,
 )
 

tests/test_replay_verify.py

@@ -29,13 +29,15 @@
     VERDICT_FAILED,
     VERDICT_OK,
     aggregate_verdicts,
-    main as replay_verify_main,
     promotion_decision_for,
     promotion_decision_for_report,
     simulate_bundles,
     verify_golden_history,
     verify_replay,
 )
+from durable_workflow.replay_verify import (
+    main as replay_verify_main,
+)
 from durable_workflow.workflow import WorkflowContext