Fix cancelling a local activity during its retry backoff

[RachelWanggg]

Fixes #1971.

When Core resolves a local activity with DoBackoff, _resolve_backoff moves the handle to a new seq and sets _started = False. The ScheduleLocalActivity command for that seq is only sent after the backoff workflow_sleep in run_activity. If the handle is cancelled during that sleep (a client-side workflow cancel, handle.cancel(), or asyncio.wait_for around execute_local_activity), cancel() sees _started as false and calls _request_cancel(). That emits RequestCancelLocalActivity for a seq Core never saw, and every workflow task after that fails with [TMPRL1100] Nondeterminism error: Missing associated machine for LocalActivity(N). The workflow stays RUNNING and the backoff timer is never cancelled.

_started only exists so cancel() can send a command for a handle whose coroutine has not run yet and so cannot trap the cancel. During the backoff sleep the coroutine is running, so this change stops resetting it in _resolve_backoff, and cancel() falls through to a plain task cancel. That cancels the backoff timer the same way as any other workflow.sleep. run_activity now sets _started once instead of on every loop iteration.

After the fix, a cancel during timer backoff raises asyncio.CancelledError from the handle (a workflow cancel ends the workflow cancelled, wait_for raises TimeoutError), while a cancel during a running attempt still raises ActivityError with a CancelledError cause. That is the same split as the TypeScript and .NET SDKs, which sleep on a timer during backoff and only send a local activity cancel for a seq that is still pending. Cancels outside the backoff window produce the same commands as before. On main this path always fails the workflow task, so no completed history contains it.

test_workflow_local_activity_cancel_during_backoff raises ApplicationError with next_retry_delay above the default 60s local_retry_threshold, so Core switches to timer backoff after the first attempt with only start_to_close_timeout set. It waits for TIMER_STARTED, then cancels in two ways:

• the client cancels the workflow, which must fail with a CancelledError cause
• a signal makes the workflow call handle.cancel(), and await handle must raise asyncio.CancelledError

Both assert TIMER_CANCELED and no WORKFLOW_TASK_FAILED in history. Both fail on main at the 60s pytest timeout, with the nondeterminism error above logged on every workflow task retry, and pass with the fix. Putting self._started = False back in _resolve_backoff, or dropping handle._started = True from run_activity, fails both again.

Validation, run locally on macOS (Apple Silicon) with Python 3.12, Rust bridge built with poe build-develop:

• poe lint: ruff import and format checks clean, pyright 0 errors, mypy no issues in 272 source files, basedpyright 0 errors, pydocstyle clean
• full suite, pytest -n auto --dist=worksteal -v -o timeout_func_only=false --timeout 180: 1488 passed, 9 skipped
• pytest tests/worker/test_workflow.py tests/worker/test_replayer.py -n auto --dist=worksteal: 258 passed, 1 skipped
• pytest tests/worker/test_workflow.py -k "local_activity or cancel_activity or backoff" --workflow-environment time-skipping: 13 passed
• pytest tests/worker/test_workflow.py -k test_workflow_local_activity_cancel_during_backoff: 2 passed, run twice (2 failed on main at the 60s timeout)
• the repro script from the issue: main prints status: RUNNING with the TMPRL1100 task failures, this branch prints status: CANCELED and timer cancelled

Not run: Python versions other than 3.12, Linux and Windows.

[CLAassistant]

All committers have signed the CLA.

[CLAassistant]

Thank you for your submission! We really appreciate it. Like many open source projects, we ask that you sign our Contributor License Agreement before we can accept your contribution.
_{You have signed the CLA already but the status is still pending? Let us recheck it.}

[Sushisource]

Thanks for the fix!