fix: Keep requested keys alive in the naive scheduler

[SimonHeybrock]

NaiveScheduler frees an intermediate result as soon as every provider consuming it has run. The consumer count is built from the graph's providers only, so a key that is both requested and consumed by another task reaches zero and is deleted before it can be returned:

A = NewType('A', int)
B = NewType('B', int)

def make_b(a: A) -> B:
    return B(a + 1)

pl = sciline.Pipeline((make_b,))
pl[A] = 1

pl.compute((A, B))                                    # dask:  {A: 1, B: 2}
pl.get((A, B), scheduler=NaiveScheduler()).compute()  # naive: KeyError(A)

Any compute for several targets where one target depends on another is affected; the dask scheduler is not. The regression came in with a5c9a08 ("Discard data early in naive scheduler") and is in 26.8.0, so 25.11.1 and earlier are fine.

Counting the requested keys as consumers keeps them alive until they are returned, and everything else is still discarded as early as before.

This was found from downstream fallout rather than from reading the code: ESSlivedata defaults its services to the naive scheduler, and there both detector-view construction and monitor workflow cycles fail with a KeyError on a live key. ess.reduce.streaming.StreamProcessor hits the pattern routinely, computing several targets at once where one feeds another. With this fix those workflows run again under the naive scheduler.

Test plan: the new test runs on all three scheduler fixtures and fails only on [naive] without the fix.

[jl-wynen]

Suggested change

	requested: Container[Hashable],
	) -> None:
	"""Discard results that no remaining provider needs.
	Requested keys are kept: they are returned to the caller, so their consumer
	count reaching zero does not mean they are no longer needed.
	"""
	for arg in provider.arg_spec.keys():
	counts[arg] -= 1
	if counts[arg] == 0:
	if counts[arg] == 0 and arg not in requested:
	requested: set[Hashable],
	) -> None:
	"""Discard results that no remaining provider needs.
	Requested keys are kept: they are returned to the caller, so their consumer
	count reaching zero does not mean they are no longer needed.
	"""
	for arg in provider.arg_spec.keys() - requested:
	counts[arg] -= 1
	if counts[arg] == 0:

This is an internal function, so it doesn't need to have a general argument. This way, you avoid repeated lookups.

[SimonHeybrock]

Good point on the annotation, changed to set[Hashable].

I kept the loop as it was, though: ArgSpec.keys() is a generator, so keys() - requested raises TypeError -- making it work would need a set(...) construction per call. And the lookup isn't actually repeated: arg not in requested sits behind counts[arg] == 0 and short-circuits, so it runs at most once per provider, for the single argument whose count reaches zero.