From e12a480f364245c7ddde12b6f31ca9180c395c72 Mon Sep 17 00:00:00 2001 From: minelhi <3417378192@qq.com> Date: Wed, 9 Sep 2026 21:58:35 +0800 Subject: [PATCH 1/5] Python: FileCheckpointStorage save/load symmetry (#8181) --- .../agent_framework/_workflows/_checkpoint.py | 37 ++++++++++++++++--- .../core/tests/workflow/test_checkpoint.py | 37 +++++++++++++++++++ .../test_checkpoint_unrestricted_pickle.py | 17 +++++++++ 3 files changed, 86 insertions(+), 5 deletions(-) diff --git a/python/packages/core/agent_framework/_workflows/_checkpoint.py b/python/packages/core/agent_framework/_workflows/_checkpoint.py index d9b30d83919..7c113c36767 100644 --- a/python/packages/core/agent_framework/_workflows/_checkpoint.py +++ b/python/packages/core/agent_framework/_workflows/_checkpoint.py @@ -320,12 +320,26 @@ async def save(self, checkpoint: WorkflowCheckpoint) -> CheckpointID: Returns: The unique ID of the saved checkpoint. + + Raises: + WorkflowCheckpointException: If the checkpoint cannot be encoded or would + fail to decode under this storage's ``allowed_checkpoint_types``. """ - from ._checkpoint_encoding import encode_checkpoint_value + from ._checkpoint_encoding import decode_checkpoint_value, encode_checkpoint_value file_path = self._validate_file_path(checkpoint.checkpoint_id) checkpoint_dict = checkpoint.to_dict() encoded_checkpoint = encode_checkpoint_value(checkpoint_dict) + # Fail at save time if this storage could not restore the payload (#8181). + try: + decode_checkpoint_value(encoded_checkpoint, allowed_types=self._allowed_types) + except WorkflowCheckpointException: + raise + except Exception as ex: + raise WorkflowCheckpointException( + f"Checkpoint {checkpoint.checkpoint_id} cannot be restored under this " + "storage's allowed types; refusing to save." + ) from ex def _write_atomic() -> None: tmp_path = file_path.with_suffix(".json.tmp") @@ -360,7 +374,12 @@ def _read() -> dict[str, Any]: with open(file_path) as f: return json.load(f) - encoded_checkpoint = await asyncio.to_thread(_read) + try: + encoded_checkpoint = await asyncio.to_thread(_read) + except json.JSONDecodeError as ex: + raise WorkflowCheckpointException( + f"Checkpoint file for {checkpoint_id} is not valid JSON and cannot be loaded." + ) from ex from ._checkpoint_encoding import decode_checkpoint_value @@ -446,6 +465,8 @@ async def list_checkpoint_ids(self, *, workflow_name: str) -> list[CheckpointID] Returns: A list of checkpoint IDs for the specified workflow name. + Only includes checkpoints that can be decoded under this storage's + allowed types (aligned with :meth:`list_checkpoints`, #8181). """ def _list_ids() -> list[CheckpointID]: @@ -453,9 +474,15 @@ def _list_ids() -> list[CheckpointID]: for file_path in self.storage_path.glob("*.json"): try: with open(file_path) as f: - data = json.load(f) - if data.get("workflow_name") == workflow_name: - checkpoint_ids.append(data.get("checkpoint_id", file_path.stem)) + encoded_checkpoint = json.load(f) + from ._checkpoint_encoding import decode_checkpoint_value + + decoded_checkpoint_dict = decode_checkpoint_value( + encoded_checkpoint, allowed_types=self._allowed_types + ) + checkpoint = WorkflowCheckpoint.from_dict(decoded_checkpoint_dict) + if checkpoint.workflow_name == workflow_name: + checkpoint_ids.append(checkpoint.checkpoint_id) except Exception as e: logger.warning(f"Failed to read checkpoint file {file_path}: {e}") return checkpoint_ids diff --git a/python/packages/core/tests/workflow/test_checkpoint.py b/python/packages/core/tests/workflow/test_checkpoint.py index be8de7c13b6..1fe883abd3c 100644 --- a/python/packages/core/tests/workflow/test_checkpoint.py +++ b/python/packages/core/tests/workflow/test_checkpoint.py @@ -1289,6 +1289,43 @@ async def test_file_checkpoint_storage_corrupted_file(): assert checkpoints == [] +async def test_file_checkpoint_storage_load_invalid_json_raises(): + """Issue #8181: load wraps JSONDecodeError as WorkflowCheckpointException.""" + with tempfile.TemporaryDirectory() as temp_dir: + storage = FileCheckpointStorage(temp_dir) + bad_id = "bad-json-checkpoint" + bad_file = Path(temp_dir) / f"{bad_id}.json" + with open(bad_file, "w") as f: # noqa: ASYNC230 + f.write("{ not json") + with pytest.raises(WorkflowCheckpointException, match="not valid JSON"): + await storage.load(bad_id) + + +async def test_file_checkpoint_storage_list_ids_matches_list_decode_filter(): + """Issue #8181: list_checkpoint_ids skips undecodable files like list_checkpoints.""" + from tests.workflow.test_checkpoint_unrestricted_pickle import _AllowedTestState + + with tempfile.TemporaryDirectory() as temp_dir: + type_key = f"{_AllowedTestState.__module__}:{_AllowedTestState.__qualname__}" + writer = FileCheckpointStorage(temp_dir, allowed_checkpoint_types=[type_key]) + good = WorkflowCheckpoint(workflow_name="wf-a", graph_signature_hash="h") + await writer.save(good) + blocked = WorkflowCheckpoint( + workflow_name="wf-a", + graph_signature_hash="h", + checkpoint_id="orphan-blocked", + state={"x": _AllowedTestState(name="x", value=1)}, + ) + await writer.save(blocked) + + reader = FileCheckpointStorage(temp_dir) # no allow list + listed = await reader.list_checkpoints(workflow_name="wf-a") + ids = await reader.list_checkpoint_ids(workflow_name="wf-a") + assert [c.checkpoint_id for c in listed] == ids + assert "orphan-blocked" not in ids + assert good.checkpoint_id in ids + + async def test_file_checkpoint_storage_json_serialization(): with tempfile.TemporaryDirectory() as temp_dir: storage = FileCheckpointStorage(temp_dir) diff --git a/python/packages/core/tests/workflow/test_checkpoint_unrestricted_pickle.py b/python/packages/core/tests/workflow/test_checkpoint_unrestricted_pickle.py index 4e89dd1b2cc..02eb0601c5d 100644 --- a/python/packages/core/tests/workflow/test_checkpoint_unrestricted_pickle.py +++ b/python/packages/core/tests/workflow/test_checkpoint_unrestricted_pickle.py @@ -18,6 +18,7 @@ import tempfile from dataclasses import dataclass from datetime import datetime, timezone +from pathlib import Path from typing import Any import pytest @@ -287,6 +288,22 @@ async def test_file_storage_blocks_unlisted_user_type(): await load_storage.load(checkpoint.checkpoint_id) +async def test_file_storage_rejects_unlisted_user_type_at_save(): + """Issue #8181: same storage must refuse to save what it cannot restore.""" + from agent_framework import WorkflowCheckpoint + + with tempfile.TemporaryDirectory() as tmpdir: + storage = FileCheckpointStorage(tmpdir) + checkpoint = WorkflowCheckpoint( + workflow_name="test", + graph_signature_hash="hash", + state={"data": _AllowedTestState(name="test", value=1)}, + ) + with pytest.raises(WorkflowCheckpointException, match="deserialization blocked|Unable to save|cannot be restored"): + await storage.save(checkpoint) + assert list(Path(tmpdir).glob("*.json")) == [] + + async def test_file_storage_allows_listed_user_type(): """FileCheckpointStorage allows user types listed in allowed_checkpoint_types.""" from agent_framework import WorkflowCheckpoint From 930d6f9358151ba909cbc491a126119a783884ce Mon Sep 17 00:00:00 2001 From: minelhi <3417378192@qq.com> Date: Thu, 10 Sep 2026 09:56:27 +0800 Subject: [PATCH 2/5] Python: harden FileCheckpointStorage exception contract (#8181) Include encoding in save-time validation, wrap invalid UTF-8 on load, and delegate list_checkpoint_ids to list_checkpoints. --- .../agent_framework/_workflows/_checkpoint.py | 34 +++++-------------- .../core/tests/workflow/test_checkpoint.py | 11 ++++++ .../test_checkpoint_unrestricted_pickle.py | 2 +- 3 files changed, 20 insertions(+), 27 deletions(-) diff --git a/python/packages/core/agent_framework/_workflows/_checkpoint.py b/python/packages/core/agent_framework/_workflows/_checkpoint.py index 7c113c36767..69914377766 100644 --- a/python/packages/core/agent_framework/_workflows/_checkpoint.py +++ b/python/packages/core/agent_framework/_workflows/_checkpoint.py @@ -329,16 +329,16 @@ async def save(self, checkpoint: WorkflowCheckpoint) -> CheckpointID: file_path = self._validate_file_path(checkpoint.checkpoint_id) checkpoint_dict = checkpoint.to_dict() - encoded_checkpoint = encode_checkpoint_value(checkpoint_dict) - # Fail at save time if this storage could not restore the payload (#8181). + # Fail at save time if encoding or restore validation fails (#8181). try: + encoded_checkpoint = encode_checkpoint_value(checkpoint_dict) decode_checkpoint_value(encoded_checkpoint, allowed_types=self._allowed_types) except WorkflowCheckpointException: raise except Exception as ex: raise WorkflowCheckpointException( - f"Checkpoint {checkpoint.checkpoint_id} cannot be restored under this " - "storage's allowed types; refusing to save." + f"Checkpoint {checkpoint.checkpoint_id} cannot be encoded or restored under " + "this storage's allowed types; refusing to save." ) from ex def _write_atomic() -> None: @@ -371,12 +371,12 @@ async def load(self, checkpoint_id: CheckpointID) -> WorkflowCheckpoint: raise WorkflowCheckpointException(f"No checkpoint found with ID {checkpoint_id}") def _read() -> dict[str, Any]: - with open(file_path) as f: + with open(file_path, encoding="utf-8") as f: return json.load(f) try: encoded_checkpoint = await asyncio.to_thread(_read) - except json.JSONDecodeError as ex: + except (json.JSONDecodeError, UnicodeDecodeError) as ex: raise WorkflowCheckpointException( f"Checkpoint file for {checkpoint_id} is not valid JSON and cannot be loaded." ) from ex @@ -468,23 +468,5 @@ async def list_checkpoint_ids(self, *, workflow_name: str) -> list[CheckpointID] Only includes checkpoints that can be decoded under this storage's allowed types (aligned with :meth:`list_checkpoints`, #8181). """ - - def _list_ids() -> list[CheckpointID]: - checkpoint_ids: list[CheckpointID] = [] - for file_path in self.storage_path.glob("*.json"): - try: - with open(file_path) as f: - encoded_checkpoint = json.load(f) - from ._checkpoint_encoding import decode_checkpoint_value - - decoded_checkpoint_dict = decode_checkpoint_value( - encoded_checkpoint, allowed_types=self._allowed_types - ) - checkpoint = WorkflowCheckpoint.from_dict(decoded_checkpoint_dict) - if checkpoint.workflow_name == workflow_name: - checkpoint_ids.append(checkpoint.checkpoint_id) - except Exception as e: - logger.warning(f"Failed to read checkpoint file {file_path}: {e}") - return checkpoint_ids - - return await asyncio.to_thread(_list_ids) + checkpoints = await self.list_checkpoints(workflow_name=workflow_name) + return [checkpoint.checkpoint_id for checkpoint in checkpoints] diff --git a/python/packages/core/tests/workflow/test_checkpoint.py b/python/packages/core/tests/workflow/test_checkpoint.py index 1fe883abd3c..ed0dde0ce9b 100644 --- a/python/packages/core/tests/workflow/test_checkpoint.py +++ b/python/packages/core/tests/workflow/test_checkpoint.py @@ -1301,6 +1301,17 @@ async def test_file_checkpoint_storage_load_invalid_json_raises(): await storage.load(bad_id) +async def test_file_checkpoint_storage_load_invalid_utf8_raises(): + """Issue #8181: load wraps UnicodeDecodeError as WorkflowCheckpointException.""" + with tempfile.TemporaryDirectory() as temp_dir: + storage = FileCheckpointStorage(temp_dir) + bad_id = "bad-utf8-checkpoint" + bad_file = Path(temp_dir) / f"{bad_id}.json" + bad_file.write_bytes(b'{"x": "\xff\xfe"}') + with pytest.raises(WorkflowCheckpointException, match="not valid JSON"): + await storage.load(bad_id) + + async def test_file_checkpoint_storage_list_ids_matches_list_decode_filter(): """Issue #8181: list_checkpoint_ids skips undecodable files like list_checkpoints.""" from tests.workflow.test_checkpoint_unrestricted_pickle import _AllowedTestState diff --git a/python/packages/core/tests/workflow/test_checkpoint_unrestricted_pickle.py b/python/packages/core/tests/workflow/test_checkpoint_unrestricted_pickle.py index 02eb0601c5d..c6241a1a522 100644 --- a/python/packages/core/tests/workflow/test_checkpoint_unrestricted_pickle.py +++ b/python/packages/core/tests/workflow/test_checkpoint_unrestricted_pickle.py @@ -299,7 +299,7 @@ async def test_file_storage_rejects_unlisted_user_type_at_save(): graph_signature_hash="hash", state={"data": _AllowedTestState(name="test", value=1)}, ) - with pytest.raises(WorkflowCheckpointException, match="deserialization blocked|Unable to save|cannot be restored"): + with pytest.raises(WorkflowCheckpointException, match="deserialization blocked|Unable to save|cannot be restored|cannot be encoded"): await storage.save(checkpoint) assert list(Path(tmpdir).glob("*.json")) == [] From b443bbfbcc4cb958fc08e156163de3e2930b1e10 Mon Sep 17 00:00:00 2001 From: lsmlhi_25 Date: Thu, 10 Sep 2026 13:59:42 +0800 Subject: [PATCH 3/5] Python: write FileCheckpointStorage JSON as UTF-8 (#8181) Match load()'s explicit UTF-8 encoding so save/load stay symmetric on non-UTF-8 default locales. --- .../packages/core/agent_framework/_workflows/_checkpoint.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/python/packages/core/agent_framework/_workflows/_checkpoint.py b/python/packages/core/agent_framework/_workflows/_checkpoint.py index 69914377766..d0c6e2093bc 100644 --- a/python/packages/core/agent_framework/_workflows/_checkpoint.py +++ b/python/packages/core/agent_framework/_workflows/_checkpoint.py @@ -343,7 +343,7 @@ async def save(self, checkpoint: WorkflowCheckpoint) -> CheckpointID: def _write_atomic() -> None: tmp_path = file_path.with_suffix(".json.tmp") - with open(tmp_path, "w") as f: + with open(tmp_path, "w", encoding="utf-8") as f: json.dump(encoded_checkpoint, f, indent=2, ensure_ascii=False) os.replace(tmp_path, file_path) @@ -405,7 +405,7 @@ def _list_checkpoints() -> list[WorkflowCheckpoint]: checkpoints: list[WorkflowCheckpoint] = [] for file_path in self.storage_path.glob("*.json"): try: - with open(file_path) as f: + with open(file_path, encoding="utf-8") as f: encoded_checkpoint = json.load(f) from ._checkpoint_encoding import decode_checkpoint_value From 3dc4f3022b553ae9cdd59edda8a7860cbab38d2d Mon Sep 17 00:00:00 2001 From: minelhi <3417378192@qq.com> Date: Thu, 10 Sep 2026 22:16:21 +0800 Subject: [PATCH 4/5] Python: clarify UTF-8 vs JSON load errors for FileCheckpointStorage (#8181) Copilot review: UnicodeDecodeError should not claim invalid JSON. --- .../packages/core/agent_framework/_workflows/_checkpoint.py | 6 +++++- python/packages/core/tests/workflow/test_checkpoint.py | 2 +- 2 files changed, 6 insertions(+), 2 deletions(-) diff --git a/python/packages/core/agent_framework/_workflows/_checkpoint.py b/python/packages/core/agent_framework/_workflows/_checkpoint.py index d0c6e2093bc..c6a6a9dc891 100644 --- a/python/packages/core/agent_framework/_workflows/_checkpoint.py +++ b/python/packages/core/agent_framework/_workflows/_checkpoint.py @@ -376,7 +376,11 @@ def _read() -> dict[str, Any]: try: encoded_checkpoint = await asyncio.to_thread(_read) - except (json.JSONDecodeError, UnicodeDecodeError) as ex: + except UnicodeDecodeError as ex: + raise WorkflowCheckpointException( + f"Checkpoint file for {checkpoint_id} is not valid UTF-8 and cannot be loaded." + ) from ex + except json.JSONDecodeError as ex: raise WorkflowCheckpointException( f"Checkpoint file for {checkpoint_id} is not valid JSON and cannot be loaded." ) from ex diff --git a/python/packages/core/tests/workflow/test_checkpoint.py b/python/packages/core/tests/workflow/test_checkpoint.py index ed0dde0ce9b..5c06b617c89 100644 --- a/python/packages/core/tests/workflow/test_checkpoint.py +++ b/python/packages/core/tests/workflow/test_checkpoint.py @@ -1308,7 +1308,7 @@ async def test_file_checkpoint_storage_load_invalid_utf8_raises(): bad_id = "bad-utf8-checkpoint" bad_file = Path(temp_dir) / f"{bad_id}.json" bad_file.write_bytes(b'{"x": "\xff\xfe"}') - with pytest.raises(WorkflowCheckpointException, match="not valid JSON"): + with pytest.raises(WorkflowCheckpointException, match="not valid UTF-8"): await storage.load(bad_id) From 1034538d315e971a81e2bd353eb72edece84de8e Mon Sep 17 00:00:00 2001 From: minelhi <3417378192@qq.com> Date: Thu, 10 Sep 2026 23:02:45 +0800 Subject: [PATCH 5/5] Python: avoid blocking Path.glob in async checkpoint test (#8181) Package Checks ruff rule blocking-path-method-in-async-function failed on Path(tmpdir).glob in an async test. Use asyncio.to_thread like other core tests. --- .../core/tests/workflow/test_checkpoint_unrestricted_pickle.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/python/packages/core/tests/workflow/test_checkpoint_unrestricted_pickle.py b/python/packages/core/tests/workflow/test_checkpoint_unrestricted_pickle.py index c6241a1a522..1f78f4e6ad4 100644 --- a/python/packages/core/tests/workflow/test_checkpoint_unrestricted_pickle.py +++ b/python/packages/core/tests/workflow/test_checkpoint_unrestricted_pickle.py @@ -11,6 +11,7 @@ - Built-in safe types and framework types are always allowed """ +import asyncio import base64 import enum import os @@ -301,7 +302,7 @@ async def test_file_storage_rejects_unlisted_user_type_at_save(): ) with pytest.raises(WorkflowCheckpointException, match="deserialization blocked|Unable to save|cannot be restored|cannot be encoded"): await storage.save(checkpoint) - assert list(Path(tmpdir).glob("*.json")) == [] + assert not await asyncio.to_thread(lambda: list(Path(tmpdir).glob("*.json"))) async def test_file_storage_allows_listed_user_type():