[Bug] workflow.sleep() timer not canceled when wrapping asyncio.Task is canceled

[carlosa54]

What are you really trying to do?

I'm creating concurrent workflow tasks using asyncio.create_task(workflow.sleep(...)) and canceling them based on conditions. I expect that when a task wrapping workflow.sleep() is canceled, the underlying Temporal timer is also canceled — matching the behavior of asyncio.sleep() in the same scenario.

Describe the bug

When an asyncio.Task wrapping workflow.sleep() is canceled, the SDK does not send a CancelTimer command to the Temporal server. The timer persists in workflow history and the callback still fires at the originally scheduled time.

This manifests in two ways:

1. The Temporal timer is never canceled on the server. No TimerCanceled event appears in workflow history. The timer runs to completion as if the task was never canceled.

2. The callback fires on the already-canceled future, causing asyncio.InvalidStateError (related to [Bug] cancelled timer callback causes asyncio.exceptions.InvalidStateError #782, but that issue only mentions the exception not the leaked timer)

Note

asyncio.sleep() properly cancels its timer when the wrapping task is canceled, theTimerCanceled command is sent and the timer is not fired

Expected behavior: workflow.sleep() and asyncio.sleep() should behave the same way when a wrapping task is canceled, both should cancel the underlying Temporal timer.

Minimal Reproduction

Reproduction repo: https://github.com/carlosa54/temporal-workflow-sleep-bug-repro

import asyncio
import uuid
from datetime import timedelta

from temporalio import workflow
from temporalio.client import Client
from temporalio.worker import Worker


@workflow.defn
class SleepCancelBugRepro:
    @workflow.run
    async def run(self) -> str:
        # BUG: workflow.sleep() timer fires even after the task is canceled
        bug_task = asyncio.create_task(
            workflow.sleep(10, summary="workflow.sleep(10) — BUG: timer not canceled")
        )

        # WORKAROUND: asyncio.sleep() properly cancels its timer
        workaround_task = asyncio.create_task(asyncio.sleep(10))

        # Let both timers run for 2s, then cancel
        await workflow.sleep(2, summary="workflow.sleep(2) — wait before canceling")

        bug_task.cancel()
        workaround_task.cancel()

        results = await asyncio.gather(bug_task, workaround_task, return_exceptions=True)
        workflow.logger.info(f"Both tasks canceled: {results}")

        # Wait past the 10s mark — the workflow.sleep timer will spuriously fire at ~10s
        await workflow.sleep(15, summary="workflow.sleep(15) — observing timer behavior")

        return "done"


async def main():
    client = await Client.connect("localhost:7233")

    async with Worker(
        client,
        task_queue="sleep-cancel-bug-repro",
        workflows=[SleepCancelBugRepro],
    ):
        result = await client.execute_workflow(
            SleepCancelBugRepro.run,
            id=f"sleep-cancel-bug-{uuid.uuid4().hex[:8]}",
            task_queue="sleep-cancel-bug-repro",
            execution_timeout=timedelta(seconds=60),
        )
        print(f"Result: {result}")


if __name__ == "__main__":
    asyncio.run(main())

Environment/Versions

• OS and processor: M4 Mac (arm64)
• Temporal Python SDK: 1.21.1
• Temporal CLI dev server: 1.6.1

Additional context