diff --git a/azure-functions-durable/CHANGELOG.md b/azure-functions-durable/CHANGELOG.md
index 0aba2344..aadbd839 100644
--- a/azure-functions-durable/CHANGELOG.md
+++ b/azure-functions-durable/CHANGELOG.md
@@ -20,6 +20,11 @@ minimum updates belong in the dedicated release PR.
 
 FIXED
 
+- Reconstructing `HttpRetryOptions` from JSON via `from_json()` now defensively
+handles missing optional fields (`maxRetryIntervalInMilliseconds`,
+`retryTimeoutInMilliseconds`, `backoffCoefficient`, and `statusCodesToRetry`) and
+accepts `snake_case` keys, preventing unexpected `KeyError` exceptions when parsing
+minimal or cross-platform wire payloads.
 - With the corresponding core SDK update, activity and sub-orchestration
 `retry_timeout` decisions use replay-safe orchestration time rather than
 wall-clock time. See the [core upgrade guidance](../CHANGELOG.md#unreleased)
diff --git a/azure-functions-durable/azure/durable_functions/http/models.py b/azure-functions-durable/azure/durable_functions/http/models.py
index 59264e4c..0bcfb49c 100644
--- a/azure-functions-durable/azure/durable_functions/http/models.py
+++ b/azure-functions-durable/azure/durable_functions/http/models.py
@@ -70,16 +70,49 @@ def to_json(self) -> dict[str, Any]:
 
     @classmethod
     def from_json(cls, value: dict[str, Any]) -> "HttpRetryOptions":
-        """Reconstruct a snapshotted policy from its wire payload."""
-        max_interval = value["maxRetryIntervalInMilliseconds"]
-        timeout = value["retryTimeoutInMilliseconds"]
+        """Reconstruct a snapshotted policy from its wire payload.
+
+        Accepts both camelCase and snake_case key naming defensively, and handles
+        missing optional fields (max retry interval, retry timeout, backoff
+        coefficient, and status codes) using their class defaults.
+        """
+        first_interval = (
+            value["firstRetryIntervalInMilliseconds"]
+            if "firstRetryIntervalInMilliseconds" in value
+            else value.get("first_retry_interval_in_milliseconds", 0)
+        )
+        max_attempts = (
+            value["maxNumberOfAttempts"]
+            if "maxNumberOfAttempts" in value
+            else value.get("max_number_of_attempts", 1)
+        )
+        backoff = (
+            value["backoffCoefficient"]
+            if "backoffCoefficient" in value
+            else value.get("backoff_coefficient", 1.0)
+        )
+        max_interval = (
+            value["maxRetryIntervalInMilliseconds"]
+            if "maxRetryIntervalInMilliseconds" in value
+            else value.get("max_retry_interval_in_milliseconds", 518400000)
+        )
+        timeout = (
+            value["retryTimeoutInMilliseconds"]
+            if "retryTimeoutInMilliseconds" in value
+            else value.get("retry_timeout_in_milliseconds")
+        )
+        status_codes = (
+            value["statusCodesToRetry"]
+            if "statusCodesToRetry" in value
+            else value.get("status_codes_to_retry")
+        )
         return cls(
-            first_retry_interval=timedelta(milliseconds=value["firstRetryIntervalInMilliseconds"]),
-            max_number_of_attempts=value["maxNumberOfAttempts"],
-            backoff_coefficient=value["backoffCoefficient"],
+            first_retry_interval=timedelta(milliseconds=first_interval),
+            max_number_of_attempts=max_attempts,
+            backoff_coefficient=backoff if backoff is not None else 1.0,
             max_retry_interval=timedelta(milliseconds=max_interval) if max_interval is not None else None,
             retry_timeout=timedelta(milliseconds=timeout) if timeout is not None else None,
-            status_codes_to_retry=value.get("statusCodesToRetry"))
+            status_codes_to_retry=status_codes)
 
 
 class DurableHttpRequest:
diff --git a/tests/azure-functions-durable/test_http_retry.py b/tests/azure-functions-durable/test_http_retry.py
index 3273bef1..da0d90fb 100644
--- a/tests/azure-functions-durable/test_http_retry.py
+++ b/tests/azure-functions-durable/test_http_retry.py
@@ -49,6 +49,49 @@ def test_http_retry_defaults_and_round_trip():
     assert df.HttpRetryOptions.from_json(json.loads(json.dumps(policy.to_json()))).to_json() == policy.to_json()
 
 
+def test_http_retry_from_json_with_missing_optional_fields():
+    # Only the required fields provided (as when serialized by minimal JSON or external SDKs)
+    wire = {
+        "firstRetryIntervalInMilliseconds": 1500,
+        "maxNumberOfAttempts": 4,
+    }
+    policy = df.HttpRetryOptions.from_json(wire)
+    assert policy.first_retry_interval == timedelta(milliseconds=1500)
+    assert policy.max_number_of_attempts == 4
+    assert policy.backoff_coefficient == 1.0
+    assert policy.max_retry_interval == timedelta(days=6)
+    assert policy.retry_timeout is None
+    assert policy.status_codes_to_retry is None
+
+    # Wire payload explicitly setting maxRetryInterval to null (unbounded)
+    unbounded_wire = {
+        "firstRetryIntervalInMilliseconds": 2000,
+        "maxNumberOfAttempts": 2,
+        "maxRetryIntervalInMilliseconds": None,
+    }
+    unbounded_policy = df.HttpRetryOptions.from_json(unbounded_wire)
+    assert unbounded_policy.max_retry_interval is None
+    assert unbounded_policy.retry_timeout is None
+
+
+def test_http_retry_from_json_with_snake_case_keys():
+    snake_wire = {
+        "first_retry_interval_in_milliseconds": 2500,
+        "max_number_of_attempts": 5,
+        "backoff_coefficient": 1.5,
+        "max_retry_interval_in_milliseconds": 60000,
+        "retry_timeout_in_milliseconds": 120000,
+        "status_codes_to_retry": [429, 503],
+    }
+    policy = df.HttpRetryOptions.from_json(snake_wire)
+    assert policy.first_retry_interval == timedelta(milliseconds=2500)
+    assert policy.max_number_of_attempts == 5
+    assert policy.backoff_coefficient == 1.5
+    assert policy.max_retry_interval == timedelta(seconds=60)
+    assert policy.retry_timeout == timedelta(seconds=120)
+    assert policy.status_codes_to_retry == [429, 503]
+
+
 def test_request_snapshots_policy_and_status_list():
     codes = [429, 503]
     policy = _options(
@@ -393,3 +436,21 @@ def caller(context):
         completion = parent.only_action("completeOrchestration").completeOrchestration
         assert completion.orchestrationStatus == pb.ORCHESTRATION_STATUS_FAILED
         assert "status code 503" in str(completion.failureDetails)
+
+
+def test_poll_orchestrator_accepts_partial_retry_options_payload():
+    payload = {
+        "method": "POST",
+        "uri": "https://example.com/start",
+        "retryOptions": {
+            "firstRetryIntervalInMilliseconds": 2000,
+            "maxNumberOfAttempts": 2,
+        },
+    }
+    history = _History(builtin_http_poll_orchestrator, payload, parent="caller")
+    history.run_activity(status=503)
+    history.fire_timer(timedelta(seconds=2))
+    history.run_activity()
+    assert history.result().status_code == 200
+    assert history.result().content == "body"
+