Repository navigation
Conversation
RunnerImpl discarded pending state in the finally of every uncommitted superstep, but State is shared by all runs on a Workflow. When a dropped stream was finalized after a new run had started, that discard removed the new run's staged writes. The runner now records a token for the current superstep and only discards when the finishing superstep still owns it.
There was a problem hiding this comment.
🟡 Changes recommended
Ownership transfers too late to protect successor entry-checkpoint snapshots from stale cleanup.
1 open issue
What changed in this PR
Addresses #9473 by guarding shared workflow state against cleanup from a dropped stream.
Changes:
- Adds a superstep ownership token to guard pending-state discard.
- Adds regression coverage for preserving a successor run’s state writes.
| File | Description |
|---|---|
| python/packages/core/tests/workflow/test_workflow.py | Tests successor state preservation after deferred stream cleanup. |
| python/packages/core/agent_framework/_workflows/_runner.py | Restricts discard to the owning superstep. |
Review effort: Balanced
| superstep_token = object() | ||
| self._superstep_token = superstep_token |
There was a problem hiding this comment.
You are right: the successor's entry checkpoint stages executor snapshots before its first superstep takes ownership, so the dropped run's finalizer could discard them while a save hook was suspended, and the entry checkpoint then held the previous run's snapshot. The new commit has checkpoint preparation take ownership of pending writes before it calls the executors' save hooks, so a stale finalizer no longer matches and leaves them alone. The test test_workflow_partial_stream_does_not_discard_successor_entry_checkpoint_snapshots drops a stream mid-superstep, suspends the second executor's on_checkpoint_save during the next run's entry checkpoint while the stale finalizer runs, and asserts the entry checkpoint stores the new writer snapshot.
A dropped run's deferred finalizer could still own the pending-writes token while a successor run built its entry checkpoint, so it discarded executor snapshots staged between awaited on_checkpoint_save hooks and the checkpoint stored a stale snapshot. Checkpoint preparation now takes ownership of pending writes before calling the hooks. A checkpoint-enabled regression test suspends a save hook while the stale finalizer runs.
@microsoft-github-policy-service agree |
Somay kousis (Somay-kousis)
left a comment
There was a problem hiding this comment.
The token fix works for what it targets: run B's writes are no longer thrown away by A's late finally.
What I ran:
- Your two new tests against main's
_runner.py: both fail (assert None == 'from-run-B',assert {'saves': 1} == {'saves': 2}). On your head both pass.
One thing in the same race that the PR doesn't cover, worth considering. The description says a failed or cancelled superstep still discards its own pending writes. In the scenario from your first test, A's write is never discarded. It gets committed when B starts, because _run_core calls self._runner.state.commit() right after setting the run kwargs (_workflow.py:636), and A's uncommitted a_key is still pending at that point. B's executor also reads it during its superstep.
Probe: your first test plus a read of a_key from B's handler and a dump of State at the end:
== your head
B saw a_key during its superstep: from-run-A
committed after B: b_key = from-run-B | a_key = from-run-A
== main
B saw a_key during its superstep: from-run-A
committed after B: b_key = None | a_key = from-run-A
So it's not new, main does it too. But after this PR the dropped run's half-finished superstep ends up committed next to B's writes, which is the opposite of what the description says.
I tried a small change on your branch: discard pending writes at the start of _run_core, before the run kwargs are set and committed:
# Pending writes here can only come from a superstep that never committed
# (e.g. a dropped stream whose finalizer has not run yet); do not commit them below.
self._runner.state.discard()Results with it: the probe prints B saw a_key ...: None and committed after B: b_key = from-run-B | a_key = None. I'm not sure there is no legitimate case of pending writes at run start that I'm missing, so treat this as a suggestion. If you take it, an assert workflow._runner.state.get("a_key") is None at the end of your first test would cover it.
A new run discards any pending state writes before it commits its run kwargs. Those writes can only come from a superstep or checkpoint save that never committed, such as a dropped stream whose finalizer has not run yet, so committing them put a half-finished superstep into the successor run's state.
|
Good catch, thanks. A dropped run's half-finished superstep is now thrown away when the next run starts, before that run |

Motivation & Context
closes #9473
RunnerImpl.run_until_convergencediscards pending state in thefinallyof every superstep that did not commit, but theStateobject is shared by every run on aWorkflow. If a stream is dropped mid superstep and a new run starts before the old generator is finalized, the latediscard()removes the new run's staged writes. The new run still endsIDLEwith its outputs, so the write is lost silently.Description & Review Guide
finallyonly callsdiscard()when the finishing superstep still owns that token. Cancelling the iteration and event tasks is unchanged. This mirrors howowns_runinWorkflow._run_corekeeps a stale finalizer from clearing a successor's run lock and runtime storage.I added
test_workflow_partial_stream_does_not_discard_successor_state_writesnext totest_workflow_partial_stream_does_not_clobber_successor_active_run. It drops stream A while its executor has staged a write and is blocked, starts run B, which stagesb_key, lets A's deferred finalizer run, then releases B. It checks that B finishesIDLEwith its output and thatb_keyis committed. Without the change the last assertion fails withassert None == 'from-run-B'.Related Issue
See the link under Motivation & Context.
Contribution Checklist
breaking changelabel (or add "[BREAKING]" to the title prefix, before or after any language prefix).