fix(sdks/python): send aio_put_stream events with grpc.aio

[juanluisdb]

Description

Fixes #4230.

This changes ctx.aio_put_stream so it sends PutStreamEvent with a lazy grpc.aio events stub.

Before this change, ctx.aio_put_stream called the synchronous put_stream path through asyncio.to_thread. Under high streaming load, those synchronous gRPC calls can keep the shared thread pool busy. The runner also uses that pool to read assigned actions from its queue, so task starts can be delayed.

The stream index is still assigned before the RPC is awaited. This keeps the existing stream ordering behavior.

Type of change

• Bug fix (non-breaking change which fixes an issue)
• Test changes (add, refactor, improve or change a test)

What's Changed

• Add a lazy async events stub to the Python events client.
• Send aio_put_stream events with grpc.aio instead of asyncio.to_thread.
• Share stream request construction between the sync and async stream paths.
• Add unit coverage for async stream retry behavior and stream index ordering.

Checklist

Changes have been:

• Tested (unit, integration, or manually with steps specified)
• Linted and formatted
• Documented (where applicable)
• Added to CHANGELOG (where applicable)

Testing

I ran these checks locally with the project virtualenv:

pytest tests/unit/test_aio_put_stream.py -q -p no:cacheprovider
pytest tests/unit -q -p no:cacheprovider
ruff check .
mypy --config-file=pyproject.toml hatchet_sdk/clients/events.py hatchet_sdk/context/context.py tests/unit/test_aio_put_stream.py
black hatchet_sdk/clients/events.py hatchet_sdk/context/context.py tests/unit/test_aio_put_stream.py --check --diff --workers 1
pydoclint .

🤖 AI Disclosure

• I acknowledge that an LLM was used in the creation of this Pull Request, in accordance with Hatchet's AI_POLICY.md.

• Details: I used Claude Code and Codex to work on this.

[vercel]

@juanluisdb is attempting to deploy a commit to the Hatchet Team on Vercel.

A member of the Team first needs to authorize it.

[mrkaye97]

so there's one very important implementation detail here that I need to think about it a bit, which is that this is not concurrency-safe. We might need to wrap this in a lock to make sure that works properly, I'll think about this a bit 🤔

[juanluisdb]

ok, i see, the same counter is used by both put_stream and aio_put_stream right?

[mrkaye97]

Sorry for just circling back! But yes, it's the same counter

[mrkaye97]

Anyways, I think this is probably okay - I struggle to see a case where you'd put stream events concurrently are care about ordering

[mrkaye97]

maybe we can factor this out so we don't re-create it on every call? small optimization

[juanluisdb]

makes sense. Updated this so aio_stream is not recreated per call, while keeping the aio stub lazy and the retry behavior the same.

[mrkaye97]

Hey @juanluisdb - I sort of commandeered your PR to expand the fixes to a couple more problematic places 😅 I'm going to keep working on it a bit, and hoping to get it merged tomorrow or Weds after some eyes from someone else on the team too. Thanks for the fix here!