Short answer: Yes — in Prefect 3.x you can lose upstream task_inputs when a plain helper function materializes or reshapes a task’s result before passing it to another task. Prefect links dependencies by object identity recorded when a task returns; if a helper creates a new object (copy/slice/index/deeper extract), that identity isn’t in the cache and the downstream task’s task_inputs won’t show the upstream edge. Minimal reproduction

python
from prefect import task, flow
from prefect.context import TaskRunContext

@task
def upstream():
    # Note: the list is nested 2 levels deep
    return {"outer": {"inner": [1, 2, 3]}}

@task
def consumer(values):
    # See which upstreams Prefect recorded for this task run
    ti = TaskRunContext.get().task_run.task_inputs
    print(f"task_inputs: {ti}")  # → will be {} in the broken case
    return len(values)

# Ordinary helper that MATERIALIZES and extracts a deeper-nested value
def helper_broken():
    data = upstream().result()          # materialize Python object
    return data["outer"]["inner"]       # new object id, not linked

@flow
def demo_broken():
    # Loses upstream linkage: consumer gets a plain list with no recorded source
    consumer(helper_broken())

if __name__ == "__main__":
    demo_broken()

Why it breaks - Prefect records object identity for the task’s return (and only 1 level of nested children). - The helper selects a value 2 levels deep, producing an object whose id was never linked to the upstream run. - collect_task_run_inputs cannot map that object back to the upstream task, so task_inputs is empty. Minimal fixes - Keep futures intact (don’t materialize/reshape in helpers)

python
def helper_min_fix():
    # Just orchestrate: pass the PrefectFuture through unchanged
    return consumer(upstream())

@flow
def demo_fixed_1():
    helper_min_fix()  # consumer’s task_inputs correctly include upstream

- Or, move any data-shaping into its own task so lineage is explicit

python
@task
def pick_inner(d):
    return d["outer"]["inner"]

@flow
def demo_fixed_2():
    consumer(pick_inner(upstream()))  # upstream -> pick_inner -> consumer

Notes - This is a metadata/lineage issue only; tasks still execute. - The same loss can happen with copies/slices (e.g., xs = upstream().result(); ys = xs[:]) or constructing new containers (dict/list/tuple/dataclass) from task results. - Best practice: orchestrate with helpers (pass PrefectFutures unchanged) and do data manipulation inside tasks.