Python: [Bug]: Pending State writes from a failed superstep leak into a later successful run #7859

Description

Description

State implements superstep-caching semantics: set() stages a write in a _pending buffer, commit() moves _pending into _committed at a superstep boundary, and discard() is documented as the way to abandon a superstep's staged writes without committing them.

However, State.discard() is never called by the workflow runner on failure or cancellation paths.

This matters because a Workflow intentionally keeps the same RunnerImpl / State instance alive across multiple run() calls. Therefore, pending writes left behind by a failed run can survive until the next successful run.

The sequence is:

1. An executor calls ctx.set_state(key, value), which stages the write in State._pending.
2. The executor then raises an exception, causing the current superstep to fail.
3. RunnerImpl.run_until_convergence() catches the exception and re-raises it, but does not discard the pending state writes. The same applies to the cancellation path.
4. The failed run therefore leaves the staged write in State._pending.
5. The same Workflow instance can be reused for another run() call.
6. During the later successful run, the runner reaches a superstep boundary and calls State.commit().
7. commit() then commits the stale pending write from the previous failed run, even though the later run never wrote that state.

Actual behavior

A state value written during a failed superstep can silently become committed state during a later successful and otherwise unrelated Workflow.run().

There is no error or warning associated with the leaked state.

Expected behavior

State writes staged during a failed or cancelled superstep should be discarded and should never be committed by a later successful run.

Root cause

State.discard() already exists and clears the pending buffer, but it is not invoked by the runner's exception/cancellation paths.

This appears to be separate from #7683. #7683 concerns state isolation across checkpoint/restore/storage boundaries, while this issue concerns pending state writes surviving a failed superstep and being committed by a later run. The two issues have different triggers and mechanisms and can be fixed independently.

I'm happy to open a PR with a regression test and a minimal fix using the existing State.discard() mechanism, if this behavior is confirmed as unintended.

Code Sample

import asyncio
from dataclasses import dataclass

from agent_framework import Executor, WorkflowBuilder, WorkflowContext, handler


@dataclass
class Msg:
    fail: bool


class FlakyExecutor(Executor):
    @handler
    async def run(self, message: Msg, ctx: WorkflowContext[Msg, str]) -> None:
        if message.fail:
            # This write is staged in State._pending.
            ctx.set_state("secret", "leaked-from-failed-run")

            # The superstep then fails before the pending state is committed.
            raise RuntimeError("simulated transient failure")

        # The successful run does not modify "secret".
        await ctx.yield_output("ok")


async def main() -> None:
    workflow = WorkflowBuilder(
        start_executor=FlakyExecutor(id="flaky")
    ).build()

    # Run 1: fails after staging a state write.
    try:
        async for _ in workflow.run(Msg(fail=True), stream=True):
            pass
    except RuntimeError:
        pass

    # Run 2: succeeds and does not write "secret".
    async for _ in workflow.run(Msg(fail=False), stream=True):
        pass

    committed = workflow._runner.state.export_state()
    print(committed)

    # Expected:
    # {'_workflow_run_kwargs': {}}
    #
    # Actual on the affected behavior:
    # {'_workflow_run_kwargs': {}, 'secret': 'leaked-from-failed-run'}


if __name__ == "__main__":
    asyncio.run(main())

Error Messages / Stack Traces

No exception is raised for the state leak itself.

Run 1 raises the expected:
RuntimeError("simulated transient failure")

Run 2 completes successfully, but the committed workflow state incorrectly
contains the "secret" value written during the failed Run 1.

This makes the issue a silent state-consistency/data-corruption bug rather
than a crash.

Package Versions

agent-framework-core: reproduced against main

Python Version

Python 3.13.15

Additional Context

Additional Context

• Reproduced against the unmodified main branch.
• After the first failed run, the state write remains pending.
• After the second successful run, the stale state is incorrectly present in committed state.
• The existing workflow behavior intentionally allows the same Workflow instance to be reused across multiple run() calls, which is what allows the stale pending state to survive between runs.
• State.discard() already exists and is intended to clear pending state without committing it.
• The proposed fix is to discard pending state when a superstep terminates through an exception or cancellation, before propagating the failure.
• This issue is related to Python: [Bug]: Checkpoint state is not isolated from live workflow state across restoration and storage boundaries #7683 because both involve workflow state consistency, but it has a different trigger and mechanism. Python: [Bug]: Checkpoint state is not isolated from live workflow state across restoration and storage boundaries #7683 concerns checkpoint/restore/storage state isolation, while this issue concerns pending writes surviving a failed superstep.
• The issue can be addressed independently of Python: [Bug]: Checkpoint state is not isolated from live workflow state across restoration and storage boundaries #7683.
• I'm happy to submit a PR with a focused regression test and minimal fix if the behavior is confirmed as a bug.