Fix rebase issues: restore missing _conversation_state.py and checkpoint decode logic

- Add back _conversation_state.py (encode/decode_chat_messages) lost in rebase
- Fix on_checkpoint_restore to decode cache/conversation with decode_chat_messages
- Fix on_checkpoint_restore to use decode_checkpoint_value for pending requests
- Add tests/workflow/__init__.py for relative import support
- Fix test_agent_executor checkpoint selection (checkpoints[1] not superstep)

Author: eavanvalkenburg

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

@@ -14,7 +14,7 @@
 from .._sessions import AgentSession
 from .._types import AgentResponse, AgentResponseUpdate, Message
 from ._agent_utils import resolve_agent_id
-from ._checkpoint_encoding import encode_checkpoint_value
+from ._checkpoint_encoding import decode_checkpoint_value, encode_checkpoint_value
 from ._const import WORKFLOW_RUN_KWARGS_KEY
 from ._conversation_state import encode_chat_messages
 from ._executor import Executor, handler
@@ -245,11 +245,27 @@ async def on_checkpoint_restore(self, state: dict[str, Any]) -> None:
         Args:
             state: Checkpoint data dict
         """
+        from ._conversation_state import decode_chat_messages
+
         cache_payload = state.get("cache")
-        self._cache = cache_payload or []
+        if cache_payload:
+            try:
+                self._cache = decode_chat_messages(cache_payload)
+            except Exception as exc:
+                logger.warning("Failed to restore cache: %s", exc)
+                self._cache = []
+        else:
+            self._cache = []
 
         full_conversation_payload = state.get("full_conversation")
-        self._full_conversation = full_conversation_payload or []
+        if full_conversation_payload:
+            try:
+                self._full_conversation = decode_chat_messages(full_conversation_payload)
+            except Exception as exc:
+                logger.warning("Failed to restore full conversation: %s", exc)
+                self._full_conversation = []
+        else:
+            self._full_conversation = []
 
         session_payload = state.get("agent_session")
         if session_payload:
@@ -263,11 +279,11 @@ async def on_checkpoint_restore(self, state: dict[str, Any]) -> None:
 
         pending_requests_payload = state.get("pending_agent_requests")
         if pending_requests_payload:
-            self._pending_agent_requests = pending_requests_payload
+            self._pending_agent_requests = decode_checkpoint_value(pending_requests_payload)
 
         pending_responses_payload = state.get("pending_responses_to_agent")
         if pending_responses_payload:
-            self._pending_responses_to_agent = pending_responses_payload
+            self._pending_responses_to_agent = decode_checkpoint_value(pending_responses_payload)
 
     def reset(self) -> None:
         """Reset the internal cache of the executor."""

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

@@ -0,0 +1,75 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+from collections.abc import Iterable
+from typing import Any, cast
+
+from agent_framework import Message
+
+from ._checkpoint_encoding import decode_checkpoint_value, encode_checkpoint_value
+
+"""Utilities for serializing and deserializing chat conversations for persistence.
+
+These helpers convert rich `Message` instances to checkpoint-friendly payloads
+using the same encoding primitives as the workflow runner. This preserves
+`additional_properties` and other metadata without relying on unsafe mechanisms
+such as pickling.
+"""
+
+
+def encode_chat_messages(messages: Iterable[Message]) -> list[dict[str, Any]]:
+    """Serialize chat messages into checkpoint-safe payloads."""
+    encoded: list[dict[str, Any]] = []
+    for message in messages:
+        encoded.append({
+            "role": encode_checkpoint_value(message.role),
+            "contents": [encode_checkpoint_value(content) for content in message.contents],
+            "author_name": message.author_name,
+            "message_id": message.message_id,
+            "additional_properties": {
+                key: encode_checkpoint_value(value) for key, value in message.additional_properties.items()
+            },
+        })
+    return encoded
+
+
+def decode_chat_messages(payload: Iterable[dict[str, Any]]) -> list[Message]:
+    """Restore chat messages from checkpoint-safe payloads."""
+    restored: list[Message] = []
+    for item in payload:
+        if not isinstance(item, dict):
+            continue
+
+        role_value = decode_checkpoint_value(item.get("role"))
+        if isinstance(role_value, str):
+            role = role_value
+        elif isinstance(role_value, dict) and "value" in role_value:
+            # Handle legacy serialization format
+            role = role_value["value"]
+        else:
+            role = "assistant"
+
+        contents_field = item.get("contents", [])
+        contents: list[Any] = []
+        if isinstance(contents_field, list):
+            contents_iter: list[Any] = contents_field  # type: ignore[assignment]
+            for entry in contents_iter:
+                decoded_entry: Any = decode_checkpoint_value(entry)
+                contents.append(decoded_entry)
+
+        additional_field = item.get("additional_properties", {})
+        additional: dict[str, Any] = {}
+        if isinstance(additional_field, dict):
+            additional_dict = cast(dict[str, Any], additional_field)
+            for key, value in additional_dict.items():
+                additional[key] = decode_checkpoint_value(value)
+
+        restored.append(
+            Message(  # type: ignore[call-overload]
+                role=role,
+                contents=contents,
+                author_name=item.get("author_name"),
+                message_id=item.get("message_id"),
+                additional_properties=additional,
+            )
+        )
+    return restored

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

@@ -89,6 +89,11 @@ async def test_agent_executor_checkpoint_stores_and_restores_state() -> None:
         "and the second one is after the agent execution."
     )
 
+    # Get the second checkpoint which should contain the state after processing
+    # the first message by the start executor in the sequential workflow
+    checkpoints.sort(key=lambda cp: cp.timestamp)
+    restore_checkpoint = checkpoints[1]
+
     # Verify checkpoint contains executor state with both cache and session
     assert "_executor_state" in restore_checkpoint.state
     executor_states = restore_checkpoint.state["_executor_state"]