[skrawcz]

Stopping a streaming action early (break, closing stream_iterate / astream_iterate, cancelling the consuming task) leaves the action's generator chain suspended until garbage collection (sync) or loop.shutdown_asyncgens() (async). Anything the action holds open (HTTP/SSE streams, DB cursors, OTel spans, async with blocks) is released late and in an arbitrary context. The cause is that every forwarding layer (container → runner → action wrapper → pydantic wrapper → user generator) iterates with a bare for / async for and never closes the generator it reads from.

async with contextlib.aclosing(app.astream_iterate(halt_after=["act"])) as steps:
    async for _, container in steps:
        async for _ in container:
            break
        break
# main:   the action's `finally` hasn't run (it runs at loop shutdown, with
#         "aclose(): asynchronous generator is already running")
# branch: it has run, and the step has been reported

Behavior changes

	main	this PR
Action cleanup after an early stop	deferred to GC / event-loop shutdown	runs immediately: at the break (sync), or when aclosing / aclose() / task cancellation unwinds (async)
post_run_step for the stopped step	GeneratorExit, late, in an arbitrary context	after cleanup, exactly once: GeneratorExit when closed, CancelledError when cancelled. Persisters still save failed; state is not updated
Errors while stopping	a cleanup error is only logged by the finalizer	the first error wins: a cleanup error (also what hooks receive), else a post_run_step hook error. close() / aclose() raise it, and every repeat call raises the same error; logged at warning after a bare break. Never replaces an error that was already stopping the stream
container.close() / await container.aclose()	—	new: stop without draining. Idempotent; a no-op once the stream completed
Closing a stream while it is producing an item (from a hook or another task)	—	raises RuntimeError telling you to cancel the consuming task; the stream is left alone
exception type in post_run_step, post_run_execute_call, tracking client	Exception / Optional[Exception]	Optional[BaseException] (annotation only)

Unchanged: fully consumed streams, and ASGI client disconnects. On a disconnect, Starlette/FastAPI cancel through an anyio cancel scope, which keeps cancelling every await inside it. The action's cleanup starts, but an await inside it is cancelled unless wrapped in anyio.CancelScope(shield=True). Burr's await on async post_run_step hooks and persisters is cancelled too, so the stopped step isn't recorded. Resume still re-runs it, because the last saved step is the one before. Whether Burr should shield post-step hooks is a separate question.

Release-notes worthy: cleanup errors now reach hooks, and close() / aclose() raise errors from cleanup or from post_run_step hooks.

What users do

Plain Python:

• sync: break;
• async: break inside async with contextlib.aclosing(...), or call await container.aclose(). Python never closes an async generator at a break.
• web servers: the async pattern; the server's cancellation does the rest.

The docs ("Stopping a stream early" in streaming-actions.rst) and the new examples/streaming-early-stop cover each case. The example prints a timeline for each demo and includes a FastAPI server.

Implementation

• Containers: the new close() / aclose(); iterating closes the action before reporting.
• stream_iterate / astream_iterate: closing either one closes the in-flight container.
• Runners: the streaming _run_* / _arun_* runners close the action generator in finally.
• Action wrappers: FunctionBasedStreamingAction, type_eraser and the pydantic wrappers use closing / aclosing.
• Example tooling: anyio is added to the tests extra.

Testing

• New early-stop tests: a "Stopping a streaming action early" section in tests/core/test_application.py, covering sync and async, single- and multi-step. It tests every stop path, ordering, exactly-once reporting, persistence and resume, cleanup errors, first-error-wins (cleanup vs. hook errors, the same error on every repeat close), close-while-producing, timing independence, and the anyio cancel scope (with shielded and unshielded awaited cleanup). Against main, 30 of these fail.
• Pydantic and example tests: closing a pydantic streaming action mid-stream (in test_burr_pydantic.py), and the example's timelines (in test_streaming_early_stop_example.py).
• Full suite: 772 passed. The only failures are environment-only (a langchain serde test, and langfuse not installed) and happen on main too.
• Pre-commit is green.
• Python versions: the streaming tests pass on 3.10, 3.12, 3.13 and 3.14.
• Real server: uvicorn + curl hang-up.
• Downstream: an internal consumer could drop its AsyncExitStack workaround.

Checklist

• PR has an informative and human-readable title (this will be pulled into the release notes)
• Changes are limited to a single goal (no scope creep)
• Code passed the pre-commit check & code is left cleaner/nicer than when first encountered.
• Any change in functionality is tested
• New functions are documented (with a description, list of inputs, and expected output)
• Placeholder code is flagged / future TODOs are captured in comments
• Project documentation has been updated if adding/changing functionality.

[jernejfrank]

hey, lgtm. Makes sense to me to propagate closing exceptions as well

[xihaian251]

Thanks for the implementation. I independently exercised a 35-scenario matrix covering synchronous and asynchronous early termination. I agree that the core close-propagation mechanism is sound, and I recommend merging the PR.

I found two additional edge cases worth documenting and addressing:

1. AnyIO cancellation can interrupt awaited resource cleanup (Medium–High).

I reproduced a case where an async streaming action owns a resource through async with, and its __aexit__ performs an awaited release operation. When the consumer is cancelled through an AnyIO task-group cancel scope, the observed result is:

release_started  = True
release_finished = False

The same behavior reproduces with a bare async generator and no Burr code involved. Plain asyncio.Task.cancel() completes the release, so this is specifically related to AnyIO's level-cancellation semantics.

This makes the new documentation statement that "the action is still cleaned up" too strong. The current cancellation test also does not exercise an awaited cleanup operation.

I suggest clarifying that cleanup may be interrupted when it contains awaits under an active cancellation scope, and adding a regression test that captures this behavior.

2. Repeated close/aclose calls can expose inconsistent exceptions (Medium).

I reproduced the same issue in both synchronous and asynchronous containers.

When both the action cleanup and post_run_step hook raise:

• First close/aclose raises HookError.
• Subsequent calls raise CleanupError.

When only the hook raises:

• First close/aclose raises HookError.
• The second call returns None, although the step was never recorded.

This appears to result from the callback exception replacing the cleanup exception in the first call, combined with _callback_realized being set before callback execution.

The behavior is inconsistent with the aclose() docstring, which promises that cleanup exceptions are raised again on subsequent calls.

I suggest defining an explicit exception-precedence contract and adding tests for both failure combinations.

Neither finding invalidates the main early-close implementation. I am sharing these as collaborative follow-up findings rather than requesting changes to the core mechanism.

[skrawcz]

Thanks for the thorough matrix, both findings were real.

1. Agreed: under an anyio cancel scope, an await in the action's own cleanup is cancelled (not Burr-specific, as you found). The docs and example no longer say the action "is still cleaned up". They now say cleanup starts and an await in it is cancelled unless wrapped in anyio.CancelScope(shield=True), and the cancel-scope test pins both the unshielded and the shielded case.

2. Fixed in 049ee0a. Sync and async now follow one rule: the first error wins (the cleanup error, else the post_run_step hook error), a hook error superseded by a cleanup error is logged, and every close()/aclose() raises the same error. Both failure combinations are covered, for sync and async, via explicit close and via break then close.

[xihaian251]

Thanks for addressing both findings so thoroughly. I appreciate the clarification around AnyIO cancellation semantics and the consistent error-handling contract. The added tests cover the cases I reported. LGTM from my side!