Python: [Bug]: Checkpoint state is not isolated from live workflow state across restoration and storage boundaries

[Namraa310806]

Description

WorkflowCheckpoint is intended to represent a stable snapshot of workflow state — its docstring states that checkpoints "can be shared and restored across different workflow instances of the same workflow definition." That's undermined by two related but distinct isolation gaps:

1. Restoration boundary. State.export_state() and State.import_state() — the methods used everywhere a checkpoint is built or restored — perform shallow copies:

def export_state(self) -> dict[str, Any]:
return dict(self._committed)

def import_state(self, state: dict[str, Any]) -> None:
self._committed.update(state)

RunnerContext.build_checkpoint() calls state.export_state() to populate WorkflowCheckpoint.state, and both Runner.restore_from_checkpoint() and Runner.restore_checkpoint() call self._state.import_state(checkpoint.state) to restore it. Because both are shallow, any state value that's a mutable container (list, dict — the normal way to accumulate conversation history, tool results, etc.) ends up aliased by reference between a checkpoint and whatever State it was built from or restored into. If a resumed workflow does the ordinary "read a state value, mutate it in place, write it back" pattern, it also mutates the checkpoint object it was restored from — with no error, and no requirement that any CheckpointStorage even be involved (Runner.restore_checkpoint(checkpoint) accepts a WorkflowCheckpoint directly, e.g. for sub-workflow checkpoints embedded in a parent WorkflowExecutor's state).

2. Storage boundary. Separately, InMemoryCheckpointStorage.save() deep-copies before storing (self._checkpoints[id] = copy.deepcopy(checkpoint)), but load(), list_checkpoints(), and get_latest() all return the internally stored object by reference:

save() -> self._checkpoints[id] = copy.deepcopy(checkpoint) # isolated
load() -> return checkpoint # NOT isolated
list() -> return [cp for cp in self._checkpoints.values() ...] # NOT isolated
latest() -> return latest_checkpoint # NOT isolated

So even reading a checkpoint out of the default in-memory store hands the caller a mutable alias into the store's internal state, independent of issue (1).

This appears inconsistent with the isolation already explicitly applied to messages in the same checkpoint-construction path — RunnerContext.build_checkpoint() copies that field with the comment:

Copy the per-source lists so the snapshot is isolated from later context mutations.

messages={source_id: list(messages) for source_id, messages in self._messages.items()},

state receives no equivalent treatment.

Expected behavior: reading or restoring a checkpoint should never give a caller — including a resumed workflow's own live state — a reference that can mutate the checkpoint snapshot or the storage backend's internal representation. A checkpoint should stay stable for the life of the process, consistent with it being restorable/shareable across instances.

Root cause:

• State.export_state() / import_state() are shallow (used at exactly the three checkpoint build/restore call sites in RunnerContext.build_checkpoint() and Runner.restore_from_checkpoint() / restore_checkpoint() — not used anywhere else in the codebase).
• InMemoryCheckpointStorage.load() / list_checkpoints() / get_latest() return internal objects by reference, asymmetric with save().

Possible direction — there are a couple of reasonable places to draw the isolation boundary, and I'd like input before picking one:

• Deep-copy inside State.export_state() / import_state(), which would fix both the storage-backed and direct/storage-less restore paths from a single change (mirrors the isolation already applied to messages).
• Deep-copy inside InMemoryCheckpointStorage.load() / list_checkpoints() / get_latest(), matching the existing save() behavior — this closes the storage leak but not the direct-restore path (Runner.restore_checkpoint(checkpoint) bypassing storage entirely).
• Both, for defense-in-depth.

The trade-off worth flagging: deep-copying in State runs on every checkpoint build/restore, and if a workflow's state is large this could add measurable cost — I don't have a strong opinion on whether that's acceptable versus, e.g., scoping the copy to only the values being handed across the boundary. Happy to implement whichever direction maintainers prefer, with a regression test for each of the two reproductions below, in a small follow-up PR.

Not a known regression against older releases specifically — State.export_state() / import_state() appear to have always done a shallow copy, so this is likely long-standing rather than recent.

Code Sample

# Repro 1 -- restoration boundary, no CheckpointStorage involved
from agent_framework._workflows._state import State

# Mirrors RunnerContext.build_checkpoint(): state=state.export_state()
source = State()
source.set("history", ["step-1"])
source.commit()
checkpoint_state = source.export_state()

# Mirrors Runner.restore_checkpoint(checkpoint): state.import_state(checkpoint.state)
restored = State()
restored.import_state(checkpoint_state)

# Ordinary "read, mutate, write back" pattern on the resumed state
history = restored.get("history")
history.append("step-2")
restored.set("history", history)
restored.commit()

print(checkpoint_state)
# Actual:   {'history': ['step-1', 'step-2']}
# Expected: {'history': ['step-1']}


# Repro 2 -- storage boundary
import asyncio
from datetime import datetime, timezone
from agent_framework._workflows._checkpoint import InMemoryCheckpointStorage, WorkflowCheckpoint

async def main():
    storage = InMemoryCheckpointStorage()
    checkpoint = WorkflowCheckpoint(
        workflow_name="demo", graph_signature_hash="hash-1", checkpoint_id="cp-1",
        previous_checkpoint_id=None, timestamp=datetime.now(timezone.utc).isoformat(),
        messages={}, state={"history": ["step-1"]},
        pending_request_info_events={}, iteration_count=1,
    )
    await storage.save(checkpoint)

    loaded = await storage.load("cp-1")
    loaded.state["history"].append("step-2")   # mutate what load() handed us

    reloaded = await storage.load("cp-1")
    print(reloaded.state)
    # Actual:   {'history': ['step-1', 'step-2']}
    # Expected: {'history': ['step-1']}

asyncio.run(main())

Error Messages / Stack Traces

No exception is raised — this is a silent data-corruption bug, not a crash.
Both reproductions print the actual (corrupted) vs. expected checkpoint state
via print(); there is no traceback to attach.

Package Versions

agent-framework-core==1.14.0

Python Version

3.11.5 and 3.13.15 (both verified, same result on each)

Additional Context

Repro independently verified on Windows, both Python 3.11.5 and 3.13.15:

python repro1.py (Python 3.11.5)
{'history': ['step-1', 'step-2']}

py -3.13 repro1.py (Python 3.13.15)
{'history': ['step-1', 'step-2']}

python repro2.py (Python 3.11.5)
{'history': ['step-1', 'step-2']}

py -3.13 repro2.py (Python 3.13.15)
{'history': ['step-1', 'step-2']}

All four runs match the documented "Actual" output — checkpoint state is mutated
after resume / after load(), where "Expected" is {'history': ['step-1']} in every
case. Also independently verified on Linux, Python 3.12.3, against main @ commit
12621e0 (2026-08-14) — identical result. The bug reproduces consistently across OS
and Python minor version, consistent with it being pure Python container-reference/
dict-merge logic (copy.deepcopy, dict, list) with no version- or OS-specific code path.

Both reproductions are independent of each other:

• Repro 1 demonstrates the aliasing purely through State.export_state() /
  import_state(), with zero involvement of any CheckpointStorage backend.
• Repro 2 demonstrates a second, separate leak specific to InMemoryCheckpointStorage,
  whose load() / list_checkpoints() / get_latest() are asymmetric with save()
  (which already deep-copies).

Full existing tests/workflow/ suite (81 tests) passes unmodified on main at the
commit above, confirming this isn't already covered/caught by current tests.

Happy to open a PR with regression tests for both reproductions plus a fix, once
there's agreement on which boundary (State, InMemoryCheckpointStorage, or both)
should own the isolation.

[github-actions]

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

Agent analysis

Repro: Two isolation gaps in checkpoint state handling. (1) State.export_state() at python/packages/core/agent_framework/_workflows/_state.py:111 returns dict(self._committed) which is a shallow copy — mutable values (lists, dicts) remain aliased between checkpoint snapshot and live State after import_state(). (2) InMemoryCheckpointStorage.load()/list_checkpoints()/get_latest() at _checkpoint.py:219,225,240 return stored objects by reference without deepcopy, asymetric with save() which already deepcopies. Minimal repro: export state containing a list, import into new State, mutate the list via restored.get(), observe checkpoint dict is mutated.

• Failing test: python/packages/core/tests/workflow/test_checkpoint_isolation_bug_7683.py
• Files examined: python/packages/core/agent_framework/_workflows/_state.py, python/packages/core/agent_framework/_workflows/_checkpoint.py
• Tests run: python/packages/core/tests/workflow/test_checkpoint_isolation_bug_7683.py
• Reported version: 1.14.0
• Current version: 1.14.0

[Namraa310806]

Proposed fix

I think the isolation can be addressed at two boundaries:

• Deep-copy State.export_state() / import_state() so mutable state values cannot be shared between a checkpoint snapshot and live workflow state. This covers both storage-backed and direct Runner.restore_checkpoint() restoration.
• Deep-copy the checkpoints returned by InMemoryCheckpointStorage.load(), list_checkpoints(), and get_latest() to prevent callers from mutating the storage's internal checkpoint objects, matching the existing defensive copy in save().

I believe this provides isolation for both reproductions while keeping the change localized to the existing checkpoint/state boundaries.

I'd be happy to implement this approach, including regression tests for both cases, and open a PR if the maintainers are comfortable with this direction. If you'd prefer the isolation to live at a different boundary, I'm happy to follow that approach instead.