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 |
| self._runner.reset_iteration_count() | ||
|
|
||
| # Pending writes here come from a superstep or checkpoint save that never committed; do not commit them. | ||
| self._runner.state.discard() |
There was a problem hiding this comment.
This discard only removes predecessor writes already pending when B reaches this point. A's finalizer can run later: its cancelled handler may call ctx.set_state() from finally after this discard, and once B has replaced _superstep_token, A's cleanup skips its own discard. B's next superstep commit can then persist A's cleanup write. Please drain or track predecessor finalization, or isolate pending state by run/epoch, and add a regression where A writes state during cancellation after B has started.
There was a problem hiding this comment.
Thanks, good catch. Fixed in e92c3f9: pending writes are now scoped per superstep, so the dropped run's cleanup write is ignored. New test: test_workflow_partial_stream_cancellation_write_not_committed_by_successor.
There was a problem hiding this comment.
Thanks, good catch. Fixed in e92c3f9: pending writes are now scoped per superstep, so the dropped run's cleanup write is ignored. Regression: test_workflow_partial_stream_cancellation_write_not_committed_by_successor.
Pending state writes are now tied to the superstep that owns the shared State, and writes from executor tasks of a superstep that no longer owns it are ignored. A dropped run's executor that writes state during cancellation after a newer run has started could otherwise have its write committed by the newer run.
|
Sorry for the late reply. Reran my leak probe on e92c3f9: the dropped run's write no longer shows up, |
|
|
||
| def clear(self) -> None: | ||
| """Clear both committed and pending state.""" | ||
| if self._is_stale_write(): |
There was a problem hiding this comment.
digit50 Stale contexts can still mutate successor state through other State methods — WorkflowContext.state exposes the shared State object, but epoch checks are only applied to set, delete, and clear. A late cancellation handler from a dropped run can call ctx.state.discard() after the successor has staged writes, clearing those writes; commit() can publish them before the superstep boundary, and import_state() can directly alter committed state. These mutations bypass the ownership guard and let a stale run affect its successor.
There was a problem hiding this comment.
Now fixed in 9af013f: commit, discard and import_state get the same stale-epoch check as set, delete and clear, and so does begin_epoch, which a stale handler could otherwise call to make the successor's own writes look stale. Framework callers of these methods all run outside any superstep's bound context, so they are unaffected. Regression: test_workflow_partial_stream_cancellation_state_operation_ignored_for_successor, parametrized over the four methods.
…s it commit, discard, import_state and begin_epoch now ignore calls from a superstep that no longer owns the State, as set, delete and clear already did. A dropped run's cancelled handler could otherwise reach them through ctx.state to clear, publish or overwrite a successor run's state, or make the successor's own writes look stale.

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
Fixes #9473
Contribution Checklist
breaking changelabel (or add "[BREAKING]" to the title prefix, before or after any language prefix).