Python: fix(core): refuse SecretString at checkpoint encode with a precise error

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

@@ -54,6 +54,7 @@
 from enum import EnumMeta
 from typing import Any, cast
 
+from .._settings import SecretString
 from ..exceptions import WorkflowCheckpointException
 
 logger = logging.getLogger("agent_framework")
@@ -370,6 +371,18 @@ def _decode(value: Any, *, allowed_types: frozenset[str] | None = None) -> Any:
 
 def _encode_pickle(value: Any) -> dict[str, str]:
     """Encode a value as a pickle envelope."""
+    # SecretString is deliberately unpicklable: protocols >= 2 fail at restore
+    # on its immutability guard, and a working pickle would write the
+    # credential to durable storage in cleartext. Refuse at encode time with a
+    # precise error instead of the opaque restore failure (#9514); the session
+    # state path already rejects it the same way (#8874).
+    if isinstance(value, SecretString):
+        raise WorkflowCheckpointException(
+            "SecretString cannot be checkpointed: it is unpicklable by design, and "
+            "persisting it would write the credential to durable storage in cleartext. "
+            "Keep credentials out of checkpointed state and re-resolve them from "
+            "settings or configuration on restore."
+        )
     return {
         _PICKLE_MARKER: _pickle_to_base64(value),
         _TYPE_MARKER: _value_type_to_key(value),

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

@@ -1170,6 +1170,20 @@ def _build_workflow() -> Any:
 
 # region FileCheckpointStorage
 
+async def test_file_checkpoint_storage_save_refuses_secret_string(tmp_path):
+    """The precise encode error surfaces unwrapped at save time (#9514)."""
+    from agent_framework._settings import SecretString
+
+    storage = FileCheckpointStorage(str(tmp_path))
+    checkpoint = WorkflowCheckpoint(
+        workflow_name="w",
+        graph_signature_hash="h",
+        state={"credential": SecretString("sk-secret")},
+    )
+    with pytest.raises(WorkflowCheckpointException, match="SecretString cannot be checkpointed"):
+        await storage.save(checkpoint)
+
+
 
 async def test_file_checkpoint_storage_save_and_load():
     with tempfile.TemporaryDirectory() as temp_dir:

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

@@ -6,6 +6,10 @@
 from datetime import datetime, timezone
 from typing import Any, cast
 
+import pytest
+
+from agent_framework._settings import SecretString
+from agent_framework.exceptions import WorkflowCheckpointException
 from agent_framework._workflows._checkpoint_encoding import (
     _PICKLE_MARKER,  # pyright: ignore[reportPrivateUsage]
     _TYPE_MARKER,  # pyright: ignore[reportPrivateUsage]
@@ -437,3 +441,19 @@ def test_decode_preserves_user_dict_matching_old_escape_shape() -> None:
     """Test that user data shaped like an old escape envelope remains unchanged."""
     data = {"__agent_framework_checkpoint_dict__": True, "value": {"safe": "data"}}
     assert decode_checkpoint_value(data) == data
+
+
+def test_encode_secret_string_rejected_precisely() -> None:
+    """A credential in checkpointed state fails at encode, not at restore (#9514)."""
+    with pytest.raises(WorkflowCheckpointException, match="SecretString cannot be checkpointed"):
+        encode_checkpoint_value(SecretString("sk-secret"))
+
+
+def test_encode_secret_string_nested_in_dict_rejected() -> None:
+    with pytest.raises(WorkflowCheckpointException, match="SecretString cannot be checkpointed"):
+        encode_checkpoint_value({"api_key": SecretString("sk-secret")})
+
+
+def test_encode_secret_string_nested_in_list_rejected() -> None:
+    with pytest.raises(WorkflowCheckpointException, match="SecretString cannot be checkpointed"):
+        encode_checkpoint_value({"keys": [SecretString("sk-secret")]})