Python: Allow checkpoint deserialization for orchestrations

python/packages/core/agent_framework/_workflows/_checkpoint_encoding.py

@@ -68,8 +68,11 @@
 # Types that are natively JSON-serializable and don't need pickling
 _JSON_NATIVE_TYPES = (str, int, float, bool, type(None))
 
-# Module prefix for framework-internal types that are always allowed
-_FRAMEWORK_MODULE_PREFIX = "agent_framework."
+# Module prefixes for framework-internal types that are always allowed
+_FRAMEWORK_MODULE_PREFIXES = (
+    "agent_framework.",
+    "agent_framework_orchestrations.",
+)
 
 # Module prefix for OpenAI SDK types that are always allowed
 _OPENAI_MODULE_PREFIX = "openai.types."
@@ -150,7 +153,7 @@ def _is_allowed_type(self, resolved: type) -> bool:
         return (
             type_key in _BUILTIN_ALLOWED_TYPE_KEYS
             or type_key in self._allowed_types
-            or resolved.__module__.startswith(_FRAMEWORK_MODULE_PREFIX)
+            or resolved.__module__.startswith(_FRAMEWORK_MODULE_PREFIXES)
             or resolved.__module__.startswith(_OPENAI_MODULE_PREFIX)
         )
 
@@ -197,7 +200,7 @@ def find_class(self, module: str, name: str) -> Any:
                 return resolved
             raise pickle.UnpicklingError(f"Checkpoint deserialization blocked for non-type global '{type_key}'.")
 
-        if module.startswith(_FRAMEWORK_MODULE_PREFIX) or module.startswith(_OPENAI_MODULE_PREFIX):
+        if module.startswith(_FRAMEWORK_MODULE_PREFIXES) or module.startswith(_OPENAI_MODULE_PREFIX):
             # Pickle dotted names traverse attributes on an allowed module; keep the prefix allowlist to concrete
             # top-level classes rather than helper callables reachable through module attributes.
             if "." in name:

python/packages/core/tests/workflow/test_checkpoint_orchestrations_deserialization.py

@@ -0,0 +1,37 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+import pytest
+
+pytest.importorskip("agent_framework_orchestrations")
+
+from agent_framework import AgentResponse, Message
+from agent_framework._workflows._checkpoint_encoding import decode_checkpoint_value, encode_checkpoint_value
+from agent_framework.orchestrations import HandoffAgentUserRequest, MagenticPlanReviewRequest
+
+
+@pytest.mark.parametrize(
+    ("request_value", "request_type"),
+    [
+        (
+            HandoffAgentUserRequest(
+                agent_response=AgentResponse(messages=[Message("assistant", ["handoff response"])])
+            ),
+            HandoffAgentUserRequest,
+        ),
+        (
+            MagenticPlanReviewRequest(
+                plan=Message("assistant", ["review this plan"]),
+                current_progress=None,
+                is_stalled=False,
+            ),
+            MagenticPlanReviewRequest,
+        ),
+    ],
+)
+def test_restricted_decode_roundtrips_orchestration_requests(request_value: object, request_type: type[object]) -> None:
+    """Pending orchestration requests can be restored from a restricted checkpoint."""
+    encoded = encode_checkpoint_value(request_value)
+
+    decoded = decode_checkpoint_value(encoded, allowed_types=frozenset())
+
+    assert isinstance(decoded, request_type)