Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
@@ -1,5 +1,7 @@
# Copyright (c) Microsoft. All rights reserved.

from __future__ import annotations

import sys
import typing
from collections.abc import Callable, Mapping, Sequence
Expand Down Expand Up @@ -399,6 +401,13 @@ def _coerce_dict_to_dataclass(data: dict[str, Any], target_cls: type[Any]) -> An
return data


# ``type(None)`` reports ``builtins`` as its module and ``NoneType`` as its qualified name, but
# ``builtins`` never binds that name: the class is only reachable as ``types.NoneType`` (3.10+)
# or ``type(None)``. The name that ``serialize_type`` produces therefore has to come from this
# table instead of from the module namespace.
_SERIALIZED_TYPE_ALIASES: Mapping[str, type[Any]] = {"builtins.NoneType": type(None)}


def serialize_type(t: type) -> str:
"""Serialize a type to a string.
Expand All @@ -416,7 +425,8 @@ def deserialize_type(
) -> type:
"""Deserialize a serialized type string.
Resolution is limited to exact caller-supplied types or types already present
Resolution is limited to exact caller-supplied types, a fixed alias table for builtin
names that their module never binds (``builtins.NoneType``), or types already present
in loaded module namespaces. This function never imports a module selected by
the serialized value.
Expand All @@ -427,6 +437,7 @@ def deserialize_type(
For example,
deserialize_type("builtins.int") => int
deserialize_type("builtins.NoneType") => type(None)
"""
if allowed_types is not None and serialized_type_string in allowed_types:
resolved = allowed_types[serialized_type_string]
Expand All @@ -436,6 +447,11 @@ def deserialize_type(
raise ValueError(f"allowed_types entry {serialized_type_string!r} does not match the supplied type.")
return resolved

# Types whose serialized name is not an attribute of their module.
alias = _SERIALIZED_TYPE_ALIASES.get(serialized_type_string)
if alias is not None:
return alias

module_name, _, type_name = serialized_type_string.rpartition(".")
module = sys.modules.get(module_name)
if not isinstance(module, ModuleType):
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@
from dataclasses import dataclass, field
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

import pytest

Expand Down Expand Up @@ -51,6 +52,32 @@ class TimedApproval:
issued_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))


@pytest.mark.parametrize(
("request_data", "response_type"),
[(1, int), (None, str), ("payload", type(None))],
)
def test_workflow_event_from_dict_roundtrips_none_type(request_data: Any, response_type: type[Any]) -> None:
"""A request whose data or response is ``None`` survives a to_dict/from_dict round trip.
``type(None)`` serializes as ``builtins.NoneType``, a name the ``builtins`` module does not
bind, so reconstructing the event used to raise AttributeError.
"""
event = WorkflowEvent.request_info(
request_id="request-123",
source_executor_id="review_gateway",
request_data=request_data,
response_type=response_type,
)

restored = WorkflowEvent.from_dict(event.to_dict())

assert restored.request_id == "request-123"
assert restored.source_executor_id == "review_gateway"
assert restored.data == request_data
assert restored.request_type is type(request_data)
assert restored.response_type is response_type


def test_workflow_event_from_dict_accepts_explicit_allowed_types() -> None:
"""Request-info reconstruction accepts exact trusted custom types."""

Expand Down
5 changes: 4 additions & 1 deletion python/packages/core/tests/workflow/test_typing_utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -304,6 +304,8 @@ def test_serialize_type() -> None:
assert serialize_type(dict) == "builtins.dict"
assert serialize_type(tuple) == "builtins.tuple"
assert serialize_type(set) == "builtins.set"
# ``type(None)`` claims the builtins module even though it is bound as ``types.NoneType``.
assert serialize_type(type(None)) == "builtins.NoneType"

# Test custom class
@dataclass
Expand All @@ -326,12 +328,13 @@ def test_deserialize_type() -> None:
assert deserialize_type("builtins.dict") is dict
assert deserialize_type("builtins.tuple") is tuple
assert deserialize_type("builtins.set") is set
assert deserialize_type("builtins.NoneType") is type(None)


def test_serialize_deserialize_roundtrip() -> None:
"""Test that serialization and deserialization are inverse operations."""
# Test built-in types
types_to_test = [int, str, float, bool, list, dict, tuple, set]
types_to_test = [int, str, float, bool, list, dict, tuple, set, type(None)]

for type_to_test in types_to_test:
serialized = serialize_type(type_to_test)
Expand Down
25 changes: 25 additions & 0 deletions python/packages/core/tests/workflow/test_workflow_agent.py
Original file line number Diff line number Diff line change
Expand Up @@ -377,6 +377,31 @@ def test_request_info_dataclass_arguments_are_serialized_when_content_is_created
assert deserialized_args.request_event.data == HandoffRequest(target_agent="helper", reason="overflow")
assert deserialized_args.request_event.response_type is str

def test_request_info_arguments_roundtrip_none_request_data(self) -> None:
"""A request without payload and with a ``None`` response type survives the envelope round trip.
``type(None)`` serializes as ``builtins.NoneType``, a name the ``builtins`` module does not
bind, so rebuilding the arguments used to raise AttributeError for these requests.
"""
executor = SimpleExecutor(id="executor1", response_text="Response")
workflow = WorkflowBuilder(start_executor=executor).build()
agent = WorkflowAgent(workflow=workflow, name="None Request Agent")
event = WorkflowEvent.request_info(
request_id="request_123",
source_executor_id="executor1",
request_data=None,
response_type=type(None),
)

request_function_call = agent._process_request_info_event(event) # pyright: ignore[reportPrivateUsage]

assert isinstance(request_function_call.arguments, dict)
deserialized_args = WorkflowAgent.RequestInfoFunctionArgs.from_dict(request_function_call.arguments)
assert deserialized_args.request_id == "request_123"
assert deserialized_args.request_event.data is None
assert deserialized_args.request_event.request_type is type(None)
assert deserialized_args.request_event.response_type is type(None)

def test_request_info_function_args_from_dict_accepts_explicit_allowed_types(self) -> None:
"""Envelope reconstruction forwards exact trusted custom types."""

Expand Down
Loading