Activity cancellation is not idempotent: cancelling an activity while the workflow itself is being cancelled fails the workflow task with TMPRL1100

[MaksBoi]

What happens

When a workflow cancels an activity handle at the same moment the workflow itself receives an
external cancellation, the workflow task fails with:

[TMPRL1100] Nondeterminism error: Invalid transition in state machine

The workflow never closes. It stays RUNNING and retries the failing workflow task forever, so
the failure is silent: no terminal state, no error surfaced to the caller.

Why

_WorkflowInstanceImpl._await_temporal_operation (temporalio/worker/_workflow_instance.py)
emits a fresh cancel command on every CancelledError it catches:

while True:
    try:
        return await _shield_await(fut)
    except asyncio.CancelledError as err:
        if fut.done():
            ...
            raise
        apply_cancel(err, self._add_command())   # request_cancel_activity, same seq
        ...

There is no "already cancelled" guard. An external workflow cancellation cascades into the
activity task, and an explicit handle.cancel() cancels the same task, so two
request_cancel_activity commands are produced for one seq. Core rejects the second
transition.

Reproduction

Attached script starts an activity, signals the workflow to cancel that activity, and cancels
the workflow at the same time. Three runs, all three end RUNNING with two failed workflow tasks.
Reproduced on temporalio 1.32.0 with WorkflowEnvironment.start_local().

What we measured

scenario	outcome
external workflow cancel only	COMPLETED, 0 failed tasks
own handle.cancel() only	COMPLETED, 0 failed tasks
signal arrives, activity left alone	COMPLETED, 0 failed tasks
both cancels at once	RUNNING, 2 failed tasks (6 of 6 runs)

The third row against the fourth isolates the cause: the second cancel command breaks it, not
the signal handler.

What does not help

• checking workflow.cancellation_reason() before cancelling: the two cancels are symmetric,
  either side can be first, and the SDK emits the second command regardless
• wrapping the activity handle in asyncio.shield: 6 of 6 runs still fail
• cancellation_type=WAIT_CANCELLATION_COMPLETED: 3 of 3 fail
• cancellation_type=ABANDON is clean, but it leaves the activity running, which defeats the
  purpose of cancelling it

Workaround we use

Delay the in-workflow handle.cancel() by a timer so the two commands land in different
workflow tasks. Measured 6 of 6 clean with the delay against 6 of 6 broken without it. It works,
but it depends on timing rather than on a contract, which is why we are reporting the underlying
issue.

Suggested fix

Track per-handle whether a cancel command was already emitted and skip emitting a second one for
the same seq.

Repro script

"""Repro: in-workflow activity cancel racing an external workflow cancel."""
import asyncio
from datetime import timedelta
from temporalio import activity, workflow
from temporalio.testing import WorkflowEnvironment
from temporalio.worker import Worker


@activity.defn(name="stuck")
async def stuck() -> str:
    while True:
        activity.heartbeat("x")
        await asyncio.sleep(0.05)


@workflow.defn(name="Mini")
class Mini:
    def __init__(self) -> None:
        self._cancel = False
        self._running = False

    @workflow.signal
    async def kill(self) -> None:
        self._cancel = True

    @workflow.query
    def phase(self) -> str:
        return "running" if self._running else "queued"

    @workflow.run
    async def run(self) -> str:
        h = workflow.start_activity(
            "stuck",
            start_to_close_timeout=timedelta(minutes=5),
            heartbeat_timeout=timedelta(seconds=30),
        )
        self._running = True

        async def watch() -> None:
            await workflow.wait_condition(lambda: self._cancel)
            h.cancel()

        w = asyncio.create_task(watch())
        try:
            return await h
        except Exception:
            return "activity-gone"
        finally:
            w.cancel()


async def main() -> None:
    async with await WorkflowEnvironment.start_local() as env:
        for i in range(3):
            worker = Worker(
                env.client, task_queue="mq", workflows=[Mini], activities=[stuck],
                default_heartbeat_throttle_interval=timedelta(milliseconds=50),
                max_heartbeat_throttle_interval=timedelta(milliseconds=50),
            )
            async with worker:
                h = await env.client.start_workflow(Mini.run, id=f"mini-{i}", task_queue="mq")
                for _ in range(200):
                    if (await h.query(Mini.phase)) == "running":
                        break
                    await asyncio.sleep(0.05)
                await asyncio.sleep(0.3)
                await asyncio.gather(h.signal(Mini.kill), h.cancel())
                await asyncio.sleep(8)
                st = (await h.describe()).status
                nd = 0
                async for e in h.fetch_history_events():
                    if e.WhichOneof("attributes") == "workflow_task_failed_event_attributes":
                        nd += 1
                print(f"mini-{i}: {st.name}, wft_failed={nd}", flush=True)


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