Python: FileCheckpointStorage saves state it cannot restore: silent save/load asymmetry, get_latest invisibility, listing disagreement, and undocumented JSONDecodeError

Description

FileCheckpointStorage accepts at save time state that it refuses at load time, and the refusal is silent end to end. Four related behaviors, all verified on main @ 8e803b9 with the standalone repro below:

Save does not validate types; load does. save() encodes any picklable object with no whitelist check; the restricted-pickle whitelist is enforced only in load() via decode_checkpoint_value. A workflow state containing an application class (not registered via allowed_checkpoint_types / register_checkpoint_type) saves successfully, then fails to restore with WorkflowCheckpointException. The failure surfaces at restore time, potentially in a different process, days later. InMemoryCheckpointStorage round-trips the same state without error, so identical workflow code works under the default test storage and breaks against the durable one.
Blocked checkpoints become invisible to get_latest. list_checkpoints skips files that fail decode with only a log warning (pinned as graceful in test_file_checkpoint_storage_corrupted_file), and get_latest builds on it. A workflow whose saved checkpoint contains an unregistered type gets get_latest() is None, which per the CheckpointStorage protocol means "no checkpoints exist" — except one does exist (see 3). A resume flow driven by get_latest sees nothing and, combined with 1, nothing was ever loud.
The two listing APIs disagree. list_checkpoint_ids reads raw JSON without decoding, so it still reports the blocked checkpoint (1 id) while list_checkpoints and get_latest report none. Two APIs answering the same question differently is a contract split of its own.
Corrupt files make load() raise json.JSONDecodeError, not the documented WorkflowCheckpointException. The CheckpointStorage.load docstring says it raises WorkflowCheckpointException "if checkpoint decoding fails", but the JSON read itself is unwrapped, so a truncated or corrupt checkpoint file surfaces as a raw json.JSONDecodeError. The pinned test only covers list_checkpoints graceful handling, not load's exception type.

I am not sure how much of 1-3 is intended design (register-the-type is the documented remedy, and fail-open listing is pinned), so the questions I'd pose: should save() validate against the whitelist (or at least warn) so the failure is loud at capture time; should get_latest distinguish "no checkpoints" from "checkpoints exist but were skipped"; should list_checkpoint_ids and list_checkpoints agree; and should load wrap corrupt-file errors in the documented exception? Item 4 looks like a plain bug regardless of the others.

Found via a differential probe of the two storages; AI assistance disclosed; repro is standalone and offline.

Reproduction Steps

import asyncio, sys, tempfile
from dataclasses import dataclass, field

sys.path.insert(0, "python/packages/core")
from agent_framework._workflows._checkpoint import (
    FileCheckpointStorage, InMemoryCheckpointStorage, WorkflowCheckpoint,
)

@dataclass
class ProbeState:  # any application class not registered for checkpoint decode
    counter: int = 0

cp = WorkflowCheckpoint(workflow_name="wf", graph_signature_hash="s", state={"e": ProbeState(7)})

async def main():
    mem = InMemoryCheckpointStorage()
    await mem.save(cp)
    print(await mem.load(cp.checkpoint_id))          # loads fine, ProbeState(7)

    with tempfile.TemporaryDirectory() as d:
        fcs = FileCheckpointStorage(d)
        await fcs.save(cp)                            # succeeds, no warning
        try:
            await fcs.load(cp.checkpoint_id)          # WorkflowCheckpointException
        except Exception as e:
            print(type(e).__name__, str(e)[:90])
        print(await fcs.list_checkpoints(workflow_name="wf"))   # []
        print(await fcs.get_latest(workflow_name="wf"))          # None
        print(await fcs.list_checkpoint_ids(workflow_name="wf")) # [id] -- disagrees

    # item 4: corrupt file
    with tempfile.TemporaryDirectory() as d:
        fcs = FileCheckpointStorage(d)
        cp2 = WorkflowCheckpoint(workflow_name="wf", graph_signature_hash="s", state={"x": 1})
        await fcs.save(cp2)
        p = __import__("pathlib").Path(d) / f"{cp2.checkpoint_id}.json"
        raw = p.read_text(); p.write_text(raw[: len(raw) // 2])
        try:
            await fcs.load(cp2.checkpoint_id)
        except Exception as e:
            print(type(e).__name__)                   # json.JSONDecodeError

asyncio.run(main())

Environment

agent-framework-core @ main 8e803b9 (editable, python/packages/core)
Python 3.12, macOS, offline repro (no external services)

one more datapoint that makes point 1 more concrete: the hitl case. a graph workflow that suspends on ctx.request_info() pickles the pending request payload into pending_request_info_events, and that payload is an application type by nature (the repo's own tests use UserApprovalRequest-style dataclasses, see tests/workflow/test_request_info_and_response.py). so a normal hitl workflow checkpointed to FileCheckpointStorage saves fine, then cannot restore without registering the payload type, discovered only at restore time. with the type registered everything round-trips correctly, and for the record the unknown-response-id path is well designed: responses={'wrong': ...} raises a loud ValueError on both storages.

Oleg Solozobov (@dev404ai)

Verified all four items on main, and they hold up. Two notes on scope before anyone starts, since parts of this are already spoken for.

Item 2 is already #7831 — list_checkpoints swallows a decode failure with logger.warning (_checkpoint.py:400) and get_latest is built on it (:425), so the newest checkpoint becomes invisible rather than an error. Oleg Solozobov (@dev404ai)'s #7832 was closed pending review comments and they have asked about reopening it, so that one has an owner.

Item 1 — encode_checkpoint_value takes no allowlist while decode_checkpoint_value does, so save cannot reject what load will refuse. #7789 is adjacent (built-in orchestration types rejected at restore, Atharva Vichare (@atty57)'s #7791 also closed), and this looks like a contract decision for the core team rather than a bug fix: either save gains the allowlist and starts rejecting today's writable payloads, or the asymmetry is documented as intended.

Item 3 — list_checkpoint_ids reads raw JSON (:456, no decode) so it still reports ids that list_checkpoints dropped. Same design call as item 1: make the two agree, and decide which direction.

Item 4 is the clean one, and that is the piece I would like to take. _read() does a bare json.load(f) (:359-363) inside asyncio.to_thread, and only the decode step is wrapped — by an except WorkflowCheckpointException: raise that is a no-op. So a truncated or non-JSON file escapes load() as json.JSONDecodeError, which is not in the documented contract; FileCheckpointStorage.load's docstring promises WorkflowCheckpointException. It is a small, self-contained change plus a test, and it does not touch anything #7831 or #7789 would.

One correction to the report: the "raises WorkflowCheckpointException if decoding fails" wording is on FileCheckpointStorage.load, not on the CheckpointStorage protocol — the protocol only documents the missing-id case.

May I take item 4 on its own and leave 1-3 for the core team to settle? Happy to fold in whatever direction you pick on the others afterwards.

🤖 Automated triage reproduction notes (agent-authored — trust but verify)

Agent analysis

Repro: python/packages/core/agent_framework/_workflows/_checkpoint.py::FileCheckpointStorage.load (lines 341-373) leaks JSONDecodeError when a checkpoint file contains malformed JSON. FileCheckpointStorage also saves unregistered application types that load rejects, after which list_checkpoints/get_latest hide the checkpoint while list_checkpoint_ids reports it. Minimal repro: save a checkpoint containing an unregistered dataclass and query all three APIs, then write malformed JSON to a checkpoint file and call load().

Failing test: python/packages/core/tests/workflow/test_checkpoint.py
Files examined: python/AGENTS.md, python/packages/core/AGENTS.md, python/packages/core/agent_framework/_workflows/_checkpoint.py, python/packages/core/agent_framework/_workflows/_checkpoint_encoding.py, python/packages/core/tests/workflow/test_checkpoint.py, python/packages/core/pyproject.toml
Tests run: test_file_checkpoint_storage_unregistered_type_is_saved_but_hidden, test_file_checkpoint_storage_load_corupted_file_raises_checkpoint_exception, existing focused FileCheckpointStorage tests
Current version: 1.17.0