Allow `async` functions to run through the Hera Runner

[elliotgunton]

Pre-bug-report checklist

1. This bug can be reproduced using pure Argo YAML

• Yes 👉 Please report a bug to the Argo Workflows GitHub 👈
• No

If yes, it is more likely to be an Argo bug unrelated to Hera. Please double check before submitting an issue to Hera.

2. I have searched for existing issues

• Yes

3. This bug occurs in Hera when...

• exporting to YAML
• submitting to Argo
• running on Argo with the Hera runner
• other:

Bug report

Describe the bug

Write a script template function using async, it will error as:

Traceback (most recent call last):
  File "<frozen runpy>", line 198, in _run_module_as_main
  File "<frozen runpy>", line 88, in _run_code
  File "/usr/local/lib/python3.13/site-packages/hera/workflows/runner.py", line 6, in <module>
    _run()
    ~~~~^^
  File "/usr/local/lib/python3.13/site-packages/hera/workflows/_runner/util.py", line 295, in _run
    print(serialize(result))
          ~~~~~~~~~^^^^^^^^
  File "/usr/local/lib/python3.13/site-packages/hera/shared/serialization.py", line 56, in serialize
    return json.dumps(value, cls=PydanticEncoder)  # None serialized as `null`
           ~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.13/json/__init__.py", line 238, in dumps
    **kw).encode(obj)
          ~~~~~~^^^^^
  File "/usr/local/lib/python3.13/json/encoder.py", line 200, in encode
    chunks = self.iterencode(o, _one_shot=True)
  File "/usr/local/lib/python3.13/json/encoder.py", line 261, in iterencode
    return _iterencode(o, 0)
  File "/usr/local/lib/python3.13/site-packages/hera/shared/serialization.py", line 42, in default
    return super().default(o)
           ~~~~~~~~~~~~~~~^^^
  File "/usr/local/lib/python3.13/json/encoder.py", line 180, in default
    raise TypeError(f'Object of type {o.__class__.__name__} '
                    f'is not JSON serializable')
TypeError: Object of type coroutine is not JSON serializable
<sys>:0: RuntimeWarning: coroutine 'hello_async' was never awaited

To Reproduce

from hera.workflows import Workflow, Steps, script
from hera.shared import global_config

global_config.image = "hera-scratch:v1"

@script(constructor="runner")
async def hello_async(s: str):
    print("Hello, {s}!".format(s=s))

with Workflow(
    generate_name="hello-world-",
    entrypoint="hello-async",
    arguments={"s": "world"},
) as w:
    hello_async()

Build/run with:

.PHONY: run
run:
	@poetry export --without-hashes --without-urls > requirements.txt
	@docker build . -t hera-scratch:v1
	poetry run python -m hera_scratch

Expected behavior
A clear and concise description of what you expected to happen:

The function is awaited by the Hera Runner.

Environment

• Hera Version: 5.20.1
• Python Version: 3.13
• Argo Version: 3.6.7

Additional context
This would be a feature request to allow async script-template functions, which then allows users to use async libraries such as nats-py (https://nats-io.github.io/nats.py/index.html) from a Hera Runner script.

We might also consider a Runner Script Constructor that can take multiple async functions and run them all in parallel (a new issue should be made for this as a full feature request).

[alicederyn]

A few thoughts:

1. Linters may flag up your calling hello_async() without doing an await; you may want to allow users to await inside their with Workflow block.
2. This will need some thought in the world of decorated step/dag functions, where the result will need to be awaited.
3. Do we want to support users writing things like asyncio.gather(step1, step2) to create parallel steps? That would allow the steps to run out-of-order locally as well as in Argo.
4. Typing this may be tricky; adding a new @async_script decorator may be simpler.

[sam2x]

Is there a way to temporary disable "async" attempt on the runner ? i haven't updated for long, and just got the issue trying latest version of the git. I make it works with an asyncio workaround, but it's kinda ugly.