Skip to content

fix(core): report a streaming step's own error, not the caller's or a hook's - #1033

Open
xihaian251 wants to merge 2 commits into
apache:mainfrom
xihaian251:streammain36
Open

xihaian251 wants to merge 2 commits into
apache:mainfrom
xihaian251:streammain36

Conversation

@xihaian251

Copy link
Copy Markdown

A streaming step currently decides whether it failed by reading sys.exc_info()[1]
inside the iterator's finally. That value is thread state, not the step's own
exception: when the caller consumes a successful stream from inside an except
block, the caller's exception is reported as the step's failure, persisted as
failed, and the step is executed a second time on resume.

This change records the exception from the iterator's own except clauses instead,
keeps the step's own error when a lifecycle hook fails, and stops a failing
post_run_step hook from skipping post_end_stream. close() / aclose() are
untouched.

Problem

Two symptoms, one finally in StreamingResultContainer.__iter__ /
AsyncStreamingResultContainer.__aiter__:

  1. A successful step is recorded as failed. sys.exc_info()[1] inherited the
    caller's in-flight exception. PersisterHook stores failed,
    _init_state_from_persister calls reset_to_entrypoint() for anything not
    completed, and resume re-runs the action — so a non-idempotent side effect
    (a charge, an email, an order) happens twice. stream_iterate /
    astream_iterate create one container per step, so every step in the chain is
    affected.
  2. A hook error replaces the action's error. post_run_step is dispatched
    from the same finally; a plain Exception raised there overwrites the
    action's exception, and the remaining lifecycle callbacks
    (post_end_stream, execute-method post) never run. That leaves the
    pre_start_stream / post_end_stream pair unbalanced — LocalTrackingClient
    releases its per-stream state only in post_end_stream, so a telemetry hook
    failure leaks stream_state and writes an initialize_stream with no matching
    end_stream.

Minimal reproduction (sync; async behaves the same):

try:
    raise ValueError("the caller's own error")
except ValueError:
    _, container = app.stream_result(halt_after=["charge"])
    container.get()          # charge succeeds and commits state

# main:  post_run_step sees ValueError, SQLite row status='failed'
#        resume -> next action is 'charge' again -> charge runs twice
# patch: post_run_step sees None, status='completed'
#        resume -> next action is 'done'

Root Cause

sys.exc_info() returns the exception state of the current thread, and a
generator inherits that state when it is first resumed. After
except StopIteration: return, the finally still sees the inherited exception.
Burr used that value as the step's exception= argument, which feeds both
post_run_step hooks and PersisterHook.save(..., status).

The ownership boundary belongs on the iterator frame: only an exception raised
inside this iterator's own try and caught by its own except clauses describes
this step. Reading thread state crosses that boundary.

Two consequences of the same design mistake:

  • an exception that does not belong to the step changes the persisted outcome;
  • an exception raised while reporting the step can replace the exception that
    caused the step to fail.

Wrapping the call in a broader except would not help: it would still be reading
thread state rather than the step's own control flow, and would risk swallowing
KeyboardInterrupt / CancelledError / SystemExit.

Changes

  • burr/core/action.py
    • StreamingResultContainer.__iter__ and
      AsyncStreamingResultContainer.__aiter__ now keep a local exc, assigned only
      by this iterator's own except GeneratorExit / except BaseException
      clauses. except StopIteration / except StopAsyncIteration leave it None.
      sys.exc_info() is no longer used in either container.
    • Added _realize_callback_keeping_step_error (sync + async): when the step
      already has an error, a callback Exception is logged and does not replace
      it; when the step succeeded, the callback error still propagates. A callback
      BaseException is never caught.
    • close() / aclose() are not modified.
  • burr/core/application.py
    • The stream_result / astream_result callbacks are now two-stage: if
      post_run_step raises, post_end_stream and the execute-method post hook
      still run, then the first hook error is re-raised. The container decides
      whether that error is allowed to supersede the step's own error.
  • tests/core/test_streaming_exception_precedence.py — 30 cases (24 from the
    original analysis + 6 added for the async close/cancel race and the stream-end
    pairing).

Net: 3 files, +851 / -40 (716 lines are tests). No public API change; no new
dependency.

How I tested this

All runs used isolated git worktrees with a unique --basetemp per run. Baseline
is 36e3ff30; the branch is 36e3ff30 -> 6a840f1c (fix) -> acee6f3b (tests only).

Check Clean main With this patch
30 regression cases (3.10.22) 19 failed / 11 passed 30 passed
30 regression cases (3.11.16) 19 failed / 11 passed 30 passed
30 regression cases (3.12.14) 19 failed / 11 passed 30 passed
30 regression cases (3.13.14) 19 failed / 11 passed 30 passed
tests/core minus the dot-dependent display tests (3.13.14) 473 passed 503 passed
tests/core minus the dot-dependent display tests (3.10.22) 473 passed 503 passed
Full suite (scope note below, 3.13.14) 15 failed / 684 passed / 3 skipped 15 failed / 714 passed / 3 skipped

The 15 failures are identical in both trees (diff of the FAILED lists is empty):
5 need the dot executable, 6 are pre-existing test_apache_release* template
failures, and 4 are release/verify tool tests that fail only when run together
(they pass individually on both trees). The +30 passed is exactly the number of
new cases.

  • Async close/cancel race, reproduced deterministically. The narrow window is:
    aclose() is in flight and parked inside the action's cleanup, the iterator's
    GeneratorExit handler is parked on asyncio.wait([self._close_future]), and
    that finalizer task is cancelled. Built with asyncio.Event handshakes and
    explicit tasks (no sleep used as a synchronisation primitive). Result: main
    reports the caller's CancelledError to post_run_step; with this patch it
    reports GeneratorExit, which is what aclose's docstring and fix(core): propagate early close through streaming actions #1027 promise.
    Persisted status is failed on both trees; the action's cleanup runs exactly
    once; no unhandled task exceptions. Pinned by
    test_async_aclose_race_reports_the_early_stop_not_the_cancellation.
  • Stream-end pairing. A failing post_run_step hook used to skip
    post_end_stream. Measured with LocalTrackingClient: on main the tracker
    keeps stream_state[('app','a','pk')] forever and log.jsonl has
    initialize_stream with no end_stream; with this patch the state is released
    and the log is balanced. Pinned by
    test_a_failing_hook_does_not_leak_the_tracker_stream_state.
  • 16 mutation targets, all killed (each breaks exactly one guarantee):
    restoring sys.exc_info(); dropping the iterator's BaseException record;
    forcing exc=None; removing error precedence; catching BaseException;
    swallowing a hook error on success; treating early stop as success; letting a
    hook error beat a cleanup error; breaking only the async path; truncating the
    lifecycle; re-aborting the lifecycle on a sync hook failure; reverting only the
    async finally to sys.exc_info(); swallowing the re-raised hook error;
    letting a hook error replace the step error again; skipping post_end_stream
    in the async path only; and not recording cancellation on the async path.
  • Persistence/resume: SQLitePersister stores completed, three consecutive
    resumes never re-run the action, sequence IDs 0 -> 1 without a UNIQUE conflict.
  • fix(core): propagate early close through streaming actions #1027 early-stop contract: GeneratorExit / cancellation reporting, cleanup
    error precedence, and repeated close() / aclose() raising the same error are
    unchanged (verified on both trees).
  • black 23.11.0, isort 5.12.0, flake8 6.1.0 and
    scripts/check_asf_headers.py all pass on the three changed files (versions
    pinned by the repo's .pre-commit-config.yaml); git diff --check is clean and
    the patch applies to 36e3ff30 with git apply --check.

Scope note (not an overclaim): the repo's CI command is
pytest tests --ignore=tests/integrations/persisters --ignore=tests/integrations/test_bip0042_bedrock.py.
On this machine it stops during collection with 8 errors because optional extras
(langchain_core, langchain_community, hamilton, haystack,
opentelemetry.sdk, ray) are not installed. The table above excludes exactly
those files. CI covers Python 3.10/3.11/3.12; I ran the 30 regression cases on
3.10.22, 3.11.16, 3.12.14 and 3.13.14 with identical results, and tests/core
on 3.10.22 and 3.13.14 (473 -> 503 on both). The full-suite comparison above was
run on 3.13.14. The project has no mypy/pyright/ruff type configuration, so no type
check was run. Dependabot is already failing on main's check-runs and is
unrelated to this change.

Compatibility

  • No public API or signature change. Hook exception values for a successful step
    stay None; for a failed step they are now the step's own exception rather than
    an unrelated one.
  • BaseException (KeyboardInterrupt, SystemExit, asyncio.CancelledError,
    GeneratorExit) is never swallowed. A hook raising a BaseException still
    propagates immediately and the action error remains in __context__.
  • close() / aclose() are byte-for-byte unchanged, so the early-stop contract
    introduced in fix(core): propagate early close through streaming actions #1027 is preserved.
  • Deliberate behaviour changes (not a pure internal refactor — two of them are
    observable):
    1. After any failing post_run_step hook — including on a step that
      otherwise succeeded — post_end_stream and the execute-method post hook now
      run. Previously they were skipped. This is required to keep
      pre_start_stream / post_end_stream balanced for hooks that keep per-stream
      state (the local tracker, the S3 tracker, the OpenTelemetry adapter).
    2. On a step that already failed, a post_run_step hook raising a plain
      Exception is now logged (with traceback) instead of replacing the action's
      error. On a successful step the hook error still propagates.
    3. In the narrow async race described above, post_run_step receives
      GeneratorExit instead of CancelledError. Persisted status is failed
      either way.
  • Known limitations, pre-existing on main and out of scope:
    • a failing post_run_step hook still prevents later post_run_step hooks in
      the same dispatch loop (burr/lifecycle/internal.py) from running;
    • cancelling the consuming task during reporting still skips
      post_end_stream (the callback catches Exception, not BaseException);
    • a persister error still changes the exception the caller sees;
    • stream_result / astream_result halting at a non-streaming action
      dispatches post_run_step twice when a hook fails
      — the second dispatch
      passes the hook's own exception as the step's exception=, so the tracker
      records one step twice, the second entry as failed. Different code path
      (application.py, non-streaming branch), pre-existing, and it needs a
      dispatch-level decision rather than a local fix; I have a minimal repro and
      can file it as a separate issue.

Relation to #1027 and to open PRs

#1027 (fix(core): propagate early close through streaming actions, merged as
36e3ff30) reworked the streaming containers and established: early stop reports
GeneratorExit/cancellation, cleanup errors win on close()/aclose(), and
repeated close raises the same error. This PR does not redo any of that and
does not touch close()/aclose(); 4 regression cases specifically pin those
contracts. It addresses two gaps #1027 left open: the finally still used
sys.exc_info()[1], and a hook error could still replace the action's error.

This PR is independent of the separate non-streaming _step/_astep hook
error-precedence fix — different execution path, different patch; that fix is
not included here.

For orientation only (no code overlap):

Notes

AI Disclosure

Analysis, test design, mutation design, and the drafts of these documents were
produced with WorkBuddy AI (Tencent Hunyuan Hy). The code changes, the
reproduction scripts, and every verification step were reviewed and executed
locally by the contributor; all commands and raw outputs are retained in the
working notes and can be re-run. The tool's exact version is not exposed in this
environment, so no version string is asserted here.

Per the ASF Generative Tooling Guidance,
contributors are responsible for their submissions and are encouraged to
identify tooling in the commit message (e.g. Generated-by: <Tool> <Version>).
The current local commits do not yet carry that token; I will add it before
publication if the maintainers prefer. The Burr PR template and
CONTRIBUTING.rst have no dedicated AI field, so the disclosure is placed here.

Out of scope

  • Adapter-dispatch isolation in burr/lifecycle/internal.py (one failing hook
    blocking later hooks, including the persister).
  • The non-streaming _step/_astep hook precedence fix.
  • Cancellation during reporting still skipping post_end_stream.
  • The non-streaming branch of stream_result dispatching post_run_step twice
    when a hook fails (see Compatibility above).
  • _close_action_generator / _aclose_action_generator still take sys.exc_info()[1] as
    their in_flight argument (4 call sites in application.py). I instrumented all four
    across 9 scenarios: the value is correct except when the caller is inside an except
    block while an already-completed stream's wrapper is closed by GC — there it is the
    caller's exception. A controlled experiment (calling the helper with an explicit
    in_flight) shows that such a value would downgrade a cleanup error from "propagates"
    to "logged warning"; but the wrong value and a live cleanup error are mutually exclusive
    in the current code (closing a live generator always sets GeneratorExit), so I could
    not reach it. Business results are identical either way; it is a latent diagnostic-channel
    risk to clean up separately, not a live bug.

Checklist

  • PR has an informative and human-readable title
  • Changes are limited to a single goal (no scope creep)
  • Pre-commit Python hooks pass (black, isort, flake8, check-asf-headers,
    check-ast); frontend-lint-staged was not run (no frontend changes)
  • Any change in functionality is tested (30 cases; 16 mutation targets)
  • New functions are documented (both _realize_callback_keeping_step_error
    variants carry docstrings; no new public functions)
  • Placeholder code is flagged / future TODOs are captured in comments
  • Project documentation has been updated if adding/changing functionality
    — N/A: no public API or documented behaviour was added; the behaviour
    changes are described above and covered by tests. Happy to add a short
    docs/concepts/streaming-actions.rst note if the maintainers want one.

B and others added 2 commits November 7, 2026 12:51
The streaming containers decided what error to report for a step with
`sys.exc_info()[1]` inside the iterator's `finally`. `sys.exc_info()` is thread state: when the
caller was itself inside an `except` block, the iterator reported *that* exception as the step's
failure. A streaming action that completed successfully was persisted as `failed` -- and
`_init_state_from_persister` re-runs anything not stored as `completed`, so resuming executed the
step a second time.

Capture the exception from this iterator's own `except` clauses instead, and use it for the
post-step callback.

While there, keep the step's own error when a hook fails: `post_run_step` is dispatched from the
same `finally`, so a hook raising a plain `Exception` replaced the action's exception. The
container now logs that hook error and keeps the step's error; a hook's `BaseException`
(KeyboardInterrupt / CancelledError / SystemExit) is never caught. The application callback runs
`post_end_stream` and the execute-method post hook even when `post_run_step` fails, so a single
failing hook no longer truncates the lifecycle.

`close()` / `aclose()` keep their own first-error-wins rule: early stop still reports
`GeneratorExit` / cancellation, a cleanup error still wins, and repeated close still raises the
same error.

Tests: tests/core/test_streaming_exception_precedence.py, 24 cases; 14 of them fail on main
(36e3ff3).
The narrow aclose race is now reproducible without sleeps: hold aclose()
inside the action's cleanup, let the iterator's GeneratorExit path park on
_close_future, then cancel that finalizer task. The step was stopped early,
so GeneratorExit is the error to report; reading the thread's current
exception reported the caller's CancelledError instead.

Also pins that a failing post_run_step hook no longer skips
post_end_stream: the local tracker releases its per-stream state there, so
skipping it leaked stream_state and left an unterminated stream in
log.jsonl.
@github-actions github-actions Bot added the area/core Application, State, Graph, Actions label Nov 7, 2026
@xihaian251

Copy link
Copy Markdown
Author

Hi maintainers — this is a follow-up to #1027 (early close), already merged as
36e3ff3. I found two gaps in the streaming containers' finally and would
appreciate a review.

  1. A successful streaming step can be persisted as failed. sys.exc_info()[1]
    inside StreamingResultContainer.__iter__ / AsyncStreamingResultContainer.__aiter__
    reads thread state, so if the caller is inside an except block while consuming
    the stream, that unrelated exception becomes the step's exception=.
    PersisterHook then writes failed, and resume re-runs the action. With a
    non-idempotent action the side effect happens twice (SQLite repro: count 1 -> 2).
    stream_iterate marks every step in the chain failed the same way.

  2. A hook raising a plain Exception from that same finally replaces the
    action's real exception and truncates post_end_stream / execute-method post.
    Truncating post_end_stream is not just a missing log line: LocalTrackingClient
    releases its per-stream state only there, so the entry leaks and the tracking
    log gets an initialize_stream with no matching end_stream. I measured both
    directions on a successful step with one failing post_run_step hook.

The patch records the iterator's own exception via its except clauses, keeps the
step's error when a hook fails (hook BaseException still propagates), and lets
the remaining lifecycle callbacks run. close() / aclose() are untouched — four
regression cases pin #1027's early-stop contract.

On the async close race we discussed last time: I can now reproduce it
deterministically. Hold aclose() inside the action's cleanup, let the iterator's
GeneratorExit handler park on asyncio.wait([_close_future]), then cancel that
finalizer task. main reports the caller's CancelledError to post_run_step;
with the patch it reports GeneratorExit, which is what aclose's own docstring
and #1027 promise. Persisted status is failed on both, cleanup runs once, no
unhandled task exceptions. It is pinned by a new test — so this is a contract fix,
not a behaviour regression, and I have written it up as a behaviour change anyway.

Verification on 36e3ff30 (all re-run this round, no inherited numbers): the 30
regression cases go 19 failed/11 passed -> 30 passed on 3.10.22, 3.11.16, 3.12.14
and 3.13.14
; tests/core goes 473 -> 503 passed on both 3.10.22 and 3.13.14; the
full suite's failure list is identical before and after (15 pre-existing environment
failures) with exactly +30 passed; 16 mutation targets all killed; pinned
black/isort/flake8 + ASF header pass; git apply --check on a clean 36e3ff3 tree
passes, and applying the patch file itself and re-running gives 30 passed on all
four versions.

Three things I want to flag rather than hide:

  • I could not run CI's exact pytest tests command locally — 8 optional integration
    extras aren't installed. The PR body states exactly which files were excluded.
  • I am deliberately not fixing four pre-existing issues in this PR: a failing
    post_run_step still blocks later post_run_step hooks in
    lifecycle/internal.py; cancelling the consumer during reporting still skips
    post_end_stream; a persister error still changes the caller-visible exception;
    and — found while checking coverage — stream_result halting at a
    non-streaming action dispatches post_run_step twice when a hook fails, the
    second time with the hook's own exception as the step's, so the tracker records
    the step twice with the second entry marked failed. All four behave identically
    on main and with the patch. I have a minimal repro for the last one and can
    file them as separate issues if you agree with the direction.
  • fix(core): only end a stream that was actually started #1000 / fix(core): only end a stream that was actually started #1016 ("only end a stream that was actually started") guard the outer
    error path in astream_result so a stream that never started is not ended; this
    PR makes sure a stream that did start is not skipped. Same pairing contract,
    opposite side, no textual overlap. fix(core): retry cancelled asynchronous actions after resume #1032 (_astep cancellation) and fix: report execute hook failures #897
    (execute-hook wrappers) are also untouched; I'll rebase if they land first.

The contributor used WorkBuddy AI (Tencent Hunyuan Hy) for analysis and drafting;
all code, reproductions and verification were reviewed locally, and the disclosure
is in the PR body. Happy to add a Generated-by: commit trailer per the ASF
generative tooling guidance if you'd like one.

This branch has not been deployed

No deployments
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

area/core Application, State, Graph, Actions

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant