Serialize schedule input as .NET string; address PR review

Persist OrchestrationInput as a JSON-encoded string so the DTS dashboard can deserialize it (it types the field as string?). Also address Copilot review feedback: restrict _from_iso Z-normalization to a trailing Z, use PEP 604 unions in ScheduleStatus.from_dotnet, and preserve the generated execution_token when absent from persisted state.

Author: andystaples

CHANGELOG.md

@@ -13,7 +13,7 @@ CHANGED
 
 FIXED
 
-- Fixed schedules created with `durabletask.scheduled` not appearing in the Durable Task Scheduler dashboard. The schedule entity state is now persisted in a format compatible with the .NET SDK (the `status` is serialized as its numeric enum value and the `interval` as a .NET `TimeSpan` string, using .NET property names), so the dashboard can read it without a JSON deserialization error.
+- Fixed schedules created with `durabletask.scheduled` not appearing in the Durable Task Scheduler dashboard. The schedule entity state is now persisted in a format compatible with the .NET SDK: the `status` is serialized as its numeric enum value, the `interval` as a .NET `TimeSpan` string, and the `orchestration_input` as a JSON-encoded string, all using .NET property names, so the dashboard can read it without a JSON deserialization error.
 
 ## v1.7.0
 

durabletask/scheduled/models.py

@@ -8,9 +8,15 @@
 
 from durabletask.internal.helpers import ensure_aware
 from durabletask.scheduled.schedule_status import ScheduleStatus
+from durabletask.serialization import JsonDataConverter
 
 MINIMUM_INTERVAL = timedelta(seconds=1)
 
+# Serializer used to (de)serialize the orchestration input to/from a JSON string
+# for persistence. Matches the .NET SDK, which stores ``OrchestrationInput`` as a
+# string so the Durable Task Scheduler dashboard can read the raw entity state.
+_INPUT_CONVERTER = JsonDataConverter()
+
 
 def _validate_interval(interval: timedelta) -> timedelta:
     if interval <= timedelta(0):
@@ -29,8 +35,11 @@ def _from_iso(value: str | None) -> datetime | None:
         return None
     # .NET serializes ``DateTimeOffset`` with a numeric offset, but tolerate a
     # trailing ``Z`` so states written by other producers still parse on the
-    # Python versions that predate ``fromisoformat`` accepting ``Z``.
-    return datetime.fromisoformat(value.replace("Z", "+00:00"))
+    # Python versions that predate ``fromisoformat`` accepting ``Z``. Only the
+    # trailing designator is normalized so an interior ``Z`` is left untouched.
+    if value.endswith("Z"):
+        value = value[:-1] + "+00:00"
+    return datetime.fromisoformat(value)
 
 
 def _interval_to_seconds(value: timedelta | None) -> float | None:
@@ -130,6 +139,30 @@ def _parse_interval(data: dict[str, Any]) -> timedelta:
     raise KeyError("interval")
 
 
+def _encode_orchestration_input(value: Any) -> str | None:
+    """Serialize the orchestration input to a JSON string for persistence.
+
+    The Durable Task Scheduler dashboard (and the .NET SDK) model the persisted
+    ``OrchestrationInput`` as a string, so the raw input object is serialized to
+    a JSON string here and parsed back by :func:`_decode_orchestration_input`.
+    """
+    if value is None:
+        return None
+    return _INPUT_CONVERTER.serialize(value)
+
+
+def _decode_orchestration_input(value: Any) -> Any:
+    """Reconstruct the raw orchestration input from its persisted JSON string."""
+    if value is None or not isinstance(value, str):
+        return value
+    try:
+        return _INPUT_CONVERTER.deserialize(value)
+    except (ValueError, TypeError):
+        # Not a JSON document (an unexpected shape); return it unchanged rather
+        # than failing to load the schedule.
+        return value
+
+
 @dataclass
 class ScheduleCreationOptions:
     """Options for creating a new schedule."""
@@ -331,7 +364,7 @@ def to_json(self) -> dict[str, Any]:
             "ScheduleId": self.schedule_id,
             "OrchestrationName": self.orchestration_name,
             "Interval": _interval_to_timespan(self.interval),
-            "OrchestrationInput": self.orchestration_input,
+            "OrchestrationInput": _encode_orchestration_input(self.orchestration_input),
             "OrchestrationInstanceId": self.orchestration_instance_id,
             "StartAt": _to_iso(self.start_at),
             "EndAt": _to_iso(self.end_at),
@@ -345,7 +378,12 @@ def from_json(cls, data: dict[str, Any]) -> "ScheduleConfiguration":
             _get(data, "OrchestrationName", "orchestration_name"),
             _parse_interval(data),
         )
-        config.orchestration_input = _get(data, "OrchestrationInput", "orchestration_input")
+        if "OrchestrationInput" in data:
+            # New .NET-compatible states store the input as a JSON string.
+            config.orchestration_input = _decode_orchestration_input(data["OrchestrationInput"])
+        else:
+            # Legacy states stored the raw (already-parsed) input object.
+            config.orchestration_input = data.get("orchestration_input")
         config.orchestration_instance_id = _get(
             data, "OrchestrationInstanceId", "orchestration_instance_id")
         config.start_at = _from_iso(_get(data, "StartAt", "start_at"))
@@ -396,7 +434,12 @@ def from_json(cls, data: dict[str, Any]) -> "ScheduleState":
         # accept both the .NET-compatible and legacy snake_case shapes.
         state = cls()
         state.status = ScheduleStatus.from_dotnet(_get(data, "Status", "status"))
-        state.execution_token = _get(data, "ExecutionToken", "execution_token")
+        # Preserve the token generated by ``__init__`` when the field is absent;
+        # overwriting it with ``None`` would make every ``run_schedule`` signal
+        # look stale and silently stop the schedule.
+        token = _get(data, "ExecutionToken", "execution_token")
+        if token is not None:
+            state.execution_token = token
         state.last_run_at = _from_iso(_get(data, "LastRunAt", "last_run_at"))
         state.next_run_at = _from_iso(_get(data, "NextRunAt", "next_run_at"))
         state.schedule_created_at = _from_iso(_get(data, "ScheduleCreatedAt", "schedule_created_at"))

durabletask/scheduled/schedule_status.py

@@ -2,7 +2,6 @@
 # Licensed under the MIT License.
 
 from enum import Enum
-from typing import Union
 
 
 class ScheduleStatus(str, Enum):
@@ -29,7 +28,7 @@ def to_dotnet_ordinal(self) -> int:
         return _STATUS_TO_ORDINAL[self]
 
     @classmethod
-    def from_dotnet(cls, value: Union[int, str, None]) -> "ScheduleStatus":
+    def from_dotnet(cls, value: "int | str | None") -> "ScheduleStatus":
         """Reconstruct a status from a persisted value.
 
         Accepts the numeric ordinal written by the .NET-compatible serializer

tests/durabletask/scheduled/test_models.py

@@ -156,6 +156,37 @@ def test_from_json_accepts_legacy_snake_case(self):
         assert restored.interval == timedelta(seconds=5)
         assert restored.start_immediately_if_late is True
 
+    def test_orchestration_input_persisted_as_json_string(self):
+        # .NET / the DTS dashboard model OrchestrationInput as a string, so the
+        # raw input must be serialized to a JSON string in the persisted state.
+        config = ScheduleConfiguration.from_create_options(
+            ScheduleCreationOptions(schedule_id="s1", orchestration_name="orch",
+                                    interval=timedelta(seconds=5),
+                                    orchestration_input="hello"))
+        payload = config.to_json()
+        assert payload["OrchestrationInput"] == '"hello"'
+        restored = ScheduleConfiguration.from_json(payload)
+        assert restored.orchestration_input == "hello"
+
+    def test_orchestration_input_dict_round_trips(self):
+        config = ScheduleConfiguration.from_create_options(
+            ScheduleCreationOptions(schedule_id="s1", orchestration_name="orch",
+                                    interval=timedelta(seconds=5),
+                                    orchestration_input={"key": "value"}))
+        payload = config.to_json()
+        assert isinstance(payload["OrchestrationInput"], str)
+        restored = ScheduleConfiguration.from_json(payload)
+        assert restored.orchestration_input == {"key": "value"}
+
+    def test_orchestration_input_none_stays_none(self):
+        config = ScheduleConfiguration.from_create_options(
+            ScheduleCreationOptions(schedule_id="s1", orchestration_name="orch",
+                                    interval=timedelta(seconds=5)))
+        payload = config.to_json()
+        assert payload["OrchestrationInput"] is None
+        restored = ScheduleConfiguration.from_json(payload)
+        assert restored.orchestration_input is None
+
 
 class TestScheduleState:
     def test_round_trip_and_description(self):
@@ -213,6 +244,12 @@ def test_refresh_execution_token_changes_token(self):
         state.refresh_execution_token()
         assert state.execution_token != original
 
+    def test_from_json_preserves_default_token_when_missing(self):
+        # A payload without an execution token must not clobber the token
+        # generated by ``__init__``; otherwise every run signal looks stale.
+        restored = ScheduleState.from_json({"Status": 1})
+        assert restored.execution_token
+
 
 class TestScheduleQueryNormalization:
     def test_naive_bounds_are_coerced_to_aware_utc(self):