Python: [Bug]: AgentLoopMiddleware fires turn-scoped after_run on a failed/early-closed streaming iteration using a stale prior response #9817

Observed Behavior

When a successful AgentLoopMiddleware streaming iteration is followed by an exception or early close, the loop still invokes turn-scoped after_run providers (after_run_once_per_turn = True). The callback's SessionContext contains the final response from the previous iteration.

The streaming loop invokes the deferred providers unconditionally in its generator's finally block. The non-streaming path invokes them after try/finally, so a propagating exception skips the callback.

Source baseline:

• Streaming loop
• Non-streaming loop

Expected Behavior

Streaming and non-streaming loop runs should agree on whether a failed turn invokes deferred providers. Normal agent run finalization also skips after_run when an exception propagates.

The deferred callback should run after an intentional stop (should_continue, max_iterations, pending approval or handled MiddlewareTermination), and should be skipped when an iteration fails or the stream is abandoned.

Impact

Turn-scoped providers receive a stale response and can execute side effects after an incomplete streaming loop. With a configured after_strategy, CompactionProvider can run stored-history compaction through this unexpected callback. Its compaction strategy operates on session history rather than directly consuming the response.

The counting regressions establish the incorrect callback and streaming/non-streaming asymmetry. Provider-specific state consequences depend on the configured after_run behavior.

Steps to Reproduce

1. Register a provider with after_run_once_per_turn = True that counts after_run calls.
2. Run an agent with AgentLoopMiddleware(always_continue, max_iterations=5).
3. Use a client that succeeds on iteration 1 and raises on iteration 2.
4. Consume the streaming response and observe the expected RuntimeError.
5. Observe one deferred callback, compared with zero for the equivalent non-streaming run.

For early close, continue pulling until the second client iteration has started, then close the stream. This ensures a previous completed response exists, without relying on the number of transport updates.

Proposed Fix

Gate deferred provider finalization on a flag set only after the streaming loop ends cleanly. Convert its three intentional-stop returns to breaks. Keep inner-stream closure and iteration-token cleanup unchanged.

Draft implementation and deterministic regressions are in #9818. On current main (03341b15), substituting the unfixed streaming method produces two expected regression failures (iteration exception and early close); the non-streaming control passes. With the fix, all three pass, and 1,201 related tests pass with the current middleware termination changes.