Fix superseded async references escaping cancellation

[philippjfr]

Problem

When an asynchronous reference is re-resolved because one of its dependencies changed, param
is supposed to cancel the resolution it supersedes so only the newest one writes a value. In
practice only every other superseded resolution was cancelled, so stale coroutines ran to
completion and stale async generators kept streaming into the parameter.

class Source(param.Parameterized):
    x = param.Number(default=0)

class P(param.Parameterized):
    value = param.Parameter(allow_refs=True)

async def slow(i):
    await asyncio.sleep(0.1)
    return i

source = Source()
p = P(value=param.bind(slow, source.param.x))

for i in (1, 2, 3):
    source.x = i
    await asyncio.sleep(0.01)

Four resolutions are scheduled (for x = 0, 1, 2, 3) and three of them are obsolete before
they finish. Only 0 and 2 were cancelled; 1 and 3 both completed and both assigned to
p.value. Which value survived came down to which task happened to finish last, so a slow
early resolution could clobber the result of a fast later one.

The same thing with an async generator is worse, because a superseded generator is not a
one-off stale write but an unbounded second writer: two generators interleave their emissions
into the same parameter for as long as they both keep yielding, and nothing ever stops the
orphan.

Root cause

Parameters._async_ref tracks the resolution that owns a reference in
_param__private.async_refs[pname], and each new resolution cancels whatever it finds there:

running_task = self_.self._param__private.async_refs.get(pname)
if running_task is None:
    self_.self._param__private.async_refs[pname] = current_task
elif current_task is not running_task:
    self_.self._param__private.async_refs[pname].cancel()

The superseding task cancels the previous one but never takes ownership of the reference, and
the finally block of the task it just cancelled removes the entry it still owns. The
registry therefore ends up empty rather than pointing at the live task:

task	sees in registry	action	registry afterwards
0	empty	registers itself	task 0
1	task 0	cancels task 0, does not register	empty (task 0's finally clears it)
2	empty	registers itself	task 2
3	task 2	cancels task 2, does not register	empty

Tasks 1 and 3 are never anybody's running_task, so nothing cancels them. The registry
alternates between "owned" and "empty" and half the resolutions slip through.

This did not show up in the existing coverage because test_async_generator_ref_cancelled
and test_generator_ref_cancelled supersede a ref by reassigning it, and reassignment is
handled elsewhere: Parameter.__set__ pops and cancels the running task synchronously when a
ref is replaced, so the registry is already empty when the new task starts and the new task
registers itself. Only the dependency-change path goes through the broken branch, and only
from the second supersession onwards.

Changes

Parameters._async_ref now takes ownership before cancelling, so the live resolution is
always the registered one:

running_task = self_.self._param__private.async_refs.get(pname)
if running_task is not current_task:
    if running_task is not None:
        running_task.cancel()
    self_.self._param__private.async_refs[pname] = current_task

The finally cleanup already only clears the entry when it still points at the current task,
which is now exactly right: a cancelled task finds its successor registered and leaves the
entry alone, and only the last surviving resolution clears it. It is also no longer possible
for the registry to hold a task that has already finished, which matters for the two other
places that cancel a reference (Parameter.__set__ and Parameters._update_ref) since both
cancel whatever the registry holds.

That cleanup is additionally guarded with a pname in async_refs check. Both the old and new
ownership code leave the reference unregistered when _async_ref runs outside a task
(asyncio.current_task() returns None), and in that case the old get(pname) is current_task test compared None to None, passed, and raised KeyError on the del.

Tests

Two tests in tests/testrefs.py, both failing before this change:

• test_async_ref_cancelled_on_dependency_change asserts that of four scheduled
  resolutions only the last completes. Before: assert [1, 3] == [3].
• test_async_generator_ref_cancelled_on_dependency_change asserts that after two
  supersessions only the newest generator still emits. Before: assert {1, 2} == {2}.

The existing reassignment-based cancellation tests still pass, as does the rest of the suite
(1551 passed, 4 skipped, 2 xfailed).

AI Disclosure

Fixed with the aid of Claude Opus 5.

[codecov]

Codecov Report

❌ Patch coverage is 80.00000% with 7 lines in your changes missing coverage. Please review.
✅ Project coverage is 86.74%. Comparing base (b2f6f85) to head (d649152).
⚠️ Report is 1 commits behind head on main.

Files with missing lines	Patch %	Lines
param/parameterized.py	80.00%	7 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #1175      +/-   ##
==========================================
- Coverage   86.75%   86.74%   -0.02%     
==========================================
  Files           9        9              
  Lines        5336     5362      +26     
==========================================
+ Hits         4629     4651      +22     
- Misses        707      711       +4     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.