[Bug] Helper execute activity doesn't work as expected

[fernando0315]

What are you really trying to do?

I'm having issue with creating helper function to execute activity. My main goal is to have execute activities with the same configuration. Without helper function it works just fine, but after creating a helper function and called it inside workflow, it asks to provide task queue. I would expect the code to behave exactly the same way.

Describe the bug

Code

async def execute_activity_base(
    temporal_activity,
    job_id,
    start_to_close_timeout=timedelta(hours=1),
):
    return await workflow.execute_activity(
        temporal_activity,
        job_id,
        start_to_close_timeout=start_to_close_timeout,
        retry_policy=RetryPolicy(
            initial_interval=timedelta(seconds=30),
            backoff_coefficient=2.0,
            maximum_attempts=3,
            maximum_interval=timedelta(seconds=10),
        ),
    )

Error message

2025-05-06T03:01:36.718693Z WARN temporal_sdk_core::worker::workflow: Error while completing workflow activation error=status: InvalidArgument, message: "invalid TaskQueue on ScheduleActivityTaskCommand: missing task queue name. ActivityId=3 ActivityType=estimated_generation", details: [], metadata: MetadataMap { headers: {"content-type": "application/grpc", "server": "temporal", "date": "Tue, 06 May 2025 03:01:36 GMT"} }

Minimal Reproduction

1. Instead of calling workflow.execute_activity in workflow call the helper method defined above. And the method is define in a different file.

Environment/Versions

• Temporal Version: 1.9.0

[cretz]

I am unable to replicate. For instance, the following code works as expected:

@activity.defn
async def say_hello(name: str) -> str:
    return f"Hello, {name}!"

@workflow.defn
class ActivityHelperWorkflow:
    @workflow.run
    async def run(self, name: str) -> str:
        return await self.my_execute_activity_helper(name)

    async def my_execute_activity_helper(self, name: str) -> str:
        return await workflow.execute_activity(
            say_hello,
            name,
            start_to_close_timeout=timedelta(seconds=30)
        )

Can you provide a full standalone replication with workflow and everything? Can just alter https://github.com/temporalio/samples-python/blob/main/hello/hello_activity.py to replicate your error if that helps.

[fernando0315]

Here you go:

from .helper import execute_activity_base

@activity.defn
async def say_hello(name: str) -> str:
    return f"Hello, {name}!"

@workflow.defn
class ActivityHelperWorkflow:
    @workflow.run
    async def run(self, name: str) -> str:
        return await execute_activity_base(say_hello, name)

# THIS IS helper.py
async def execute_activity_base(
    temporal_activity,
    name,
    start_to_close_timeout=timedelta(hours=1),
):
    return await workflow.execute_activity(
        say_hello,
        name,
        start_to_close_timeout=start_to_close_timeout,
        retry_policy=RetryPolicy(
            initial_interval=timedelta(seconds=30),
            backoff_coefficient=2.0,
            maximum_attempts=3,
            maximum_interval=timedelta(seconds=10),
        ),
    )

[cretz]

The code you pasted has two bugs. First, in helper.py, you have to replace say_hello with temporal_activity. Second, if you run this you get "Maximum interval cannot be less than initial interval" because the retry policy is invalid (can set to 30 like initial interval).

I assume those two bugs aren't related to your problem though. When fixing those two bugs, I am unable to replicate anything concerning task queue, the code works just fine. If it still happens for you, we can open a draft PR on the samples repo with this as a test case to try to get it to fail CI tests to confirm it is not something unique to your environment.

[fernando0315]

This is so weird, when I set initial_interval=timedelta(seconds=1), it works perfectly. But if I change it to initial_interval=timedelta(seconds=30), it throws an “invalid TaskQueue” error. There must be a bug somewhere internally.

[cretz]

I am still not able to replicate even with that retry policy. Can you provide a full, standalone replication? If easiest, can open a draft PR to the samples-python repo, e.g. changing https://github.com/temporalio/samples-python/blob/main/hello/hello_activity.py to replicate what you are seeing.