[mohitdixit02]

Resolves #1017 (and addresses related issues #979 and #1007).

In stream_result() and astream_result(), raw inputs were passed to pre_run_step hooks before dependency injection and input validation took place. This caused hooks to miss framework-injected dependencies (like __context and __tracer) and observe undeclared inputs that the action never receives. Furthermore,
input validation errors occurred after hooks had already fired, leaving hook lifecycle state dangling.

Changes

1. stream_result() & astream_result() in burr/core/application.py:

• Reordered execution to _increment_sequence_id() → _process_inputs() → pre_run_step, matching existing step() / astep() semantics.
• The injected ApplicationContext.sequence_id now matches the announced step ID across both fresh (0) and resumed (k+1) applications.
• Validation failures abort before step hooks fire, preventing orphaned pre_run_step state.
• Injected dependencies (action_inputs) are passed directly to pre_run_step hooks.

2. _step() and _astep():

• Added _processed_inputs support so non-streaming halts in stream_result and astream_result forward the already-processed inputs rather than re-invoking dependency factories.
• Updated _astep() to dispatch action_inputs to pre_run_step (matching _step()).

3. Tests:

• tests/core/test_application.py: Added focused regression tests covering sequence ID matching, input filtering, and single-factory invocation across stream_result, astream_result, and astep (152/152 passed).
• tests/core/test_lifecycle_input_semantics.py: 135-case cross-entry lifecycle matrix contributed by @xihaian251 (131 passed, 4 expected strict xfails for fix(core): only end a stream that was actually started #1000).

How I tested this

• Ran pytest tests/core/test_application.py (152 passed).
• Ran pytest tests/core/test_lifecycle_input_semantics.py (131 passed, 4 xfailed).
• Formatted with black --line-length=100 and verified license headers via python scripts/check_asf_headers.py.

Checklist

• PR has an informative and human-readable title
• Changes are limited to a single goal
• Code passed the pre-commit check
• Any change in functionality is tested

[enekoreto]

The fix looks right to me, and it covers both entry points on its own. I checked out the branch (3d34cd1): the 3 new tests fail on main (222b32c) and pass here, also on Python 3.10, and CI's test command passes. With the reproduction from #1017, stream_result and astream_result now both give the hook ['__context', 'value']. I also tried a reserved key and a missing input on both entry points, with a plain and a streaming action and the local tracker attached: all of them raise ValueError and fire neither pre_run_step nor post_run_step. On main the async ones raise KeyError.

But pre-commit fails here: black wants to reformat the new tests in tests/core/test_application.py (blank lines after the docstrings and one long astream_result(...) line), so the pre-commit job will go red. Running pre-commit run --all-files fixes it.

Small thing, not blocking: this conflicts in application.py with the two astep fixes, #980 and #995, since both change _step/_astep. #995 adds the same _processed_inputs parameter, so that one should be easy. #1018 fixes the same issue without touching those methods, but leaves the non-streaming astream_result case to #980/#995, so it's mostly a question of merge order.

I ran this on Python 3.13 on macOS, and the tests also on 3.10.

[mohitdixit02]

Thanks a lot @enekoreto for the detailed review and for testing across Python 3.10 and 3.13!

I've formatted the test file with black --line-length=100 and pushed the update (commit 05ddd2e). All 145 tests continue to pass cleanly.

Regarding the overlap with #995: glad to hear that #995 is using the same _processed_inputs pattern. If needed, I can easily rebase this branch once the related PRs land.

[xihaian251]

Thanks for picking this up from #1017, Mohit — and for the black fix in 05ddd2ef, I re-ran the repo-pinned black 23.11.0 --line-length=100 plus isort --profile=black and flake8 on this head and burr/core/application.py is clean now.

I want to start with the part I think is right, because it is the architectural decision: forwarding the already-validated inputs into _step/_astep instead of re-running the dependency factories is the better shape, and it is strictly better than what I did in my own #1018. #1018 deliberately leaves step/astep untouched and hands them to #980/#995, so it cannot fix the astep() half of #1017 by itself; your branch does the whole thing in one pass. If this lands I'm happy for #1018 to be closed, and for #1008 (also mine, same region) to be dropped — I checked that #1019's head makes #1008's two regression tests in tests/core/test_stream_input_processing.py pass unmodified.

Two things in 05ddd2ef I'd like to fix here rather than after a merge. Both are visible from the outside — one hook plus one action that reports its own injected __context — so nothing below depends on reading private state.

---

1) In both streaming entry points the injected ApplicationContext now carries the previous sequence id

_process_inputs() injects with whatever the counter holds at the moment it runs (application.py:1069: self._dependency_factory[input_](action, self.sequence_id)), and _context_factory() writes that value straight into the context (:947). In this branch _process_inputs() runs one statement before _increment_sequence_id() — stream_result at :1562-1563, astream_result at :1821-1822 — while step()/astep() still allocate first (:943, :1103).

Minimal reproduction (identical on main and on this branch; only the second line differs):

REPORT = {}

@action(reads=[], writes=["value"])
def record(state: State, __context: ApplicationContext) -> State:
    REPORT["ctx"] = __context.sequence_id
    return state.update(value=1)

class Watch(PreRunStepHook):
    def pre_run_step(self, *, sequence_id, inputs, **kwargs):
        REPORT["hook"] = sequence_id

app = (ApplicationBuilder().with_actions(record).with_entrypoint("record")
       .with_transitions(("record", "record")).with_hooks(Watch()).build())   # fresh app
_, c = await app.astream_result(halt_after=["record"])
_ = [r async for r in c]; await c.get()
print(REPORT["hook"], REPORT["ctx"])

fresh application      main: 0 0        this branch: 0 None
resumed at seq=5       main: 6 6        this branch: 6 5

Two distinct effects, and I measured both rather than inferring them:

• fresh application → None. _increment_sequence_id() maps "absent" to 0 (:2155-2159) and sequence_id is self._state.get(SEQUENCE_ID) (:2137-2143), so on the first step the context is built while the counter is still unset. The receiving function is annotated for a number — def _context_factory(self, action: Action, sequence_id: int) (:947) — while the field itself is Optional[int] (ApplicationIdentifiers, :625), which is why the None is stored silently instead of surfacing.
• resumed application → off-by-one. The context says k, the announcement and the action say k+1, so the injected object describes where the application was rather than the step being run.

Which of the two is the intended one is stated in the code: the docstring on Application.sequence_id (:2138-2140) reads "This is incremented prior to every step. Any logging, etc... will use the current step's sequence ID", and on main the with self.context: scope in _step/_astep (:968, :1114) is entered after the allocation, so it has always agreed.

Why I don't think it is cosmetic: parallelism.py:106-112 feeds parent_context.sequence_id directly into .with_spawning_parent(sequence_id=...) for every child application, TracerFactoryContextHook keys token_pointer_map on (app_id, sequence_id) and resets through a bare subscript (:826, guarded only by except ValueError), and SEQUENCE_ID lives inside self._state — so it is persisted per state line and written back as prior_state["__SEQUENCE_ID"] when a run is replayed (burr/tracking/client.py:370, :658). I have not observed corrupted tracking data; what I measured is the number that reaches these consumers, and I'd rather flag it than assert downstream damage.

2) _astep() computes action_inputs but still announces the raw inputs

:1120-1124 hoists the processing, and :1130 is still inputs=inputs, inside the pre_run_step call. So for astep() — and for the non-streaming halt branches of astream_result/astream_iterate — the hook receives the user's dict while the action receives the injected one:

this branch:   hook sees an injected __context = False
with the fix:  hook sees an injected __context = True

That is the first behaviour listed in #1017 ("pre_run_step seeing the raw input dictionary rather than the processed/injected inputs"), and the branch is one word away from it:

@@ -1127,7 +1127,7 @@ class Application(Generic[ApplicationStateType]):
                     "pre_run_step",
                     action=next_action,
                     state=self._state,
-                    inputs=inputs,
+                    inputs=action_inputs,
                     sequence_id=self.sequence_id,

---

On the ordering question — I don't want to overrule your design intent here

You wrote in #1017: "On validation failure, the exception raises before sequence increment or any lifecycle hook dispatch (preventing orphan tracking entries and avoiding the KeyError)." That is a coherent goal, and this branch implements it in the two streaming entry points — which is exactly why a rejected input there consumes nothing.

My only concern is that step()/astep() still consume a number for a step that never ran, so the branch ends up with two numbering rules at once. I verified that reading directly:

missing required input, fresh app     main: step/astep/stream_result/astream_result/…  consumed 1
                                        this branch: step/astep consumed 1, streaming entries consumed nothing
resumed at 5, reserved key            same split

There are two consistent ways out, and the choice belongs to you and the maintainers, not to me:

(a) allocate, then validate — keep step()/astep() as the reference and move the increment back above _process_inputs() in both streaming entries:

@@ -1559,8 +1559,8 @@   # and identically around :1818 in astream_result
+            self._increment_sequence_id()
             action_inputs = self._process_inputs(inputs, next_action)
-            self._increment_sequence_id()

Together with (2) that is 3 changed lines, it restores the documented "incremented prior to every step" property, and it changes no numbering that existing users already have persisted: a rejected step consumes a number on every entry point, exactly as step() does today.

(b) validate, then commit, but stamp the prospective number into the context — keep your order everywhere and let _process_inputs() take the number the step is about to receive (0 if unset, else sequence_id + 1), with step()/astep() no longer pre-allocating. I built this variant to find out whether it is actually cheap, because "validation failures should not consume a sequence id" is a position I share. It is not free:

• +42/−18 in application.py: a new private helper, an optional argument on _process_inputs, and edits in step/astep/_step/_astep — i.e. it lands directly on top of the #980/#995 region (:958) that is already contended.
• it removes one incoherence and creates another: with self.context: is entered at :968/:1114, i.e. before the allocation the variant performs, so ApplicationContext.get() — the documented way to connect two applications, and what parallelism.py:112 reads — still lags one step (4 measured mismatches: published None/0/5/6 against announced 0/1/6/7). Closing that needs a restructure of the with self.context: scope in both methods.

So my lean is (a) for this PR — 3 lines, keeps your architecture and your tests — with (b) as its own follow-up if the maintainers agree that a rejected step should not consume a sequence_id. My argument for separating them is not taste: SEQUENCE_ID is persisted state and is replayed from the tracking service, so (b) changes a user-visible numbering property, and I don't think that should arrive as a side effect of a lifecycle-ordering fix. If (b) is preferred, I'll write it up as a proposal against #1017 with the measurements above, and equally happy if you take it yourself.

Tests

If it is useful, take the regression matrix instead of my description of it: tests/core/test_lifecycle_input_semantics.py, 120 collected / 106 effective — the 14 skips are deliberately invalid axes (e.g. driving a streaming action through step()) and are never counted as passes. It covers the six entry points (step, astep, stream_result, astream_result, stream_iterate, astream_iterate) × sync/async/streaming actions × sync/async hooks × tracking on/off, plus fresh vs resumed applications and rejected inputs, and it judges by _process_inputs call counts and injected-object identity (is), not by return values.

On this head, re-measured today:

this head (05ddd2ef)          34 failed / 72 passed / 14 skipped
  24  test_injected_context_sequence_id_matches_the_announced_step      <- point 1
   4  test_hook_and_action_receive_the_same_injected_object[astep-*]    <- point 2
   2  test_rejected_step_follows_one_sequence_id_rule_across_entry_points
   2  test_sync_action_via_async_entry_processes_once_and_shares_the_context[astream_*]
   2  test_action_exception_type_message_and_records_survive[astream_*]  <- #1000's issue, not yours
+ 1 line (point 2)            30 failed / 76 passed / 14 skipped
+ 3 lines (points 1 + 2)       2 failed / 104 passed / 14 skipped
+ #1000 stacked                0 failed / 106 passed / 14 skipped

The same suite is also what makes a mutation run meaningful: reverting the ordering, stamping the previous number into the context, re-computing the inputs in the delegate, re-announcing a rejected input, double-dispatching post_run_step, and rebuilding a dependency all produce 28/35/20/40/9/3 newly failing cases respectively, so the green result after a fix is discriminating rather than vacuous.

I'd rather see this file land in your PR than in a competing one. Say the word and I'll open it as a test-only PR against your branch, or push it somewhere you own — whichever is easier for you. And since #995 adds the same _processed_inputs idea, the :958 conflict you flagged with @enekoreto looks mechanical to resolve in a rebase once the related PRs land.

Scope and verification limits, stated plainly

• Verified locally on Python 3.12.14 and 3.13.1 (Windows); identical results on both. I have no 3.10/3.11 on this machine, so the 3.10 evidence in this thread is @enekoreto's, not mine.
• Hosted CI has not run the test workflow on this head: Build Burr and Release Validation are action_required (awaiting maintainer approval) and documentation is in startup_failure — that last one also on main's own recent pushes, so it is upstream infrastructure rather than this PR. Nothing here should be read as "CI passed", and nothing here is a claim about the test suite on Linux.
• On this machine tests/core + tests/tracking at this head gives 11 failures: 9 are tests/core/test_graphviz_display.py cases that fail on main here too because the graphviz package is not installed in this venv (they surface as AttributeError: 'NoneType' object has no attribute 'edge_attr' from burr/core/graph.py:232), and the remaining 2 are #1000's astream_result error path. Mentioning this only so nobody re-diagnoses those as regressions introduced by this PR.
• One thing I believe is not this PR's problem, so it doesn't get lost: a failing non-streaming action under astream_result still fires post_end_stream for a stream that never started, and the hook's KeyError replaces the action's ValueError. #1000 covers that separately, and applying #1019 (with or without the 3 lines) leaves it in place — which is why the two PRs are complementary rather than duplicated.

Overall I think this branch is the right home for the fix, and both points above are 3 lines. Marking this as a comment rather than a change request, since the remaining question — whether a rejected step should consume a sequence_id — is a design decision for the maintainers.

[xihaian251]

Re-ran my independent verification against the current head 3833608a on CPython 3.10.22 / 3.11.16 / 3.12.14 / 3.13.1. All results below are local; hosted CI has not executed pytest on this PR yet.

The implementation correctly resolves #1017, #979 and #1007.

• The 135-case cross-entry matrix reports 65 failures on main and 131 passed / 4 expected xfails / 0 failures on this PR, consistently across all four Python versions.
• Running with --runxfail exposes the four underlying failures, all belonging to astream_result() masks action exceptions with a tracker KeyError when the action is non-streaming #999 (the independent lifecycle issue addressed by fix(core): only end a stream that was actually started #1000).
• Applying fix(core): only end a stream that was actually started #1000 on top of this head gives 135 passed / 0 failed under --runxfail, with no regression in the repository suites.
• tests/core/test_application.py: 147 passed.
• tests/core + tests/tracking: 444 passed / 2 skipped, excluding test_graphviz_display.py. That file separately has 9 failed / 3 passed on both main and this PR because Graphviz is unavailable in my local environment.
• Object identity and lifecycle behavior were verified: the hook and action receive the same __context object, dependency factories execute once per step, and sequence IDs remain consistent across sync/async and fresh/resumed applications.
• black 23.11.0, isort, flake8 and check_asf_headers.py all pass.

Two remaining points:

1. The four expected failures belong to astream_result() masks action exceptions with a tracker KeyError when the action is non-streaming #999 / fix(core): only end a stream that was actually started #1000, not this PR. fix(core): only end a stream that was actually started #1000's stream_started guard applies cleanly on top of this head. I recommend keeping the two PRs separate because they address distinct lifecycle issues.
2. The current PR description needs a factual correction. At 3833608a, the order is _increment_sequence_id() → _process_inputs() → pre_run_step, matching the existing step() / astep() semantics. A rejected input consumes its sequence number but aborts before either step hook fires. The description should also reflect the updated 147-test count and the _astep() / _processed_inputs changes.

CI status:

Build Burr (41450356289) and Release Validation (41450355638) remain action_required and have never executed. Both require maintainer approval. The documentation workflow's startup_failure is also present on main, so it appears unrelated to this PR.

I have not submitted an APPROVE review. These local results should not be confused with hosted CI results.

I also verified that merging the latest main into this head is conflict-free and preserves the regression matrix results.

Once the two pending workflows have been approved and their results reviewed, this PR should be ready for a maintainer's merge decision.

Thanks for the work on this fix.

[xihaian251]

Follow-up to the earlier review — this is about integration sequencing rather than the code, which I re-ran against current main (3f16c1e6) in a local worktree.

Relationship to #995
This PR is a strict superset of #995 in burr/core/application.py. If #995 lands first, a rebase leaves exactly four meaningful hunks here: the _astep _processed_inputs parameter, and the process-before-hook plus halt-forwarding changes in stream_result and astream_result. The _step/_astep hook-side hunks would become duplicates and can be dropped, which makes the remaining diff much easier to review.

What this PR is uniquely responsible for
#995 does not touch the streaming entry points. On a 19-case cross-entry consistency matrix (step, astep, stream_result, astream_result × sync/async × streaming/halt × valid/missing/reserved inputs): main 16 failed / 3 passed, #995 alone 12 failed / 7 passed, this PR 19 passed. It is also the only change that fixes the non-streaming halt processing inputs twice (#1007) — verified with a separate 8-case probe that fails 6/8 on main and on #995, and passes 6/6 here (2 cases skipped because stream_result cannot drive an async action).

Two suggestions after a rebase

1. Port the three tests from fix(core): process and validate inputs before pre_run_step in astep (#979) #995. This PR has no assertion that _astep's dependency factory runs exactly once, nor that the hook and the action share one injected object. test_step_and_astep_share_one_processed_inputs_dict is currently the only test that pins that guarantee, and it would be lost if this PR merges alone.
2. Add an astream_result counterpart to test_stream_result_shares_one_processed_inputs_dict. The async halt path is the only part of this change that depends on the new _astep parameter, and it is the part with no committed test. A working version is available if useful.

Minor
# In this case we allow the self._step to do input processing and # In this case we want to process inputs because we run the function directly are no longer accurate after this change — #995 removes both.

CI
Hosted CI has not run here either: Build Burr is action_required (awaiting maintainer approval to run), and documentation shows startup_failure on every PR head including unrelated ones. All figures above are local, CPython 3.12.

[xihaian251]

Hi Mohit and maintainers,

Thanks for merging my test-only PR #1 into fix/stream-result-process-inputs. I appreciate the collaboration.

The merged commit (31b0edc) adds tests/core/test_lifecycle_input_semantics.py, a 135-case regression matrix covering all six execution entry points, action types, hook configurations, tracking modes, and fresh/resumed applications.

The latest Apache Burr PR #1019 head is now 50520ef.

1. Current integration status

The latest branch includes:

• The streaming input-processing fixes for stream_result()/astream_result() fire pre_run_step before inputs are processed #1017 and stream_result()/astream_result() process inputs twice when halting on a non-streaming action #1007.
• The _processed_inputs handoff in _step() and _astep().
• Regression coverage for single dependency-factory invocation and object identity between lifecycle hooks and actions.
• The newly integrated 135-case lifecycle semantics matrix.
• Additional focused halt and shared-context tests in test_application.py.

The four strict expected failures in the new matrix are explicitly associated with #1000's separate exception-masking issue, not regressions introduced by #1019.

2. Coordination with PR fix(core): process and validate inputs before pre_run_step in astep (#979) #995

PR #995 remains open and contains the same fundamental _processed_inputs handoff for the synchronous-action path through astep().

The latest #1019 branch now covers this behavior as well, including the single-factory-call and object-identity guarantees.

I see two possible integration sequences:

Option A: Merge #995 first, then rebase #1019 and remove the overlapping implementation hunks while retaining the broader streaming changes and regression coverage.

Option B: Use #1019 as the integration PR, since it now contains the overlapping fix and broader test coverage, and coordinate with the author of #995 on whether that PR should be superseded.

I would appreciate the maintainers' preference. I am happy to follow either sequence and do not intend to close or retarget another contributor's PR independently.

3. Hosted CI approval

On the latest #1019 head (50520ef), the required workflows are still awaiting approval:

• [Build Burr](https://github.com/apache/burr/actions/runs/41624590032) — action_required
• [Release Validation](https://github.com/apache/burr/actions/runs/41624590184) — action_required

Both currently have zero jobs, so no hosted pytest execution has taken place.

Could a committer please approve these workflows once the preferred integration sequence is agreed upon?

The branch is currently reported as mergeable, although it is 13 commits behind the latest main. Please let me know if you would prefer an update against main before running the workflows.

The documentation workflow also reports startup_failure with zero jobs; I am treating this separately from the required test workflows.

4. One documentation correction

Before the final review, I noticed that the current #1019 description still says _process_inputs() executes before _increment_sequence_id() and that rejected inputs do not consume sequence IDs.

The actual implementation increments the sequence ID before processing inputs. Validation failures therefore consume the ID but occur before lifecycle hooks are dispatched.

The new regression matrix checks consistency across execution entry points and verifies that the injected context uses the announced sequence ID. The PR description should reflect this actual behavior and include the newly integrated test suite.

All previously reported test results are local results, not hosted CI results. The latest head should be used for the final verification.

Thanks again for the work and coordination.

[mohitdixit02]

Thanks for the breakdown and for laying out the options for the maintainers @xihaian251!

I've just updated the PR description above to accurately reflect the sequence ID ordering (_increment_sequence_id → _process_inputs → pre_run_step) and included the updated test details.

Completely open to whichever integration path the maintainers prefer (Option A or B). If maintainers would like this rebased onto the latest main before approving the CI runs, just let me know and I'm happy to push a quick rebase.