[Steven Huang (remote-controlled-man)]

Motivation and Context

Fixes #14759.

When a local process function awaits, later messages can reuse its in-flight input dictionary. The sequence a=1, b=1, a=2, b=2 can execute an extra (2,1) invocation. When the second b arrives in a later superstep, completion of the first invocation can also discard the pending a=2.

Description

Consume the filled input buffer before invoking the local step, so later messages populate a fresh dictionary. Completion preserves that newer buffer on both success and error. Public signatures and concurrent local dispatch are unchanged.

Four regressions use ProcessBuilder and the public local start() entry, covering same/later-superstep delivery and successful/failing first invocations.

The Dapr review follow-up is covered by four queue-path controls using a real kernel and step. StepActor.process_incoming_messages() awaits each handle_message() before dequeuing another; handle_message is not an actor RPC. These controls pass with unchanged Dapr runtime code. This assessment covers the current registered queue path with normal actor turn-based access; no live Dapr host or reentrant/direct concurrent invocation was tested.

Merged main at 25bd1bb67c38976bb13958783572efa57d3445d4. Relocated this PR's local test and localized its context import so a git merge-tree with #14732 at 45976a66f2f85fddc294ba044af2fb0d83ed6888 is clean. No code from #14732 was merged.

Validation

Final commit: 1ef82d116cebf88460055efca17babdf335b8c3a.

• All four local regressions fail with main's original local_step.py and pass with this fix. Test assertions and input sequences are unchanged by the relocation.
• All four Dapr queue controls pass with unchanged step_actor.py; as a sensitivity check, all four fail when its sequential queue pump is temporarily replaced with concurrent dispatch. That was an artificial mutation, not a reproduction of a current Dapr bug.
• Complete Python unit suite, split as in CI: 4987 passed, 1 skipped, 5 xfailed outside Dapr; 86 passed in the separate Dapr environment, including the four new controls.
• Applicable native pre-commit hooks on all three changed files passed, including the repository-pinned Ruff 0.9.6 lint/format and Bandit.
• Wheel and source distribution builds passed. Both 575-entry package lists were inspected: packaged LocalStep and StepActor match the final source; no task helpers, environments or temporary test data are included.
• Dependency consistency in both environments and git diff whitespace checks passed.
• Existing warnings remain: uv's prerelease-mode deprecation, Dapr legacy-extension/import deprecations and Pydantic/deprecated API notices.
• Independent AI review covered all three files and the real Local/Dapr dispatch paths, failure/cancellation behavior, adjacent function buffers and resource effects. Final evidence and package inventories were checked against the same commit and environment.

Environment: Python 3.12.14, uv 0.13.7, macOS/Apple Silicon. The main environment includes all optional dependencies and the development group; Dapr has a separate environment as in CI.

The main unit run allows loopback sockets and isolates only macOS system-proxy discovery (urllib.request._get_proxies) inside the test process. Application code, getproxies, proxy_bypass, proxy environment-variable tests and assertions remain unchanged. The host's system proxy otherwise produces unrelated DNS-pinning failures. No provider/service integration suite was run.

Contribution Checklist

• SDK wheel and source distribution build successfully.
• Applicable native pre-commit checks pass.
• Complete Python unit checks pass in the environment described above.
• Public API signatures and existing concurrent local dispatch are preserved.

AI assistance: Codex implemented and tested the changes. A separate AI reviewer independently reviewed the frozen changes and final evidence; no personal human code review is claimed.

[Copilot]

Copilot was unable to review this pull request because the user who requested the review has insufficient remaining quota. Try requesting a review at a lower effort level.

[Muhammad Abdullah Butt (1AbdullahButt)]

The local fix looks right. arguments is bound to the filled buffer (local_step.py:146), and only then is self.inputs[target_function] replaced with a fresh copy of initial_inputs (:154). So the invocation keeps its own dict, and messages that arrive while it awaits fill the new one. The completion path no longer touches self.inputs, so a newer partial buffer survives both OnResult and OnError. Your fail_first case covers that. A function with no inputs (initial_inputs is {}) gets {} back, which is still "invocable", the same as before.

The Dapr step actor has the same pattern. step_actor.py:349 binds arguments, the invoke awaits, and the finally at :380-381 resets self.inputs[target_function] after emit_event. If two messages for one step can be handled while an invocation is in flight there, the same (2, 1) call and the dropped pending input can happen. Dapr actors use turn-based access (Dapr docs), which rules this out only if every message goes through a separate actor call. I didn't trace whether handle_message can run more than once inside one turn. If it can, the same three-line move applies in step_actor.py, and the existing test_step_actor.py setup would take a matching test.

Overlap with #14732: it also rewrites handle_message in both local_step.py and step_actor.py. A git merge-tree of the two PR heads conflicts only in test_local_process.py, because both append tests at the end. The code changes combine cleanly. Whichever lands second only needs that test file rebased.

git merge-tree against main is clean. I didn't run the tests.

Reviewed with help from Claude Code.

[Steven Huang (remote-controlled-man)]

Muhammad Abdullah Butt (@1AbdullahButt) I traced the registered Dapr path: process_incoming_messages() awaits each handle_message() inside the turn, and handle_message is not an exposed actor method. ProcessActor gathers work across steps, rather than messages within one step. With normal Dapr turn-based access, another turn also waits for that async operation to complete, so I did not find the concurrent-buffer condition on this path.

I added four queue-path controls covering same/next-batch delivery and a successful/failing first invocation, using the real kernel and step. They pass with unchanged Dapr runtime code; deliberately making the queue pump concurrent makes all four fail. This is limited to the current registered path, not a live-sidecar or reentrancy test.

I also merged main at 25bd1bb and moved my local regression test plus its context import away from the overlap. A merge-tree with #14732's current head 45976a6 is now clean; none of that PR's code was merged.

Validation at 1ef82d1 passed: complete Python units outside Dapr 4987 passed, 1 skipped, 5 xfailed; Dapr 86 passed, including the four new controls. Native pre-commit, both dependency checks, wheel/sdist build and package inspection also passed. No live Dapr host or provider/service integration suite was run.