Fix TASK_SOURCE cache policy for remote execution with cloudpickle

[desertaxle]

Summary

• Fixes an issue where the TASK_SOURCE cache policy fails when used with remote execution decorators like @ecs that use cloudpickle to ship code to remote environments
• inspect.getsource() fails on cloudpickled functions because the original source file doesn't exist on the remote machine
• The previous fallback to hashing __code__.co_code (bytecode) led to inconsistent cache keys because bytecode varies across operating systems and architectures
• Stores the function's source code on the Task object during initialization so it survives cloudpickle serialization

Changes

src/prefect/tasks.py

• Added source_code attribute to Task.__init__() that captures inspect.getsource(fn) during task initialization
• Gracefully handles TypeError and OSError when source code is unavailable (e.g., callable objects)

src/prefect/cache_policies.py

• Updated TaskSource.compute_key() to check for stored source_code attribute first
• Falls back to inspect.getsource() for backward compatibility when source_code is not available

Tests

• Added TestTaskSourceCode class in tests/test_tasks.py with tests for source code capture and cloudpickle survival
• Updated tests/test_cache_policies.py with tests for the new stored source code behavior

🤖 Generated with Claude Code

[codspeed-hq]

CodSpeed Performance Report

Merging #19926 will degrade performance by 22.92%

_{Comparing fix/task-source-cache-policy-cloudpickle (84cd604) with main (7d73faa)}

Summary

❌ 1 (👁 1) regression
✅ 1 untouched

Benchmarks breakdown

	Benchmark	BASE	HEAD	Efficiency
👁	bench_task_decorator	463.3 µs	601.1 µs	-22.92%

[zzstoatzz]

do you think the codspeed report is just flapping or a real regression?

[desertaxle]

do you think the codspeed report is just flapping or a real regression?

I think it's a real regression casued by the new inspect.getsource call, but it's a difference of <150 microseconds, so I think it's acceptable.