From 6fd3677c15ee5c68a6519955b739ef423269bd20 Mon Sep 17 00:00:00 2001 From: ptimizeroracle Date: Thu, 10 Sep 2026 11:01:03 +0200 Subject: [PATCH 1/2] fix: raise WorkflowCheckpointException when loading a corrupted checkpoint file (#8181) Signed-off-by: ptimizeroracle --- .../agent_framework/_workflows/_checkpoint.py | 10 ++++++++- .../core/tests/workflow/test_checkpoint.py | 21 +++++++++++++++++++ 2 files changed, 30 insertions(+), 1 deletion(-) diff --git a/python/packages/core/agent_framework/_workflows/_checkpoint.py b/python/packages/core/agent_framework/_workflows/_checkpoint.py index d9b30d83919..f17f4716f38 100644 --- a/python/packages/core/agent_framework/_workflows/_checkpoint.py +++ b/python/packages/core/agent_framework/_workflows/_checkpoint.py @@ -358,7 +358,15 @@ async def load(self, checkpoint_id: CheckpointID) -> WorkflowCheckpoint: def _read() -> dict[str, Any]: with open(file_path) as f: - return json.load(f) + try: + return json.load(f) + except json.JSONDecodeError as exc: + # `load` is documented to raise WorkflowCheckpointException + # when checkpoint decoding fails; a truncated or corrupted + # file should surface as that, not a raw json error (#8181). + raise WorkflowCheckpointException( + f"Checkpoint file for ID {checkpoint_id} is corrupted: {exc}" + ) from exc encoded_checkpoint = await asyncio.to_thread(_read) diff --git a/python/packages/core/tests/workflow/test_checkpoint.py b/python/packages/core/tests/workflow/test_checkpoint.py index be8de7c13b6..30a03f9b2fb 100644 --- a/python/packages/core/tests/workflow/test_checkpoint.py +++ b/python/packages/core/tests/workflow/test_checkpoint.py @@ -1275,6 +1275,27 @@ async def test_file_checkpoint_storage_directory_creation(): assert file_path.exists() +async def test_file_checkpoint_storage_load_corrupted_raises_checkpoint_exception(): + """`load` on a corrupted file raises the documented exception (#8181, item 4). + + Distinct from the graceful-list behavior pinned by + `test_file_checkpoint_storage_corrupted_file`: loading a truncated + checkpoint must surface as `WorkflowCheckpointException`, per the + `CheckpointStorage.load` contract, not as a raw `json.JSONDecodeError`. + """ + with tempfile.TemporaryDirectory() as temp_dir: + storage = FileCheckpointStorage(temp_dir) + checkpoint = WorkflowCheckpoint(workflow_name="wf", graph_signature_hash="sig", state={"x": 1}) + await storage.save(checkpoint) + + file_path = Path(temp_dir) / f"{checkpoint.checkpoint_id}.json" + raw = file_path.read_text() + file_path.write_text(raw[: len(raw) // 2]) # truncate mid-JSON + + with pytest.raises(WorkflowCheckpointException, match="corrupted"): + await storage.load(checkpoint.checkpoint_id) + + async def test_file_checkpoint_storage_corrupted_file(): with tempfile.TemporaryDirectory() as temp_dir: storage = FileCheckpointStorage(temp_dir) From 2db04de7c800159ee71c13889729a32215366ddf Mon Sep 17 00:00:00 2001 From: ptimizeroracle Date: Thu, 10 Sep 2026 11:29:45 +0200 Subject: [PATCH 2/2] fix: also wrap invalid utf-8 decode errors in WorkflowCheckpointException (PR review) Signed-off-by: ptimizeroracle --- .../core/agent_framework/_workflows/_checkpoint.py | 7 ++++--- .../core/tests/workflow/test_checkpoint.py | 14 ++++++++++++++ 2 files changed, 18 insertions(+), 3 deletions(-) diff --git a/python/packages/core/agent_framework/_workflows/_checkpoint.py b/python/packages/core/agent_framework/_workflows/_checkpoint.py index f17f4716f38..270e6c138da 100644 --- a/python/packages/core/agent_framework/_workflows/_checkpoint.py +++ b/python/packages/core/agent_framework/_workflows/_checkpoint.py @@ -360,10 +360,11 @@ def _read() -> dict[str, Any]: with open(file_path) as f: try: return json.load(f) - except json.JSONDecodeError as exc: + except (json.JSONDecodeError, UnicodeDecodeError) as exc: # `load` is documented to raise WorkflowCheckpointException - # when checkpoint decoding fails; a truncated or corrupted - # file should surface as that, not a raw json error (#8181). + # when checkpoint decoding fails; a truncated file or one + # with invalid utf-8 should surface as that, not a raw + # json/unicode error (#8181). raise WorkflowCheckpointException( f"Checkpoint file for ID {checkpoint_id} is corrupted: {exc}" ) from exc diff --git a/python/packages/core/tests/workflow/test_checkpoint.py b/python/packages/core/tests/workflow/test_checkpoint.py index 30a03f9b2fb..16329ef2603 100644 --- a/python/packages/core/tests/workflow/test_checkpoint.py +++ b/python/packages/core/tests/workflow/test_checkpoint.py @@ -1296,6 +1296,20 @@ async def test_file_checkpoint_storage_load_corrupted_raises_checkpoint_exceptio await storage.load(checkpoint.checkpoint_id) +async def test_file_checkpoint_storage_load_invalid_utf8_raises_checkpoint_exception(): + """`load` on a file with invalid utf-8 raises the documented exception (PR review).""" + with tempfile.TemporaryDirectory() as temp_dir: + storage = FileCheckpointStorage(temp_dir) + checkpoint = WorkflowCheckpoint(workflow_name="wf", graph_signature_hash="sig", state={"x": 1}) + await storage.save(checkpoint) + + file_path = Path(temp_dir) / f"{checkpoint.checkpoint_id}.json" + file_path.write_bytes(b'{"workflow_name": "' + b"\xff\xfe" + b'"}') # invalid utf-8 + + with pytest.raises(WorkflowCheckpointException, match="corrupted"): + await storage.load(checkpoint.checkpoint_id) + + async def test_file_checkpoint_storage_corrupted_file(): with tempfile.TemporaryDirectory() as temp_dir: storage = FileCheckpointStorage(temp_dir)