Fix custom task run name rendering

[zzstoatzz]

MRE i was using

from random import uniform
from time import sleep
from typing import Any

from prefect import flow, task
from prefect.runtime import task_run


def generate_task_run_name(parameters: dict[str, Any]) -> str:
    """previously we got parameters here from prefect.runtime.task_run.parameters, but as discussed below the values are not guaranteed to be resolved"""
    return f'{task_run.task_name} - input["val"]:  {parameters["input"]["val"]}'


@task(log_prints=True)
def pre_task() -> bool:
    return True


@task(log_prints=True, task_run_name=generate_task_run_name)
def task_1(input: dict[str, Any]) -> dict[str, Any]:
    input_val = input["val"]
    input["val"] = input_val + 1
    print(f'task_1 - input["val"]:  {input_val}, output:  {input}')

    sleep(uniform(1, 5))
    return input


@task(log_prints=True, task_run_name=generate_task_run_name)
def task_2(input: dict[str, Any]) -> dict[str, Any]:
    input_val = input["val"]
    input["val"] = input_val * 10
    print(f'task_2 - input["val"]:  {input_val}, output:  {input}')

    sleep(uniform(1, 5))
    return input


@task(log_prints=True, task_run_name=generate_task_run_name)
def task_3(input: dict[str, Any]) -> None:
    input_val = input["val"]
    print(f'task_3 - input["val"]: {input_val}')

    sleep(uniform(1, 5))
    return


@flow
def my_flow() -> None:
    pre_task()
    inputs: list[dict[str, Any]] = [
        {"val": 1, "something_else": True},
        {"val": 2, "something_else": True},
        {"val": 3, "something_else": True},
        {"val": 4, "something_else": True},
        {"val": 5, "something_else": True},
        {"val": 6, "something_else": True},
        {"val": 7, "something_else": True},
        {"val": 8, "something_else": True},
    ]

    result_1 = task_1.map(input=inputs)
    result_2 = task_2.map(input=result_1)
    final_result = task_3.map(input=result_2)

    final_result.wait()


if __name__ == "__main__":
    my_flow()

closes #15747

the issue (which #15770 does not address) is that the TaskRunContext may receive a non-resolved dict of parameters when it is created, and the TaskRunContext is treated as immutable. this itself feels like an incongruence to address - but this PR is mostly concerned with the custom task run name

with CSTRO I'm not sure that we can move the future resolution before the creation / entrance of the TaskRunContext (happy to talk this through sync) while also correctly handling UpstreamTaskError exceptions, so in addition to fixing the templating syntax like task_run_name="using {input[val]}" by deferring the attempt to render the custom name, this PR proposes a new DX when providing a callable task_run_name that needs access to the rendered parameters

def generate_task_run_name(parameters: dict) -> str:
    return f'{task_run.task_name} - input: {parameters["input"]["number"]}'

if the user provides a function that accepts a parameters dict, then we will populate it with the dict of parameters, which at that point is guaranteed to be resolved

if we are okay with this slight pivot, then I think we should deprecate / remove the get_parameters from runtime.task_run, since it does not currently work for cases where futures need to be resolved

[codspeed-hq]

CodSpeed Performance Report

Merging #15773 will not alter performance

_{Comparing fix-custom-name-rendering (32348bb) with main (d09ef03)}

Summary

✅ 3 untouched benchmarks

[desertaxle]

Makes sense to me!

[zzstoatzz]

we pass the parameters if the author of the callable requests parameters