Python: fix(workflow): discard pending state after failed supersteps

[Patel Namraa (Namraa310806)]

Motivation & Context

Fixes a state isolation bug where pending State writes from a failed or cancelled workflow superstep can leak into a later successful Workflow.run() on the same Workflow instance.

State.set() stages writes in a pending buffer, which is committed only when a superstep completes successfully. However, when a superstep fails or is cancelled, the pending writes were not discarded.

Because workflow state persists across run() calls, a later successful superstep could call State.commit() and unintentionally commit stale writes from the previous failed run.

This results in silent state corruption: state written by a failed run can become part of the committed state of a later successful run.

Description & Review Guide

• What are the major changes?

  • Discard pending state when a superstep fails with an exception.
  • Discard pending state when a superstep is cancelled.
  • Ensure concurrent edge-runner execution cannot continue staging state after the failed superstep reaches the discard boundary.
  • Add regression tests covering failed supersteps and concurrent execution.

• What is the impact of these changes?

  • Pending writes from failed or cancelled supersteps are no longer able to leak into later successful runs.
  • Committed state from previous successful runs is preserved.
  • Successful supersteps continue to use the existing State.commit() behavior.
  • No public API or checkpoint serialization behavior is changed.

• What do you want reviewers to focus on?

  • Whether the failure/cancellation boundary correctly prevents pending state from a failed superstep from reaching a later commit.
  • Whether the concurrent edge-runner handling preserves the existing execution and error-propagation semantics.

Related Issue

Fixes #7859

Contribution Checklist

• The code builds clean without any errors or warnings
• All unit tests pass, and I have added new tests where possible
• The PR follows the Contribution Guidelines
• This PR is linked to an issue and there is no other open PR for this issue (see Related Issue above).
• This is not a breaking change. If it is a breaking change, add the breaking change label (or add "[BREAKING]" to the title prefix) — a workflow keeps the label and title prefix in sync automatically.

[Copilot]

Copilot was unable to review this pull request because the user who requested the review has reached their quota limit.

[Patel Namraa (Namraa310806)]

Evan Mattson (@moonbox3) The current PR fixes the confirmed sequential state-leak case by discarding pending state on the runner's exception/cancellation paths.

While reviewing the failure boundary more deeply, I noticed a potential concurrent-sibling case: if multiple edge/executor tasks are running in the same superstep and one fails, I want to verify that another sibling cannot continue staging state after State.discard() has already run.

I can add a deterministic regression test for this case and, if the current execution model allows the leak, make the minimal additional change needed to ensure all work from the failed superstep has stopped before the pending state is discarded.

Would you like me to investigate and include this concurrent-sibling case in this PR, or would you prefer to keep #7880 focused on the currently confirmed sequential failure case?

[Shivani . (Shivani767)]

Thanks for the focused fix, Patel Namraa (@Namraa310806) — this matches the #7859 repro and uses the existing State.discard() contract the way the runner was supposed to.

What looks good

• Calling discard() on both the CancelledError path (after the iteration task has been cancelled and awaited) and the exception path (before re-raising) is the right boundary: a failed or cancelled superstep must not reach commit().
• The regression test covers the actual leak: a failed run() followed by a successful run() on the same Workflow instance, with "secret" never landing in committed state.
• Committed state from earlier successful supersteps is left alone (discard() only clears _pending), so this does not change successful-run semantics.

On the concurrent-sibling question
I would keep this PR scoped to the confirmed sequential leak.

_run_iteration() delivers through asyncio.gather(*), which waits for remaining edge-runner tasks when one fails rather than abandoning them. discard() then runs only after await iteration_task (or after the cancelled iteration task has been awaited). Siblings can still set_state() before that await returns — those writes belong to the failed superstep and should be discarded with it. They cannot keep staging after discard(), because the iteration task is already done.

A concurrent regression test would be nice later, but it is not required to land the #7859 fix.

Non-blocking nits

• On the exception path, discard() currently runs after yielding drained events. If the consumer cancels at that yield, discard() is skipped. Safer to discard first (or in a finally) and then surface the failure events.
• A cancellation-path test is not strictly needed, but it would lock in the CancelledError discard that the issue also calls out.
• The PR description mentions concurrent edge-runner handling that is not in the diff; worth trimming that bullet so reviewers do not look for a change that is not here.

LGTM for the sequential #7859 fix.

[Shivani . (Shivani767)]

Non-blocking: consider discarding before yielding the drained events (or wrapping this in finally).

If the consumer cancels while this yield is in progress, discard() below is skipped and the pending write can still leak into a later run() — the original #7859 failure mode.

except Exception:
    self._state.discard()
    if await self._ctx.has_events():
        for event in await self._ctx.drain_events():
            yield event
    raise

The cancellation path already discards after await iteration_task, which is the right order there.